<commit_before>// Copyright 2018 Google Inc. All Rights Reserved.
// This file is available under the Apache license.

// Package file provides an abstraction over files and named pipes being tailed
// by `mtail`.
package file

import (
	"bytes"
	"expvar"
	"io"
	"os"
	"path/filepath"
	"time"
	"unicode/utf8"

	"github.com/golang/glog"
	"github.com/google/mtail/logline"
	"github.com/pkg/errors"
	"github.com/spf13/afero"
)

var (
	// logErrors counts the number of IO errors per log file
	logErrors = expvar.NewMap("log_errors_total")
	// lineCount counts the numbre of lines read per log file
	lineCount = expvar.NewMap("log_lines_total")
	// logTruncs counts the number of log truncation events per log
	logTruncs = expvar.NewMap("log_truncates_total")
)

// File contains the state for a tailed file.
type File struct {
	Name     string // Given name for the file (possibly relative, used for displau)
	Pathname string // Full absolute path of the file used internally
	file     afero.File
	partial  *bytes.Buffer
	lines    chan<- *logline.LogLine // output channel for lines read
}

// New returns a new File named by the given pathname.  seenBefore indicates
// thta mtail believes it's seen this pathname nbefore, and seekToStart
// indicates that the file should be tailed from offset 0, not EOF.
func New(fs afero.Fs, pathname string, lines chan<- *logline.LogLine, seenBefore, seekToStart bool) (*File, error) {
	glog.V(2).Infof("file.New(%s, %v, %v)", pathname, seenBefore, seekToStart)
	absPath, err := filepath.Abs(pathname)
	if err != nil {
		return nil, err
	}
	retries := 3
	retryDelay := 1 * time.Millisecond
	shouldRetry := func() bool {
		// seenBefore indicates also that we're rotating a file that previously worked, so retry.
		if !seenBefore {
			return false
		}
		return retries > 0
	}
	var f afero.File
Retry:
	f, err = fs.Open(absPath)
	if err != nil {
		logErrors.Add(absPath, 1)
		if shouldRetry() {
			retries = retries - 1
			time.Sleep(retryDelay)
			retryDelay = retryDelay + retryDelay
			goto Retry
		}
	}
	if err != nil {
		glog.Infof("openLogPath failed all retries")
		return nil, err
	}
	glog.V(2).Infof("open succeeded %s", absPath)
	fi, err := f.Stat()
	if err != nil {
		// Stat failed, log error and return.
		logErrors.Add(absPath, 1)
		return nil, errors.Wrapf(err, "Failed to stat %q: %s", absPath)
	}
	switch m := fi.Mode(); {
	case m.IsRegular():
		seekWhence := io.SeekEnd
		if seekToStart {
			seekWhence = io.SeekCurrent
		}
		if _, err := f.Seek(0, seekWhence); err != nil {
			return nil, errors.Wrapf(err, "Seek failed on %q", absPath)
		}
		// Named pipes are the same as far as we're concerned, but we can't seek them.
		fallthrough
	case m&os.ModeType == os.ModeNamedPipe:
	default:
		return nil, errors.Errorf("Can't open files with mode %v: %s", m&os.ModeType, absPath)
	}
	return &File{pathname, absPath, f, bytes.NewBufferString(""), lines}, nil
}

// Read blocks of 4096 bytes from the File, sending LogLines to the given
// channel as newlines are encountered.  If EOF is read, the partial line is
// stored to be concatenated to on the next call.  At EOF, checks for
// truncation and resets the file offset if so.
func (f *File) Read() error {
	b := make([]byte, 0, 4096)
	totalBytes := 0
	for {
		n, err := f.file.Read(b[:cap(b)])
		glog.V(2).Infof("Read count %v err %v", n, err)
		totalBytes += n
		b = b[:n]

		if err == io.EOF && totalBytes == 0 {
			glog.V(2).Info("Suspected truncation.")
			// If there was nothing to be read, perhaps the file just got truncated.
			truncated, terr := f.checkForTruncate()
			if terr != nil {
				glog.Infof("checkForTruncate returned with error '%v'", terr)
			}
			if truncated {
				// Try again: offset was greater than filesize and now we've seeked to start.
				continue
			}
		}

		var (
			rune  rune
			width int
		)
		for i := 0; i < len(b) && i < n; i += width {
			rune, width = utf8.DecodeRune(b[i:])
			switch {
			case rune != '\n':
				f.partial.WriteRune(rune)
			default:
				f.sendLine()
			}
		}

		// Return on any error, including EOF.
		if err != nil {
			return err
		}
	}
}

// sendLine sends the contents of the partial buffer off for processing.
func (f *File) sendLine() {
	f.lines <- logline.NewLogLine(f.Name, f.partial.String())
	lineCount.Add(f.Name, 1)
	// reset partial accumulator
	f.partial.Reset()
}

// checkForTruncate checks to see if the current offset into the file
// is past the end of the file based on its size, and if so seeks to
// the start again.
func (f *File) checkForTruncate() (bool, error) {
	currentOffset, err := f.file.Seek(0, io.SeekCurrent)
	glog.V(2).Infof("current seek position at %d", currentOffset)
	if err != nil {
		return false, err
	}

	fi, err := f.file.Stat()
	if err != nil {
		return false, err
	}

	glog.V(2).Infof("File size is %d", fi.Size())
	if currentOffset == 0 || fi.Size() >= currentOffset {
		glog.V(2).Info("no truncate appears to have occurred")
		return false, nil
	}

	f.partial.Reset()

	p, serr := f.file.Seek(0, io.SeekStart)
	glog.V(2).Infof("Truncated?  Seeked to %d: %v", p, serr)
	logTruncs.Add(f.Name, 1)
	return true, serr
}

func (f *File) Stat() (os.FileInfo, error) {
	return f.file.Stat()
}

func (f *File) Close() error {
	return f.file.Close()
}<commit_msg>Send a partial line out at truncate if there's anything in there.<commit_after>// Copyright 2018 Google Inc. All Rights Reserved.
// This file is available under the Apache license.

// Package file provides an abstraction over files and named pipes being tailed
// by `mtail`.
package file

import (
	"bytes"
	"expvar"
	"io"
	"os"
	"path/filepath"
	"time"
	"unicode/utf8"

	"github.com/golang/glog"
	"github.com/google/mtail/logline"
	"github.com/pkg/errors"
	"github.com/spf13/afero"
)

var (
	// logErrors counts the number of IO errors per log file
	logErrors = expvar.NewMap("log_errors_total")
	// lineCount counts the numbre of lines read per log file
	lineCount = expvar.NewMap("log_lines_total")
	// logTruncs counts the number of log truncation events per log
	logTruncs = expvar.NewMap("log_truncates_total")
)

// File contains the state for a tailed file.
type File struct {
	Name     string // Given name for the file (possibly relative, used for displau)
	Pathname string // Full absolute path of the file used internally
	file     afero.File
	partial  *bytes.Buffer
	lines    chan<- *logline.LogLine // output channel for lines read
}

// New returns a new File named by the given pathname.  seenBefore indicates
// thta mtail believes it's seen this pathname nbefore, and seekToStart
// indicates that the file should be tailed from offset 0, not EOF.
func New(fs afero.Fs, pathname string, lines chan<- *logline.LogLine, seenBefore, seekToStart bool) (*File, error) {
	glog.V(2).Infof("file.New(%s, %v, %v)", pathname, seenBefore, seekToStart)
	absPath, err := filepath.Abs(pathname)
	if err != nil {
		return nil, err
	}
	retries := 3
	retryDelay := 1 * time.Millisecond
	shouldRetry := func() bool {
		// seenBefore indicates also that we're rotating a file that previously worked, so retry.
		if !seenBefore {
			return false
		}
		return retries > 0
	}
	var f afero.File
Retry:
	f, err = fs.Open(absPath)
	if err != nil {
		logErrors.Add(absPath, 1)
		if shouldRetry() {
			retries = retries - 1
			time.Sleep(retryDelay)
			retryDelay = retryDelay + retryDelay
			goto Retry
		}
	}
	if err != nil {
		glog.Infof("openLogPath failed all retries")
		return nil, err
	}
	glog.V(2).Infof("open succeeded %s", absPath)
	fi, err := f.Stat()
	if err != nil {
		// Stat failed, log error and return.
		logErrors.Add(absPath, 1)
		return nil, errors.Wrapf(err, "Failed to stat %q: %s", absPath)
	}
	switch m := fi.Mode(); {
	case m.IsRegular():
		seekWhence := io.SeekEnd
		if seekToStart {
			seekWhence = io.SeekCurrent
		}
		if _, err := f.Seek(0, seekWhence); err != nil {
			return nil, errors.Wrapf(err, "Seek failed on %q", absPath)
		}
		// Named pipes are the same as far as we're concerned, but we can't seek them.
		fallthrough
	case m&os.ModeType == os.ModeNamedPipe:
	default:
		return nil, errors.Errorf("Can't open files with mode %v: %s", m&os.ModeType, absPath)
	}
	return &File{pathname, absPath, f, bytes.NewBufferString(""), lines}, nil
}

// Read blocks of 4096 bytes from the File, sending LogLines to the given
// channel as newlines are encountered.  If EOF is read, the partial line is
// stored to be concatenated to on the next call.  At EOF, checks for
// truncation and resets the file offset if so.
func (f *File) Read() error {
	b := make([]byte, 0, 4096)
	totalBytes := 0
	for {
		n, err := f.file.Read(b[:cap(b)])
		glog.V(2).Infof("Read count %v err %v", n, err)
		totalBytes += n
		b = b[:n]

		if err == io.EOF && totalBytes == 0 {
			glog.V(2).Info("Suspected truncation.")
			// If there was nothing to be read, perhaps the file just got truncated.
			truncated, terr := f.checkForTruncate()
			if terr != nil {
				glog.Infof("checkForTruncate returned with error '%v'", terr)
			}
			if truncated {
				// Try again: offset was greater than filesize and now we've seeked to start.
				continue
			}
		}

		var (
			rune  rune
			width int
		)
		for i := 0; i < len(b) && i < n; i += width {
			rune, width = utf8.DecodeRune(b[i:])
			switch {
			case rune != '\n':
				f.partial.WriteRune(rune)
			default:
				f.sendLine()
			}
		}

		// Return on any error, including EOF.
		if err != nil {
			return err
		}
	}
}

// sendLine sends the contents of the partial buffer off for processing.
func (f *File) sendLine() {
	f.lines <- logline.NewLogLine(f.Name, f.partial.String())
	lineCount.Add(f.Name, 1)
	// reset partial accumulator
	f.partial.Reset()
}

// checkForTruncate checks to see if the current offset into the file
// is past the end of the file based on its size, and if so seeks to
// the start again.
func (f *File) checkForTruncate() (bool, error) {
	currentOffset, err := f.file.Seek(0, io.SeekCurrent)
	glog.V(2).Infof("current seek position at %d", currentOffset)
	if err != nil {
		return false, err
	}

	fi, err := f.file.Stat()
	if err != nil {
		return false, err
	}

	glog.V(2).Infof("File size is %d", fi.Size())
	if currentOffset == 0 || fi.Size() >= currentOffset {
		glog.V(2).Info("no truncate appears to have occurred")
		return false, nil
	}

	// We're about to lose all data because of the truncate so if there's
	// anything in the buffer, send it out.
	if f.partial.Len() > 0 {
		f.sendLine()
	}

	p, serr := f.file.Seek(0, io.SeekStart)
	glog.V(2).Infof("Truncated?  Seeked to %d: %v", p, serr)
	logTruncs.Add(f.Name, 1)
	return true, serr
}

func (f *File) Stat() (os.FileInfo, error) {
	return f.file.Stat()
}

func (f *File) Close() error {
	return f.file.Close()
}
{'fraction_non_alphanumeric': 0.12741348754780338, 'fraction_numerical': 0.007741815129185711, 'mean_word_length': 3.306024096385542, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '71309', 'n_tokens_mistral': 4098, 'n_tokens_neox': 3779, 'n_words': 1296}