Refactor file_input benchmarking and add additional cases for notable…

… scenarios (#166)

Makefile

@@ -25,7 +25,7 @@ test-only:
 
 .PHONY: bench
 bench:
-	$(MAKE) for-all CMD="go test -run=NONE -bench '.*' ./... -benchmem"
+	go test -benchmem -run=^$$ -bench ^* ./...
 
 .PHONY: clean
 clean:

operator/builtin/input/file/benchmark_test.go

@@ -15,6 +15,7 @@
 package file
 
 import (
+	"io/ioutil"
 	"os"
 	"path/filepath"
 	"testing"
@@ -27,33 +28,132 @@ import (
 
 type fileInputBenchmark struct {
 	name   string
-	config *InputConfig
+	paths  []string
+	config func() *InputConfig
+}
+
+type benchFile struct {
+	*os.File
+	log func(int)
+}
+
+func simpleTextFile(file *os.File) *benchFile {
+	line := stringWithLength(49) + "\n"
+	return &benchFile{
+		File: file,
+		log:  func(_ int) { file.WriteString(line) },
+	}
 }
 
 func BenchmarkFileInput(b *testing.B) {
 	cases := []fileInputBenchmark{
 		{
-			"Default",
-			NewInputConfig("test_id"),
+			name: "Single",
+			paths: []string{
+				"file0.log",
+			},
+			config: func() *InputConfig {
+				cfg := NewInputConfig("test_id")
+				cfg.Include = []string{
+					"file0.log",
+				}
+				return cfg
+			},
+		},
+		{
+			name: "Glob",
+			paths: []string{
+				"file0.log",
+				"file1.log",
+				"file2.log",
+				"file3.log",
+			},
+			config: func() *InputConfig {
+				cfg := NewInputConfig("test_id")
+				cfg.Include = []string{"file*.log"}
+				return cfg
+			},
+		},
+		{
+			name: "MultiGlob",
+			paths: []string{
+				"file0.log",
+				"file1.log",
+				"log0.log",
+				"log1.log",
+			},
+			config: func() *InputConfig {
+				cfg := NewInputConfig("test_id")
+				cfg.Include = []string{
+					"file*.log",
+					"log*.log",
+				}
+				return cfg
+			},
+		},
+		{
+			name: "MaxConcurrent",
+			paths: []string{
+				"file0.log",
+				"file1.log",
+				"file2.log",
+				"file3.log",
+			},
+			config: func() *InputConfig {
+				cfg := NewInputConfig("test_id")
+				cfg.Include = []string{
+					"file*.log",
+				}
+				cfg.MaxConcurrentFiles = 1
+				return cfg
+			},
 		},
 		{
-			"NoFileName",
-			func() *InputConfig {
+			name: "FngrPrntLarge",
+			paths: []string{
+				"file0.log",
+			},
+			config: func() *InputConfig {
 				cfg := NewInputConfig("test_id")
-				cfg.IncludeFileName = false
+				cfg.Include = []string{
+					"file*.log",
+				}
+				cfg.FingerprintSize = 10 * defaultFingerprintSize
 				return cfg
-			}(),
+			},
+		},
+		{
+			name: "FngrPrntSmall",
+			paths: []string{
+				"file0.log",
+			},
+			config: func() *InputConfig {
+				cfg := NewInputConfig("test_id")
+				cfg.Include = []string{
+					"file*.log",
+				}
+				cfg.FingerprintSize = defaultFingerprintSize / 10
+				return cfg
+			},
 		},
 	}
 
-	for _, tc := range cases {
-		b.Run(tc.name, func(b *testing.B) {
-			tempDir := testutil.NewTempDir(b)
-			path := filepath.Join(tempDir, "in.log")
+	for _, bench := range cases {
+		b.Run(bench.name, func(b *testing.B) {
+			rootDir, err := ioutil.TempDir("", "")
+			require.NoError(b, err)
+
+			files := []*benchFile{}
+			for _, path := range bench.paths {
+				file := openFile(b, filepath.Join(rootDir, path))
+				files = append(files, simpleTextFile(file))
+			}
 
-			cfg := tc.config
+			cfg := bench.config()
 			cfg.OutputIDs = []string{"fake"}
-			cfg.Include = []string{path}
+			for i, inc := range cfg.Include {
+				cfg.Include[i] = filepath.Join(rootDir, inc)
+			}
 			cfg.StartAt = "beginning"
 
 			ops, err := cfg.Build(testutil.NewBuildContext(b))
@@ -64,19 +164,29 @@ func BenchmarkFileInput(b *testing.B) {
 			err = op.SetOutputs([]operator.Operator{fakeOutput})
 			require.NoError(b, err)
 
+			// write half the lines before starting
+			mid := b.N / 2
+			for i := 0; i < mid; i++ {
+				for _, file := range files {
+					file.log(i)
+				}
+			}
+
+			b.ResetTimer()
 			err = op.Start(testutil.NewMockPersister("test"))
 			defer op.Stop()
 			require.NoError(b, err)
 
-			file, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0666)
-			require.NoError(b, err)
-
-			for i := 0; i < b.N; i++ {
-				file.WriteString("testlog\n")
-			}
+			// write the remainder of lines while running
+			go func() {
+				for i := mid; i < b.N; i++ {
+					for _, file := range files {
+						file.log(i)
+					}
+				}
+			}()
 
-			b.ResetTimer()
-			for i := 0; i < b.N; i++ {
+			for i := 0; i < b.N*len(files); i++ {
 				<-fakeOutput.Received
 			}
 		})

operator/builtin/input/file/config.go

@@ -42,6 +42,7 @@ func NewInputConfig(operatorID string) *InputConfig {
 		IncludeFileName:    true,
 		IncludeFilePath:    false,
 		StartAt:            "end",
+		FingerprintSize:    defaultFingerprintSize,
 		MaxLogSize:         defaultMaxLogSize,
 		MaxConcurrentFiles: defaultMaxConcurrentFiles,
 		Encoding:           helper.NewEncodingConfig(),

operator/builtin/input/file/util_test.go

@@ -64,10 +64,10 @@ func newTestFileOperator(t *testing.T, cfgMod func(*InputConfig), outMod func(*t
 	return op.(*InputOperator), fakeOutput.Received, tempDir
 }
 
-func openFile(t testing.TB, path string) *os.File {
-	file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0777)
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = file.Close() })
+func openFile(tb testing.TB, path string) *os.File {
+	file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0600)
+	require.NoError(tb, err)
+	tb.Cleanup(func() { _ = file.Close() })
 	return file
 }
 

testutil/util.go

@@ -22,6 +22,7 @@ import (
 	"sync"
 	"testing"
 
+	"go.uber.org/zap/zapcore"
 	"go.uber.org/zap/zaptest"
 
 	"github.com/open-telemetry/opentelemetry-log-collection/logger"
@@ -46,7 +47,7 @@ func NewTempDir(t testing.TB) string {
 // NewBuildContext will return a new build context for testing
 func NewBuildContext(t testing.TB) operator.BuildContext {
 	return operator.BuildContext{
-		Logger:    logger.New(zaptest.NewLogger(t).Sugar()),
+		Logger:    logger.New(zaptest.NewLogger(t, zaptest.Level(zapcore.ErrorLevel)).Sugar()),
 		Namespace: "$",
 	}
 }