[chore] [pkg/stanza] test: add benchmark for File input (#38054)

This benchmark has a scope that sits between the [Stanza File consumer
benchmark](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/v0.120.0/pkg/stanza/fileconsumer/benchmark_test.go)
and a File Log receiver benchmark (which currently does not exist).

The major difference between the File consumer benchmark and the File
input benchmark is that the File input benchmark includes measuring of
the memory allocations made in the File input's
[emit](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/v0.120.0/pkg/stanza/operator/input/file/input.go#L40)
function.

This should allow to assess performance impact of this change:
#37734,
or any similar changes in the future.

Author: andrzej-stencel

pkg/stanza/fileconsumer/attrs/attrs_test.go

@@ -11,7 +11,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/filetest"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
 )
 
 func TestResolver(t *testing.T) {

pkg/stanza/fileconsumer/benchmark_test.go

@@ -17,8 +17,8 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/emit"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/filetest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
 )
 

pkg/stanza/fileconsumer/file_test.go

@@ -22,9 +22,9 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/attrs"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/emit"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/emittest"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/filetest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/reader"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/matcher"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
 )

pkg/stanza/fileconsumer/internal/reader/factory_test.go

@@ -15,9 +15,9 @@ import (
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/attrs"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/emittest"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/filetest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/scanner"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/split"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/trim"
 )

pkg/stanza/fileconsumer/internal/reader/fingerprint_test.go

@@ -11,9 +11,9 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/filetest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/scanner"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
 )
 
 func TestReaderUpdateFingerprint(t *testing.T) {

pkg/stanza/fileconsumer/internal/reader/reader_test.go

@@ -19,9 +19,9 @@ import (
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/attrs"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/emit"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/filetest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/scanner"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
 	internaltime "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/time"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/split"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/trim"

pkg/stanza/fileconsumer/internal/reader/split_test.go

@@ -14,9 +14,9 @@ import (
 	"go.uber.org/zap/zaptest"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/textutils"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/filetest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/header"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/parser/regex"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/split"

pkg/stanza/fileconsumer/internal/reader/validate_test.go

@@ -12,7 +12,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/filetest"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
 )
 
 // When a file it moved, we should detect that our old handle is still valid.

pkg/stanza/fileconsumer/rotation_test.go

@@ -19,7 +19,7 @@ import (
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest/observer"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/filetest"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
 )
 

pkg/stanza/fileconsumer/internal/filetest/filetest.go renamed to pkg/stanza/internal/filetest/filetest.go

@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package filetest // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/filetest"
+package filetest // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
 
 import (
 	"math/rand/v2"

pkg/stanza/fileconsumer/internal/filetest/package_test.go renamed to pkg/stanza/internal/filetest/package_test.go

@@ -0,0 +1,159 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package file
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
+)
+
+func BenchmarkReadExistingLogs(b *testing.B) {
+	testCases := []struct {
+		numLines int
+	}{
+		{
+			numLines: 0,
+		},
+		{
+			numLines: 1,
+		},
+		{
+			numLines: 2,
+		},
+		{
+			numLines: 10,
+		},
+		{
+			numLines: 20,
+		},
+		{
+			numLines: 100,
+		},
+		{
+			numLines: 1000,
+		},
+	}
+
+	for _, tc := range testCases {
+		b.Run(fmt.Sprintf("%d-lines", tc.numLines), func(b *testing.B) {
+			benchmarkReadExistingLogs(b, tc.numLines)
+		})
+	}
+}
+
+func benchmarkReadExistingLogs(b *testing.B, lines int) {
+	logFilePath := generateLogFile(b, lines)
+
+	config := NewConfig()
+	config.Include = []string{
+		logFilePath,
+	}
+	// Use aggressive poll interval so we're not measuring excess sleep time
+	config.PollInterval = time.Microsecond
+	config.StartAt = "beginning"
+
+	op, err := config.Build(componenttest.NewNopTelemetrySettings())
+	require.NoError(b, err)
+	op.SetOutputIDs([]string{benchmarkOutputID})
+
+	doneChan := make(chan struct{})
+	require.NoError(b, op.SetOutputs([]operator.Operator{newBenchmarkOutput(lines, doneChan)}))
+
+	b.ResetTimer()
+	for range b.N {
+		require.NoError(b, op.Start(testutil.NewUnscopedMockPersister()))
+
+		// Wait until all logs are read
+		if lines > 0 {
+			<-doneChan
+		}
+		require.NoError(b, op.Stop())
+	}
+}
+
+func generateLogFile(tb testing.TB, numLines int) string {
+	f := filetest.OpenTemp(tb, tb.TempDir())
+	for range numLines {
+		_, err := f.Write(filetest.TokenWithLength(999))
+		require.NoError(tb, err)
+		_, err = f.WriteString("\n")
+		require.NoError(tb, err)
+	}
+	return f.Name()
+}
+
+type benchmarkOutput struct {
+	totalLogs    int
+	logsReceived int
+	doneChan     chan<- struct{}
+}
+
+// Assert that benchmarkOutput implements the operator.Operator interface
+var _ operator.Operator = (*benchmarkOutput)(nil)
+
+func newBenchmarkOutput(totalLogs int, doneChan chan<- struct{}) *benchmarkOutput {
+	return &benchmarkOutput{
+		totalLogs: totalLogs,
+		doneChan:  doneChan,
+	}
+}
+
+// CanOutput implements operator.Operator.
+func (o *benchmarkOutput) CanOutput() bool { return false }
+
+// CanProcess implements operator.Operator.
+func (o *benchmarkOutput) CanProcess() bool { return true }
+
+// GetOutputIDs implements operator.Operator.
+func (o *benchmarkOutput) GetOutputIDs() []string { return nil }
+
+const benchmarkOutputID = "benchmark"
+
+// ID implements operator.Operator.
+func (o *benchmarkOutput) ID() string { return benchmarkOutputID }
+
+// Logger implements operator.Operator.
+func (o *benchmarkOutput) Logger() *zap.Logger { return nil }
+
+// Outputs implements operator.Operator.
+func (o *benchmarkOutput) Outputs() []operator.Operator { return nil }
+
+// SetOutputIDs implements operator.Operator.
+func (o *benchmarkOutput) SetOutputIDs([]string) {}
+
+// SetOutputs implements operator.Operator.
+func (o *benchmarkOutput) SetOutputs([]operator.Operator) error { return nil }
+
+// Start implements operator.Operator.
+func (o *benchmarkOutput) Start(operator.Persister) error {
+	return nil
+}
+
+// Stop implements operator.Operator.
+func (o *benchmarkOutput) Stop() error {
+	return nil
+}
+
+// Type implements operator.Operator.
+func (o *benchmarkOutput) Type() string { return "benchmark_output" }
+
+func (o *benchmarkOutput) Process(_ context.Context, _ *entry.Entry) error {
+	o.logsReceived++
+	if o.logsReceived == o.totalLogs {
+		o.doneChan <- struct{}{}
+		o.logsReceived = 0
+	}
+	return nil
+}