Cleanup file input 2

operator/builtin/input/file/benchmark_test.go

@@ -0,0 +1,84 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package file
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/open-telemetry/opentelemetry-log-collection/operator"
+	"github.com/open-telemetry/opentelemetry-log-collection/testutil"
+)
+
+type fileInputBenchmark struct {
+	name   string
+	config *InputConfig
+}
+
+func BenchmarkFileInput(b *testing.B) {
+	cases := []fileInputBenchmark{
+		{
+			"Default",
+			NewInputConfig("test_id"),
+		},
+		{
+			"NoFileName",
+			func() *InputConfig {
+				cfg := NewInputConfig("test_id")
+				cfg.IncludeFileName = false
+				return cfg
+			}(),
+		},
+	}
+
+	for _, tc := range cases {
+		b.Run(tc.name, func(b *testing.B) {
+			tempDir := testutil.NewTempDir(b)
+			path := filepath.Join(tempDir, "in.log")
+
+			cfg := tc.config
+			cfg.OutputIDs = []string{"fake"}
+			cfg.Include = []string{path}
+			cfg.StartAt = "beginning"
+
+			ops, err := cfg.Build(testutil.NewBuildContext(b))
+			require.NoError(b, err)
+			op := ops[0]
+
+			fakeOutput := testutil.NewFakeOutput(b)
+			err = op.SetOutputs([]operator.Operator{fakeOutput})
+			require.NoError(b, err)
+
+			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")
+			}
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				<-fakeOutput.Received
+			}
+		})
+	}
+}

operator/builtin/input/file/config_test.go

@@ -22,11 +22,13 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/open-telemetry/opentelemetry-log-collection/entry"
+	"github.com/open-telemetry/opentelemetry-log-collection/operator"
 	"github.com/open-telemetry/opentelemetry-log-collection/operator/helper"
 	"github.com/open-telemetry/opentelemetry-log-collection/operator/helper/operatortest"
+	"github.com/open-telemetry/opentelemetry-log-collection/testutil"
 )
 
-func TestConfig(t *testing.T) {
+func TestUnmarshal(t *testing.T) {
 	cases := []operatortest.ConfigUnmarshalTest{
 		{
 			Name:      "default",
@@ -512,6 +514,156 @@ func TestConfig(t *testing.T) {
 	}
 }
 
+func TestBuild(t *testing.T) {
+	t.Parallel()
+	fakeOutput := testutil.NewMockOperator("$.fake")
+
+	basicConfig := func() *InputConfig {
+		cfg := NewInputConfig("testfile")
+		cfg.OutputIDs = []string{"fake"}
+		cfg.Include = []string{"/var/log/testpath.*"}
+		cfg.Exclude = []string{"/var/log/testpath.ex*"}
+		cfg.PollInterval = helper.Duration{Duration: 10 * time.Millisecond}
+		return cfg
+	}
+
+	cases := []struct {
+		name             string
+		modifyBaseConfig func(*InputConfig)
+		errorRequirement require.ErrorAssertionFunc
+		validate         func(*testing.T, *InputOperator)
+	}{
+		{
+			"Basic",
+			func(f *InputConfig) {},
+			require.NoError,
+			func(t *testing.T, f *InputOperator) {
+				require.Equal(t, f.OutputOperators[0], fakeOutput)
+				require.Equal(t, f.Include, []string{"/var/log/testpath.*"})
+				require.Equal(t, f.FilePathField, entry.NewNilField())
+				require.Equal(t, f.FileNameField, entry.NewAttributeField("file_name"))
+				require.Equal(t, f.PollInterval, 10*time.Millisecond)
+			},
+		},
+		{
+			"BadIncludeGlob",
+			func(f *InputConfig) {
+				f.Include = []string{"["}
+			},
+			require.Error,
+			nil,
+		},
+		{
+			"BadExcludeGlob",
+			func(f *InputConfig) {
+				f.Include = []string{"["}
+			},
+			require.Error,
+			nil,
+		},
+		{
+			"MultilineConfiguredStartAndEndPatterns",
+			func(f *InputConfig) {
+				f.Multiline = helper.MultilineConfig{
+					LineEndPattern:   "Exists",
+					LineStartPattern: "Exists",
+				}
+			},
+			require.Error,
+			nil,
+		},
+		{
+			"MultilineConfiguredStartPattern",
+			func(f *InputConfig) {
+				f.Multiline = helper.MultilineConfig{
+					LineStartPattern: "START.*",
+				}
+			},
+			require.NoError,
+			func(t *testing.T, f *InputOperator) {},
+		},
+		{
+			"MultilineConfiguredEndPattern",
+			func(f *InputConfig) {
+				f.Multiline = helper.MultilineConfig{
+					LineEndPattern: "END.*",
+				}
+			},
+			require.NoError,
+			func(t *testing.T, f *InputOperator) {},
+		},
+		{
+			"InvalidEncoding",
+			func(f *InputConfig) {
+				f.Encoding = helper.EncodingConfig{Encoding: "UTF-3233"}
+			},
+			require.Error,
+			nil,
+		},
+		{
+			"LineStartAndEnd",
+			func(f *InputConfig) {
+				f.Multiline = helper.MultilineConfig{
+					LineStartPattern: ".*",
+					LineEndPattern:   ".*",
+				}
+			},
+			require.Error,
+			nil,
+		},
+		{
+			"NoLineStartOrEnd",
+			func(f *InputConfig) {
+				f.Multiline = helper.MultilineConfig{}
+			},
+			require.NoError,
+			func(t *testing.T, f *InputOperator) {},
+		},
+		{
+			"InvalidLineStartRegex",
+			func(f *InputConfig) {
+				f.Multiline = helper.MultilineConfig{
+					LineStartPattern: "(",
+				}
+			},
+			require.Error,
+			nil,
+		},
+		{
+			"InvalidLineEndRegex",
+			func(f *InputConfig) {
+				f.Multiline = helper.MultilineConfig{
+					LineEndPattern: "(",
+				}
+			},
+			require.Error,
+			nil,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			tc := tc
+			t.Parallel()
+			cfg := basicConfig()
+			tc.modifyBaseConfig(cfg)
+
+			ops, err := cfg.Build(testutil.NewBuildContext(t))
+			tc.errorRequirement(t, err)
+			if err != nil {
+				return
+			}
+			op := ops[0]
+
+			err = op.SetOutputs([]operator.Operator{fakeOutput})
+			require.NoError(t, err)
+
+			fileInput := op.(*InputOperator)
+			tc.validate(t, fileInput)
+		})
+	}
+}
+
 func defaultCfg() *InputConfig {
 	return NewInputConfig("file_input")
 }