Merge pull request #44 from ansakharov/files-watcher

ansakharov · web-flow · commit 290fe0b4c15d · 2022-03-01T15:51:22.000+03:00
[fix-filewatcher]
diff --git a/Makefile b/Makefile
@@ -11,10 +11,16 @@ deps:
 
 .PHONY: cover
 cover:
-	go test -coverprofile=coverage.out ./...
+	go test -short -coverprofile=coverage.out ./...
 	go tool cover -html=coverage.out
 	rm coverage.out
 
+.PHONY: test-short
+test-short:
+	go test ./fd/ -v -count 1 -short
+	go test ./pipeline/ -v -count 1 -short
+	go test ./plugin/... -v -count 1 -short
+
 .PHONY: test
 test:
 	go test ./fd/ -v -count 1
diff --git a/fd/file.d.go b/fd/file.d.go
@@ -277,7 +277,3 @@ func (f *FileD) serveFreeOsMem(_ http.ResponseWriter, _ *http.Request) {
 func (f *FileD) serveLiveReady(_ http.ResponseWriter, _ *http.Request) {
 	logger.Infof("live/ready OK")
 }
-
-func (f *FileD) servePipelines(_ http.ResponseWriter, _ *http.Request) {
-	logger.Infof("pipelines OK")
-}
diff --git a/pipeline/batch_test.go b/pipeline/batch_test.go
@@ -22,8 +22,7 @@ func (b *batcherTail) Error(err string) {
 	logger.Panic(err)
 }
 
-func (b *batcherTail) WaitOrPanic(string) {}
-func (b *batcherTail) RecoverFromPanic()  {}
+func (b *batcherTail) RecoverFromPanic() {}
 
 func TestBatcher(t *testing.T) {
 	eventCount := 10000000
diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go
@@ -264,7 +264,7 @@ func (p *Pipeline) GetOutput() OutputPlugin {
 func (p *Pipeline) In(sourceID SourceID, sourceName string, offset int64, bytes []byte, isNewSource bool) uint64 {
 	length := len(bytes)
 
-	// don't process shit
+	// don't process shit.
 	isEmpty := length == 0 || (bytes[0] == '\n' && length == 1)
 	isSpam := p.antispamer.isSpam(sourceID, sourceName, isNewSource)
 	isLong := p.settings.MaxEventSize != 0 && length > p.settings.MaxEventSize
@@ -273,7 +273,7 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offset int64, bytes
 	}
 
 	event := p.eventPool.get()
-	dec := decoder.NO
+	var dec decoder.DecoderType
 	if p.decoder == decoder.AUTO {
 		dec = p.suggestedDecoder
 	} else {
diff --git a/pipeline/processor.go b/pipeline/processor.go
@@ -61,7 +61,6 @@ type processor struct {
 	busyActions      []bool
 	busyActionsTotal int
 	actionWatcher    *actionWatcher
-	waitOrPanic      func(msgStr string)
 	recoverFromPanic func()
 
 	heartbeatCh   chan *stream
diff --git a/plugin/action/debug/debug.go b/plugin/action/debug/debug.go
@@ -9,11 +9,7 @@ import (
 /*{ introduction
 It logs event to stdout. Useful for debugging.
 }*/
-type Plugin struct {
-	paths          [][]string
-	names          []string
-	preserveFields bool
-}
+type Plugin struct{}
 
 type Config map[string]interface{}
 
diff --git a/plugin/action/parse_es/parse_es.go b/plugin/action/parse_es/parse_es.go
@@ -12,7 +12,6 @@ It parses HTTP input using Elasticsearch `/_bulk` API format. It converts source
 }*/
 type Plugin struct {
 	logger      *zap.SugaredLogger
-	config      *Config
 	passNext    bool
 	discardNext bool
 	isStrict    bool
diff --git a/plugin/input/file/offset_test.go b/plugin/input/file/offset_test.go
@@ -1,6 +1,7 @@
 package file
 
 import (
+	"go.uber.org/atomic"
 	"os"
 	"sync"
 	"testing"
@@ -77,7 +78,7 @@ func TestParallel(t *testing.T) {
 		lastEventSeq:   0,
 		isVirgin:       false,
 		isDone:         false,
-		shouldSkip:     false,
+		shouldSkip:     *atomic.NewBool(false),
 		offsets:        offsets,
 		mu:             &sync.Mutex{},
 	}
@@ -91,7 +92,7 @@ func TestParallel(t *testing.T) {
 		lastEventSeq:   0,
 		isVirgin:       false,
 		isDone:         false,
-		shouldSkip:     false,
+		shouldSkip:     *atomic.NewBool(false),
 		offsets:        offsets,
 		mu:             &sync.Mutex{},
 	}
diff --git a/plugin/input/file/provider.go b/plugin/input/file/provider.go
@@ -66,7 +66,7 @@ type Job struct {
 
 	isVirgin   bool // it should be set to false if job hits isDone=true at the first time
 	isDone     bool
-	shouldSkip bool
+	shouldSkip atomic.Bool
 
 	// offsets is a sliceMap of streamName to offset.
 	// Unlike map[string]int, sliceMap can work with mutable strings when using unsafe conversion from []byte.
@@ -305,7 +305,7 @@ func (jp *jobProvider) addJob(file *os.File, stat os.FileInfo, filename string,
 
 		isVirgin:   true,
 		isDone:     true,
-		shouldSkip: false,
+		shouldSkip: *atomic.NewBool(false),
 
 		offsets: nil,
 
@@ -368,7 +368,7 @@ func (jp *jobProvider) initJobOffset(operation offsetsOp, job *Job) {
 		if err != nil {
 			jp.logger.Panicf("can't make job, can't seek file %d:%s: %s", job.sourceID, job.filename, err.Error())
 		}
-		job.shouldSkip = true
+		job.shouldSkip.Store(true)
 
 	case offsetsOpReset:
 		_, err := job.file.Seek(0, io.SeekStart)
diff --git a/plugin/input/file/watcher.go b/plugin/input/file/watcher.go
@@ -56,12 +56,11 @@ func (w *watcher) start() {
 	eventsCh := make(chan notify.EventInfo, 128)
 	w.watcherCh = eventsCh
 
-	events := []notify.Event{notify.Create}
+	events := []notify.Event{notify.Create, notify.Rename, notify.Remove}
 	if w.shouldWatchWrites {
 		events = append(events, notify.Write)
 	}
-
-	// watch recursivly
+	// watch recursively.
 	err := notify.Watch(filepath.Join(w.path, "..."), eventsCh, events...)
 	if err != nil {
 		w.logger.Warnf("can't create fs watcher: %s", err.Error())
diff --git a/plugin/input/file/worker.go b/plugin/input/file/worker.go
@@ -3,6 +3,7 @@ package file
 import (
 	"bytes"
 	"io"
+	"os"
 
 	"github.com/ozontech/file.d/longpanic"
 	"github.com/ozontech/file.d/pipeline"
@@ -14,6 +15,7 @@ type worker struct {
 }
 
 type inputer interface {
+	// In processes event and returns it seq number.
 	In(sourceID pipeline.SourceID, sourceName string, offset int64, data []byte, isNewSource bool) uint64
 }
 
@@ -22,24 +24,23 @@ func (w *worker) start(inputController inputer, jobProvider *jobProvider, readBu
 }
 
 func (w *worker) work(controller inputer, jobProvider *jobProvider, readBufferSize int, logger *zap.SugaredLogger) {
+	var inBuffer []byte
 	accumBuffer := make([]byte, 0, readBufferSize)
 	readBuffer := make([]byte, readBufferSize)
-	var inBuffer []byte
 	shouldCheckMax := w.maxEventSize != 0
 
 	for {
 		job := <-jobProvider.jobsChan
 		if job == nil {
 			return
 		}
-
 		job.mu.Lock()
 		file := job.file
 		isDone := job.isDone
 		isVirgin := job.isVirgin
 		sourceID := job.sourceID
 		sourceName := job.filename
-		skipLine := job.shouldSkip
+		skipLine := job.shouldSkip.Load()
 		if job.symlink != "" {
 			sourceName = job.symlink
 		}
@@ -54,7 +55,10 @@ func (w *worker) work(controller inputer, jobProvider *jobProvider, readBufferSi
 			logger.Fatalf("can't get offset, file %d:%s seek error: %s", sourceID, sourceName, err.Error())
 		}
 
+		// have end of file reached.
 		isEOF := false
+
+		// have any event put.
 		wasPut := false
 
 		readTotal := int64(0)
@@ -63,18 +67,18 @@ func (w *worker) work(controller inputer, jobProvider *jobProvider, readBufferSi
 
 		accumBuffer = accumBuffer[:0]
 		for {
+			// on each iteration cap is readBufferSize, so cut all extra bytes.
 			readBuffer = readBuffer[:readBufferSize]
 			r, err := file.Read(readBuffer)
 			read := int64(r)
 			readBuffer = readBuffer[:read]
-
+			// if we read to end of file time to check truncation etc and process next job.
 			if err == io.EOF || read == 0 {
 				isEOF = true
 				break
 			}
-
 			if err != nil {
-				logger.Fatalf("file %d:%s read error, %s read=%d", sourceID, sourceName, read, err.Error())
+				logger.Fatalf("file %d:%s read error, %s read=%d", sourceID, sourceName, err.Error(), read)
 			}
 
 			processed = 0
@@ -83,18 +87,23 @@ func (w *worker) work(controller inputer, jobProvider *jobProvider, readBufferSi
 					break
 				}
 
+				// \n is line separator, -1 means that file doesn't have new valid logs.
 				pos := int64(bytes.IndexByte(readBuffer[processed:], '\n'))
 				if pos == -1 {
 					break
 				}
+
+				// to keep our total progress over file reading.
 				pos += processed
 
-				// skip first event because file may be opened while event isn't completely written
+				// skip first event because file may be opened while event isn't completely written.
 				if skipLine {
-					job.shouldSkip = false
+					job.shouldSkip.Store(false)
 					skipLine = false
 				} else {
+					// get current offset for reading.
 					offset := lastOffset + accumulated + pos + 1
+					// put into inBuffer read data from readBuffer and accumBuffer (of not empty).
 					if len(accumBuffer) != 0 {
 						accumBuffer = append(accumBuffer, readBuffer[processed:pos+1]...)
 						inBuffer = accumBuffer
@@ -125,7 +134,7 @@ func (w *worker) work(controller inputer, jobProvider *jobProvider, readBufferSi
 			}
 		}
 
-		// don't consider accumulated buffer cause we haven't put any events
+		// don't consider accumulated buffer because we haven't put any events.
 		if !wasPut {
 			accumulated = 0
 		}
@@ -137,22 +146,34 @@ func (w *worker) work(controller inputer, jobProvider *jobProvider, readBufferSi
 				logger.Fatalf("can't set offset, file %d:%s seek error: %s", sourceID, sourceName, err.Error())
 			}
 		}
-
-		// check if file was truncated
+		// check if file was truncated.
 		if isEOF {
-			stat, err := file.Stat()
+			err := w.processEOF(file, job, jobProvider, lastOffset+readTotal)
 			if err != nil {
 				logger.Fatalf("file %d:%s stat error: %s", sourceID, sourceName, err.Error())
 			}
-
-			// file was truncated
-			if lastOffset+readTotal > stat.Size() {
-				jobProvider.truncateJob(job)
-			}
-
-			jobProvider.doneJob(job)
 		} else {
+			// put job in the end of queue.
 			jobProvider.continueJob(job)
 		}
 	}
 }
+
+func (w *worker) processEOF(file *os.File, job *Job, jobProvider *jobProvider, totalOffset int64) error {
+	stat, err := file.Stat()
+	if err != nil {
+		return err
+	}
+
+	// files truncated from time to time, after logs from file was processed.
+	// Position > stat.Size() means that data was truncated and
+	// caret pointer must be moved to start of file.
+	if totalOffset > stat.Size() {
+		jobProvider.truncateJob(job)
+	}
+
+	// Mark job as done till new lines has appeared.
+	jobProvider.doneJob(job)
+
+	return nil
+}
diff --git a/plugin/input/file/worker_test.go b/plugin/input/file/worker_test.go
@@ -60,7 +60,7 @@ func TestWorkerWork(t *testing.T) {
 			inputer := inputerMock{}
 			f, err := os.CreateTemp("/tmp", "worker_test")
 			require.NoError(t, err)
-			fmt.Fprint(f, tt.inFile)
+			_, _ = fmt.Fprint(f, tt.inFile)
 			_, err = f.Seek(0, io.SeekStart)
 			require.NoError(t, err)
 			job := &Job{
@@ -73,7 +73,7 @@ func TestWorkerWork(t *testing.T) {
 				lastEventSeq:   0,
 				isVirgin:       false,
 				isDone:         false,
-				shouldSkip:     false,
+				shouldSkip:     *atomic.NewBool(false),
 				offsets:        sliceMap{},
 				mu:             &sync.Mutex{},
 			}
diff --git a/plugin/input/journalctl/reader.go b/plugin/input/journalctl/reader.go
@@ -12,19 +12,22 @@ import (
 	"go.uber.org/zap"
 )
 
+//nolint:unused
 type journalReaderConfig struct {
 	output   io.Writer
 	cursor   string
 	logger   *zap.SugaredLogger
 	maxLines int
 }
 
+//nolint:unused
 type journalReader struct {
 	config *journalReaderConfig
 	cmd    *exec.Cmd
 	args   []string
 }
 
+//nolint:unused
 func readLines(r io.Reader, config *journalReaderConfig) {
 	reader := bufio.NewReaderSize(r, 1024*1024*10) // max message size
 	totalLines := 0
@@ -59,6 +62,7 @@ func readLines(r io.Reader, config *journalReaderConfig) {
 	}
 }
 
+//nolint:deadcode,unused
 func newJournalReader(config *journalReaderConfig) *journalReader {
 	res := &journalReader{config: config}
 	res.args = []string{
@@ -72,6 +76,7 @@ func newJournalReader(config *journalReaderConfig) *journalReader {
 	return res
 }
 
+//nolint:unused
 func (r *journalReader) start() error {
 	r.config.logger.Infof(`running "journalctl %s"`, strings.Join(r.args, " "))
 	r.cmd = exec.Command("journalctl", r.args...)
@@ -90,6 +95,7 @@ func (r *journalReader) start() error {
 	return nil
 }
 
+//nolint:unused
 func (r *journalReader) stop() error {
 	if err := r.cmd.Process.Kill(); err != nil {
 		return err
diff --git a/plugin/output/gelf/client.go b/plugin/output/gelf/client.go
@@ -11,7 +11,6 @@ type network string
 const transportTCP network = "tcp"
 
 type client struct {
-	network   network
 	useTLS    bool
 	stdClient net.Conn
 	tlsClient *tls.Conn

Original file line number	Diff line number	Diff line change
`@@ -277,7 +277,3 @@ func (f FileD) serveFreeOsMem(_ http.ResponseWriter, _ http.Request) {`
`277`	`277`	`func (f FileD) serveLiveReady(_ http.ResponseWriter, _ http.Request) {`
`278`	`278`	`logger.Infof("live/ready OK")`
`279`	`279`	`}`
`280`		`-`
`281`		`-func (f FileD) servePipelines(_ http.ResponseWriter, _ http.Request) {`
`282`		`- logger.Infof("pipelines OK")`
`283`		`-}`
Original file line number	Diff line number	Diff line change
`@@ -22,8 +22,7 @@ func (b *batcherTail) Error(err string) {`
`22`	`22`	`logger.Panic(err)`
`23`	`23`	`}`
`24`	`24`
`25`		`-func (b *batcherTail) WaitOrPanic(string) {}`
`26`		`-func (b *batcherTail) RecoverFromPanic() {}`
	`25`	`+func (b *batcherTail) RecoverFromPanic() {}`
`27`	`26`
`28`	`27`	`func TestBatcher(t *testing.T) {`
`29`	`28`	`eventCount := 10000000`