Weirdness with assert not failing tests

Author: camdencheek

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/spf13/viper v1.6.2
 	github.com/stretchr/testify v1.5.1
 	github.com/xeipuuv/gojsonschema v1.2.0
+	go.uber.org/goleak v1.0.0
 	go.uber.org/zap v1.14.0
 	golang.org/x/lint v0.0.0-20200302205851-738671d3881b // indirect
 	golang.org/x/net v0.0.0-20200301022130-244492dfa37a // indirect

go.sum

@@ -169,6 +169,8 @@ go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.5.0 h1:OI5t8sDa1Or+q8AeE+yKeB/SDYioSHAgcVljj9JIETY=
 go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
+go.uber.org/goleak v1.0.0 h1:qsup4IcBdlmsnGfqyLl4Ntn3C2XCCuKAE7DwHpScyUo=
+go.uber.org/goleak v1.0.0/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/multierr v1.3.0 h1:sFPn2GLc3poCkfrpIXGhBD2X0CMIo4Q/zSULXrj/+uc=
 go.uber.org/multierr v1.3.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4=
@@ -262,6 +264,7 @@ golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgw
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5 h1:hKsoRgsbwY1NafxrwTs+k64bikrLBkAgPir1TNCj3Zs=
 golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191130070609-6e064ea0cf2d/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=

plugin/builtin/fileinput/file_source.go

@@ -15,8 +15,6 @@ import (
 	"time"
 
 	pg "github.com/bluemedora/bplogagent/plugin"
-	// using Docker's filenotify so we can fall back to polling
-	// for envs where notify isn't available
 )
 
 func init() {
@@ -49,45 +47,52 @@ func (c FileSourceConfig) Build(buildContext pg.BuildContext) (pg.Plugin, error)
 		return nil, fmt.Errorf("build default outputter: %s", err)
 	}
 
+	// Ensure includes can be parsed as globs
 	for _, include := range c.Include {
 		_, err := filepath.Match(include, "")
 		if err != nil {
 			return nil, fmt.Errorf("parse include glob: %s", err)
 		}
 	}
 
+	// Ensure excludes can be parsed as globs
 	for _, exclude := range c.Exclude {
 		_, err := filepath.Match(exclude, "")
 		if err != nil {
 			return nil, fmt.Errorf("parse exclude glob: %s", err)
 		}
 	}
 
+	// Determine the split function for log entries
 	var splitFunc bufio.SplitFunc
 	if c.Multiline == nil {
 		splitFunc = bufio.ScanLines
-	} else if c.Multiline.LineEndPattern != "" && c.Multiline.LineStartPattern != "" {
-		return nil, fmt.Errorf("cannot configure both line_start_pattern and line_end_pattern")
-	} else if c.Multiline.LineEndPattern != "" {
-		re, err := regexp.Compile(c.Multiline.LineEndPattern)
-		if err != nil {
-			return nil, fmt.Errorf("compile line end regex: %s", err)
-		}
-		splitFunc = NewLineEndSplitFunc(re)
-	} else if c.Multiline.LineStartPattern != "" {
-		re, err := regexp.Compile(c.Multiline.LineStartPattern)
-		if err != nil {
-			return nil, fmt.Errorf("compile line start regex: %s", err)
+	} else {
+		definedLineEndPattern := c.Multiline.LineEndPattern != ""
+		definedLineStartPattern := c.Multiline.LineStartPattern != ""
+
+		switch {
+		case definedLineEndPattern == definedLineStartPattern:
+			return nil, fmt.Errorf("if multiline is configured, exactly one of line_start_pattern or line_end_pattern must be set")
+		case definedLineEndPattern:
+			re, err := regexp.Compile(c.Multiline.LineEndPattern)
+			if err != nil {
+				return nil, fmt.Errorf("compile line end regex: %s", err)
+			}
+			splitFunc = NewLineEndSplitFunc(re)
+		case definedLineStartPattern:
+			re, err := regexp.Compile(c.Multiline.LineStartPattern)
+			if err != nil {
+				return nil, fmt.Errorf("compile line start regex: %s", err)
+			}
+			splitFunc = NewLineStartSplitFunc(re)
 		}
-		splitFunc = NewLineStartSplitFunc(re)
-	} else if c.Multiline.LineEndPattern == "" && c.Multiline.LineStartPattern == "" {
-		return nil, fmt.Errorf("if multiline is configured, either line_start_pattern or line_end_pattern must be configured")
 	}
 
+	// Parse the poll interval
 	if c.PollInterval < 0 {
 		return nil, fmt.Errorf("poll_interval must be greater than zero if configured")
 	}
-
 	pollInterval := func() time.Duration {
 		if c.PollInterval == 0 {
 			return 5 * time.Second
@@ -106,7 +111,7 @@ func (c FileSourceConfig) Build(buildContext pg.BuildContext) (pg.Plugin, error)
 		PollInterval: pollInterval,
 
 		fileCreated:      make(chan string),
-		fileTouched:      make(chan string),
+		fileTouched:      make(chan struct{}),
 		fileRemoved:      make(chan *FileWatcher),
 		directoryRemoved: make(chan *DirectoryWatcher),
 	}
@@ -135,7 +140,7 @@ type FileSource struct {
 
 	fileCreated      chan string
 	fileRemoved      chan *FileWatcher
-	fileTouched      chan string
+	fileTouched      chan struct{}
 	directoryRemoved chan *DirectoryWatcher
 }
 
@@ -176,6 +181,7 @@ func (f *FileSource) Start() error {
 				f.Debugw("Received directory removed notification", "path", watcher.path)
 				f.removeDirectoryWatcher(watcher)
 			case <-f.fileTouched:
+				// swallow messages as a notification that it's safe to read?
 			}
 		}
 	}()
@@ -227,7 +233,7 @@ func (f *FileSource) tryAddFile(ctx context.Context, path string, globCheck bool
 		return
 	}
 
-	watcher, err := NewFileWatcher(path, f, startFromBeginning, f.SplitFunc, f.PollInterval, f.SugaredLogger)
+	watcher, err := NewFileWatcher(path, f.Output, startFromBeginning, f.SplitFunc, f.PollInterval, f.SugaredLogger)
 	if err != nil {
 		if pathError, ok := err.(*os.PathError); ok && pathError.Err.Error() == "no such file or directory" {
 			f.Debugw("File deleted before it could be read", "path", path)
@@ -283,6 +289,7 @@ func (f *FileSource) checkPath(path string, checkCopy bool) (createWatcher bool,
 
 	for _, watcher := range f.fileWatchers {
 		// TODO what if multiple match? anything?
+		// TODO how do links (hard and soft) interact with this logic?
 		if watcher.dev == dev && watcher.inode == inode {
 			if watcher.path == path {
 				return false, false, nil
@@ -303,6 +310,7 @@ func (f *FileSource) checkPath(path string, checkCopy bool) (createWatcher bool,
 }
 
 func (f *FileSource) fingerprint(file *os.File) string {
+	// TODO make sure resetting the seek location isn't messing with things
 	_, err := file.Seek(0, io.SeekStart)
 	if err != nil {
 		panic(err)

plugin/builtin/fileinput/file_watcher.go

@@ -13,23 +13,31 @@ import (
 	"go.uber.org/zap"
 )
 
-// FileWatcher is a wrapper around `fsnotify` that periodically polls
-// to mitigate issues with filesystems that don't support notify events
+// FileWatcher is a wrapper around `fsnotify` that periodically polls to provide
+// a fallback for filesystems and platforms that don't support event notification
 type FileWatcher struct {
 	inode        uint64
 	dev          uint64
 	path         string
 	file         *os.File
 	offset       int64
 	pollInterval time.Duration
-	fileSource   *FileSource
 	cancel       context.CancelFunc
 	splitFunc    bufio.SplitFunc
+	output       func(*entry.Entry) error
 
 	*zap.SugaredLogger
 }
 
-func NewFileWatcher(path string, fileSource *FileSource, startFromBeginning bool, splitFunc bufio.SplitFunc, pollInterval time.Duration, logger *zap.SugaredLogger) (*FileWatcher, error) {
+func NewFileWatcher(
+	path string,
+	output func(*entry.Entry) error,
+	startFromBeginning bool,
+	splitFunc bufio.SplitFunc,
+	pollInterval time.Duration,
+	logger *zap.SugaredLogger,
+) (*FileWatcher, error) {
+
 	file, err := os.Open(path)
 	if err != nil {
 		return nil, err
@@ -66,37 +74,44 @@ func NewFileWatcher(path string, fileSource *FileSource, startFromBeginning bool
 		path:          path,
 		pollInterval:  pollInterval,
 		offset:        offset,
-		fileSource:    fileSource,
 		splitFunc:     splitFunc,
-		SugaredLogger: logger.With("path", path),
+		output:        output,
+		SugaredLogger: logger.Named("file_watcher").With("path", path),
 	}, nil
 }
 
 func (w *FileWatcher) Watch(startCtx context.Context) error {
 	ctx, cancel := context.WithCancel(startCtx)
 	w.cancel = cancel
 
+	// Create the watcher
 	watcher, err := fsnotify.NewWatcher()
 	if err != nil {
 		// TODO if falling back to polling, should we set the default lower?
 		w.Infow("Failed to create notifying watcher. Falling back to polling only", "error", err)
 		watcher = &fsnotify.Watcher{} // create an empty watcher whose channels are just nil
 	} else {
+		defer watcher.Close()
 		err = watcher.Add(w.path)
 		if err != nil {
 			w.Infow("Failed to add path to watcher. Falling back to polling only", "error", err)
 			watcher = &fsnotify.Watcher{} // create an empty watcher whose channels are just nil
 		}
 	}
 
+	// Keep a persistent open file
 	file, err := os.Open(w.path)
 	if err != nil {
 		return err
 	}
 	defer file.Close()
 	w.file = file
 
-	w.checkFile(ctx) // Check it once initially for responsive startup
+	// Check it once initially for responsive startup
+	err = w.checkReadFile(ctx)
+	if err != nil {
+		return err
+	}
 
 	for {
 		// TODO actually test all these cases
@@ -109,62 +124,72 @@ func (w *FileWatcher) Watch(startCtx context.Context) error {
 		select {
 		case <-ctx.Done():
 			timer.Stop()
-			err := watcher.Close()
-			if err != nil {
-				return err
-			}
+			return nil
 		case event, ok := <-watcher.Events:
 			timer.Stop()
 			if !ok {
 				return nil
 			}
 			if event.Op&fsnotify.Remove > 0 {
-				watcher.Close()
-				w.fileSource.fileRemoved <- w
-				continue
+				return watcher.Close()
 			}
 			if event.Op&(fsnotify.Write|fsnotify.Chmod) > 0 {
-				w.checkFile(ctx)
+				err := w.checkReadFile(ctx)
+				if err != nil {
+					return err
+				}
 			}
-			// ignore chmod and rename (rename is covered by directory create)
+			// ignore rename (rename is covered by directory create)
 		case <-timer.C:
-			w.checkFile(ctx)
+			err := w.checkReadFile(ctx)
+			if err != nil {
+				return err
+			}
 		case err := <-watcher.Errors:
 			timer.Stop()
 			return err
 		}
 	}
 }
 
-func (w *FileWatcher) checkFile(ctx context.Context) {
+func (w *FileWatcher) checkReadFile(ctx context.Context) error {
+	// TODO ensure that none of the errors thrown in here are recoverable
+	// since returning an error triggers a return from the watch function
 	select {
-	case w.fileSource.fileTouched <- w.path:
 	case <-ctx.Done():
-		return
+		return nil
+	default:
 	}
 
+	// TODO check if the file still exists
+
 	fileInfo, err := w.file.Stat()
 	if err != nil {
-		w.Errorw("Failed to get file info", "error", err) // TODO is this a recoverable error?
-		return
+		return err
 	}
 
 	if fileInfo.Size() < w.offset {
 		w.Debug("Detected file truncation. Starting from beginning")
 		w.offset, err = w.file.Seek(0, 0)
 		if err != nil {
-			w.Errorw("Failed to seek to file start", "error", err)
-			return
+			return fmt.Errorf("seek to start: %s", err)
+		}
+		err := w.readToEnd(ctx)
+		if err != nil {
+			return err
 		}
-		w.readToEnd(ctx)
 	} else if fileInfo.Size() > w.offset {
-		w.readToEnd(ctx)
+		err := w.readToEnd(ctx)
+		if err != nil {
+			return err
+		}
 	}
 
 	// do nothing if the file hasn't changed size
+	return nil
 }
 
-func (w *FileWatcher) readToEnd(ctx context.Context) {
+func (w *FileWatcher) readToEnd(ctx context.Context) error {
 	// TODO seek to last offset?
 	scanner := bufio.NewScanner(w.file)
 	scanner.Split(w.splitFunc)
@@ -173,16 +198,13 @@ func (w *FileWatcher) readToEnd(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
-			return // Stop reading if closed
+			return nil // Stop reading if closed
 		default:
 		}
 
 		ok := scanner.Scan()
 		if !ok {
-			if err := scanner.Err(); err != nil {
-				w.Warn("Failed to scan file", "error", err)
-			}
-			break
+			return scanner.Err()
 		}
 
 		message := scanner.Text()
@@ -194,13 +216,17 @@ func (w *FileWatcher) readToEnd(ctx context.Context) {
 			},
 		}
 
-		w.fileSource.Output(entry)
+		err := w.output(entry)
+		if err != nil {
+			return fmt.Errorf("output entry: %s", err)
+		}
 
-		var err error
+		// TODO does this actually work how I think it does with the scanner?
+		// I'm unsure if the scanner peeks ahead, or actually advances the reader
+		// every time it tries to parse something. This needs to be tested
 		w.offset, err = w.file.Seek(0, 1) // get current file offset
 		if err != nil {
-			w.Errorw("Failed to get current offset", "error", err)
-			return
+			return fmt.Errorf("get current offset: %s", err)
 		}
 	}
 }