[pkg/stanza] Introduce batching logs in File consumer (#36663)

#### Description

Modifies the File consumer to emit logs in batches as opposed to sending
each log individually through the Stanza pipeline and on to the Log
Emitter.

Here are the changes introduced:

-
6b4c9fe
Changed the `Reader::ReadToEnd` method in File consumer to collect the
tokens scanned from the file into batches. At this point, the Reader
still emits each token individually, as the `emit.Callback` function
only accepts a single token.
-
c206995
Changed `emit.Callback` function signature to accept a slice of tokens
as opposed to a single token, and changed the Reader to emit a batch of
tokens in one request. At this point, the batches are still split into
individual tokens inside the `emit` function, because the Stanza
operators can only process one entry at a time.
-
aedda3a
Added `ProcessBatch` method to Stanza operators and used it in the
`emit` function. At this point, the batch of tokens is translated to a
batch of entries and passed to Log Emitter as a whole batch. The batch
is still split in the Log Emitter, which calls `consumeFunc` for each
entry in a loop.
-
13d6054
Changed the LogEmitter to add the whole batch to its buffer, as opposed
to adding entries one by one.

**Slice of entries `[]entry.Entry` vs. slice of pointers
`[]*entry.Entry`**

I considered whether the `ProcessBatch` method in the `Operator`
interface should accept a slice of structs `[]entry.Entry` or a slice of
pointers `[]*entry.Entry`. I ran some tests (similar to
#35454)
and they showed a 7-10% performance loss when using a slice of structs
vs. a slice of pointers. That's why I decided to use the slice of
pointers `[]*entry.Entry`.

#### Link to tracking issue

- Fixes
#35455

#### Testing

No changes in tests. The goal is for the functionality to not change and
for performance to not decrease.

I have added a new benchmark in a separate PR
#38054
that should be helpful in assessing the performance impact of this
change.

#### Documentation

These are internal changes, no user documentation needs changing.

Author: andrzej-stencel

.chloggen/batch-logs-in-file-consumer-2.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: breaking
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: pkg/stanza
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add method `ProcessBatch` to `Operator` interface in `pkg/stanza/operator` package to support batch processing.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35455]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [api]

.chloggen/batch-logs-in-file-consumer.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: breaking
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: pkg/stanza
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Change signature of `emit.Callback` function in `pkg/stanza/fileconsumer/emit` package to emit multiple tokens.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35455]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [api]

pkg/stanza/adapter/mocks_test.go

@@ -50,6 +50,10 @@ func (o *UnstartableOperator) Start(_ operator.Persister) error {
 	return errors.New("something very unusual happened")
 }
 
+func (o *UnstartableOperator) ProcessBatch(_ context.Context, _ []*entry.Entry) error {
+	return nil
+}
+
 // Process will return nil
 func (o *UnstartableOperator) Process(_ context.Context, _ *entry.Entry) error {
 	return nil

pkg/stanza/fileconsumer/benchmark_test.go

@@ -16,7 +16,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"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/fingerprint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/internal/filetest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
@@ -188,8 +187,8 @@ func BenchmarkFileInput(b *testing.B) {
 			cfg.PollInterval = time.Microsecond
 
 			doneChan := make(chan bool, len(files))
-			callback := func(_ context.Context, token emit.Token) error {
-				if len(token.Body) == 0 {
+			callback := func(_ context.Context, tokens [][]byte, _ map[string]any, _ int64) error {
+				if len(tokens) > 0 && len(tokens[len(tokens)-1]) == 0 {
 					doneChan <- true
 				}
 				return nil

pkg/stanza/fileconsumer/emit/emit.go

@@ -7,7 +7,7 @@ import (
 	"context"
 )
 
-type Callback func(ctx context.Context, token Token) error
+type Callback func(ctx context.Context, tokens [][]byte, attributes map[string]any, lastRecordNumber int64) error
 
 type Token struct {
 	Body       []byte

pkg/stanza/fileconsumer/file_test.go

@@ -1580,78 +1580,3 @@ func TestReadGzipCompressedLogsFromEnd(t *testing.T) {
 	operator.poll(context.TODO())
 	sink.ExpectToken(t, []byte("testlog4"))
 }
-
-func TestIncludeFileRecordNumber(t *testing.T) {
-	t.Parallel()
-
-	tempDir := t.TempDir()
-	cfg := NewConfig().includeDir(tempDir)
-	cfg.StartAt = "beginning"
-	cfg.IncludeFileRecordNumber = true
-	operator, sink := testManager(t, cfg)
-
-	// Create a file, then start
-	temp := filetest.OpenTemp(t, tempDir)
-	filetest.WriteString(t, temp, "testlog1\n")
-
-	require.NoError(t, operator.Start(testutil.NewUnscopedMockPersister()))
-	defer func() {
-		require.NoError(t, operator.Stop())
-	}()
-
-	sink.ExpectCall(t, []byte("testlog1"), map[string]any{
-		attrs.LogFileName:         filepath.Base(temp.Name()),
-		attrs.LogFileRecordNumber: int64(1),
-	})
-}
-
-func TestIncludeFileRecordNumberWithHeaderConfigured(t *testing.T) {
-	t.Parallel()
-
-	tempDir := t.TempDir()
-	cfg := NewConfig().includeDir(tempDir)
-	cfg.StartAt = "beginning"
-	cfg.IncludeFileRecordNumber = true
-	cfg = cfg.withHeader("^#", "(?P<header_attr>[A-z]+)")
-	operator, sink := testManager(t, cfg)
-
-	// Create a file, then start
-	temp := filetest.OpenTemp(t, tempDir)
-	filetest.WriteString(t, temp, "#abc\n#xyz: headerValue2\ntestlog1\n")
-
-	require.NoError(t, operator.Start(testutil.NewUnscopedMockPersister()))
-	defer func() {
-		require.NoError(t, operator.Stop())
-	}()
-
-	sink.ExpectCall(t, []byte("testlog1"), map[string]any{
-		attrs.LogFileName:         filepath.Base(temp.Name()),
-		attrs.LogFileRecordNumber: int64(1),
-		"header_attr":             "xyz",
-	})
-}
-
-func TestIncludeFileRecordNumberWithHeaderConfiguredButMissing(t *testing.T) {
-	t.Parallel()
-
-	tempDir := t.TempDir()
-	cfg := NewConfig().includeDir(tempDir)
-	cfg.StartAt = "beginning"
-	cfg.IncludeFileRecordNumber = true
-	cfg = cfg.withHeader("^#", "(?P<header_key>[A-z]+): (?P<header_value>[A-z]+)")
-	operator, sink := testManager(t, cfg)
-
-	// Create a file, then start
-	temp := filetest.OpenTemp(t, tempDir)
-	filetest.WriteString(t, temp, "testlog1\n")
-
-	require.NoError(t, operator.Start(testutil.NewUnscopedMockPersister()))
-	defer func() {
-		require.NoError(t, operator.Stop())
-	}()
-
-	sink.ExpectCall(t, []byte("testlog1"), map[string]any{
-		attrs.LogFileName:         filepath.Base(temp.Name()),
-		attrs.LogFileRecordNumber: int64(1),
-	})
-}

pkg/stanza/fileconsumer/internal/emittest/nop.go

@@ -5,10 +5,8 @@ package emittest // import "github.com/open-telemetry/opentelemetry-collector-co
 
 import (
 	"context"
-
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/emit"
 )
 
-func Nop(_ context.Context, _ emit.Token) error {
+func Nop(_ context.Context, _ [][]byte, _ map[string]any, _ int64) error {
 	return nil
 }

pkg/stanza/fileconsumer/internal/emittest/nop_test.go

@@ -8,10 +8,8 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
-
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/emit"
 )
 
 func TestNop(t *testing.T) {
-	require.NoError(t, Nop(context.Background(), emit.Token{}))
+	require.NoError(t, Nop(context.Background(), [][]byte{}, map[string]any{}, int64(0)))
 }

pkg/stanza/fileconsumer/internal/emittest/sink.go

@@ -52,11 +52,13 @@ func NewSink(opts ...SinkOpt) *Sink {
 	return &Sink{
 		emitChan: emitChan,
 		timeout:  cfg.timeout,
-		Callback: func(ctx context.Context, token emit.Token) error {
-			select {
-			case <-ctx.Done():
-				return ctx.Err()
-			case emitChan <- token:
+		Callback: func(ctx context.Context, tokens [][]byte, attributes map[string]any, _ int64) error {
+			for _, token := range tokens {
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case emitChan <- emit.NewToken(token, attributes):
+				}
 			}
 			return nil
 		},

pkg/stanza/fileconsumer/internal/emittest/sink_test.go

@@ -204,7 +204,7 @@ func sinkTest(t *testing.T, opts ...SinkOpt) (*Sink, []emit.Token) {
 	}
 	go func() {
 		for _, c := range testCalls {
-			assert.NoError(t, s.Callback(context.Background(), emit.NewToken(c.Body, c.Attributes)))
+			assert.NoError(t, s.Callback(context.Background(), [][]byte{c.Body}, c.Attributes, 0))
 		}
 	}()
 	return s, testCalls

pkg/stanza/fileconsumer/internal/header/output.go

@@ -30,6 +30,13 @@ func newPipelineOutput(set component.TelemetrySettings) *pipelineOutput {
 	}
 }
 
+func (e *pipelineOutput) ProcessBatch(ctx context.Context, entries []*entry.Entry) error {
+	for i := range entries {
+		_ = e.Process(ctx, entries[i])
+	}
+	return nil
+}
+
 // Drop the entry if logChan is full, in order to avoid this operator blocking.
 // This protects against a case where an operator could return an error, but continue propagating a log entry,
 // leaving an unexpected entry in the output channel.

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

@@ -24,8 +24,9 @@ import (
 )
 
 const (
-	DefaultMaxLogSize  = 1024 * 1024
-	DefaultFlushPeriod = 500 * time.Millisecond
+	DefaultMaxLogSize   = 1024 * 1024
+	DefaultFlushPeriod  = 500 * time.Millisecond
+	DefaultMaxBatchSize = 100
 )
 
 type Factory struct {
@@ -81,6 +82,7 @@ func (f *Factory) NewReaderFromMetadata(file *os.File, m *Metadata) (r *Reader,
 		includeFileRecordNum: f.IncludeFileRecordNumber,
 		compression:          f.Compression,
 		acquireFSLock:        f.AcquireFSLock,
+		maxBatchSize:         DefaultMaxBatchSize,
 		emitFunc:             f.EmitFunc,
 	}
 	r.set.Logger = r.set.Logger.With(zap.String("path", r.fileName))

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

@@ -16,7 +16,6 @@ import (
 	"golang.org/x/text/encoding"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/textutils"
-	"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/fingerprint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/header"
@@ -55,6 +54,7 @@ type Reader struct {
 	includeFileRecordNum   bool
 	compression            string
 	acquireFSLock          bool
+	maxBatchSize           int
 }
 
 // ReadToEnd will read until the end of the file
@@ -188,6 +188,8 @@ func (r *Reader) readContents(ctx context.Context) {
 
 	s := scanner.New(r, r.maxLogSize, bufferSize, r.Offset, r.contentSplitFunc)
 
+	tokenBodies := make([][]byte, r.maxBatchSize)
+	numTokensBatched := 0
 	// Iterate over the contents of the file.
 	for {
 		select {
@@ -203,27 +205,38 @@ func (r *Reader) readContents(ctx context.Context) {
 			} else if r.deleteAtEOF {
 				r.delete()
 			}
+
+			if numTokensBatched > 0 {
+				err := r.emitFunc(ctx, tokenBodies[:numTokensBatched], r.FileAttributes, r.RecordNum)
+				if err != nil {
+					r.set.Logger.Error("failed to emit token", zap.Error(err))
+				}
+				r.Offset = s.Pos()
+			}
 			return
 		}
 
-		token, err := r.decoder.Bytes(s.Bytes())
+		var err error
+		tokenBodies[numTokensBatched], err = r.decoder.Bytes(s.Bytes())
 		if err != nil {
 			r.set.Logger.Error("failed to decode token", zap.Error(err))
 			r.Offset = s.Pos() // move past the bad token or we may be stuck
 			continue
 		}
+		numTokensBatched++
 
 		if r.includeFileRecordNum {
 			r.RecordNum++
-			r.FileAttributes[attrs.LogFileRecordNumber] = r.RecordNum
 		}
 
-		err = r.emitFunc(ctx, emit.NewToken(token, r.FileAttributes))
-		if err != nil {
-			r.set.Logger.Error("failed to process token", zap.Error(err))
+		if r.maxBatchSize > 0 && numTokensBatched >= r.maxBatchSize {
+			err := r.emitFunc(ctx, tokenBodies[:numTokensBatched], r.FileAttributes, r.RecordNum)
+			if err != nil {
+				r.set.Logger.Error("failed to emit token", zap.Error(err))
+			}
+			numTokensBatched = 0
+			r.Offset = s.Pos()
 		}
-
-		r.Offset = s.Pos()
 	}
 }
 

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

@@ -347,10 +347,8 @@ func BenchmarkFileRead(b *testing.B) {
 
 	// Use a long flush period to ensure it does not expire DURING a ReadToEnd
 	counter := atomic.Int64{}
-	f := newTestFactory(b, func(_ context.Context, token emit.Token) error {
-		if len(token.Body) != 0 {
-			counter.Add(1)
-		}
+	f := newTestFactory(b, func(_ context.Context, tokens [][]byte, _ map[string]any, _ int64) error {
+		counter.Add(int64(len(tokens)))
 		return nil
 	})
 	b.ReportAllocs()

pkg/stanza/operator/helper/emitter.go

@@ -99,6 +99,31 @@ func (e *LogEmitter) Stop() error {
 	return nil
 }
 
+// ProcessBatch emits the entries to the consumerFunc
+func (e *LogEmitter) ProcessBatch(ctx context.Context, entries []*entry.Entry) error {
+	if oldBatch := e.appendEntries(entries); len(oldBatch) > 0 {
+		e.consumerFunc(ctx, oldBatch)
+	}
+
+	return nil
+}
+
+// appendEntries appends the entry to the current batch. If maxBatchSize is reached, a new batch will be made, and the old batch
+// (which should be flushed) will be returned
+func (e *LogEmitter) appendEntries(entries []*entry.Entry) []*entry.Entry {
+	e.batchMux.Lock()
+	defer e.batchMux.Unlock()
+
+	e.batch = append(e.batch, entries...)
+	if uint(len(e.batch)) >= e.maxBatchSize {
+		var oldBatch []*entry.Entry
+		oldBatch, e.batch = e.batch, make([]*entry.Entry, 0, e.maxBatchSize)
+		return oldBatch
+	}
+
+	return nil
+}
+
 // Process will emit an entry to the output channel
 func (e *LogEmitter) Process(ctx context.Context, ent *entry.Entry) error {
 	if oldBatch := e.appendEntry(ent); len(oldBatch) > 0 {

pkg/stanza/operator/helper/input.go

@@ -82,6 +82,15 @@ func (i *InputOperator) CanProcess() bool {
 	return false
 }
 
+// ProcessBatch will always return an error if called.
+func (i *InputOperator) ProcessBatch(_ context.Context, _ []*entry.Entry) error {
+	i.Logger().Error("Operator received a batch of entries, but can not process")
+	return errors.NewError(
+		"Operator can not process logs.",
+		"Ensure that operator is not configured to receive logs from other operators",
+	)
+}
+
 // Process will always return an error if called.
 func (i *InputOperator) Process(_ context.Context, _ *entry.Entry) error {
 	i.Logger().Error("Operator received an entry, but can not process")