[pkg/stanza] Simplify fingerprint updating (#31251)

Depends on #31298

Fixes #22936

This PR changes the way readers update their fingerprints.

Currently, when `reader.ReadToEnd` is called, it creates a scanner and
passes itself (the reader) in as the `io.Reader` so that a custom
implementation of `Read` will be used by the scanner. Each time the
scanner calls `Read`, we try to perform appropriate reasoning about
whether the data we've read should be appended to the fingerprint. The
problem is that the correct positioning of the bytes buffer is in some
rare cases different than the file's "offset", as we track it. See
example
[here](#22937 (comment)).

There appear to be two ways to solve this. A "simple" solution is to
independently determine the file handle's current offset with a clever
use of `Seek`, ([suggested
here](https://stackoverflow.com/a/10901436/3511338). Although this does
appear to work, it leaves open the possibility that the fingerprint is
corrupted because _if the file was truncated_, we may be updating the
fingerprint with incorrect information.

The other solution, proposed in this PR, simply has the custom `Read`
function set a flag to indicate that the fingerprint _should_ be
updated. Then, just before returning from `ReadToEnd`, we create an
entirely new fingerprint. This has the advantage of not having to manage
any kind of append operations, but also allows the the opportunity to
independently check that the fingerprint has not been altered by
truncation.

Benchmarks appear to show all three solutions are close in performance.

Author: djaglowski

.chloggen/pkg-stanza-reader-offset-update.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: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: receiver/filelog
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix issue where file fingerprint could be corrupted while reading.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [22936]
+
+# (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: []

pkg/stanza/fileconsumer/config.go

@@ -20,6 +20,7 @@ import (
 	"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/fileconsumer/internal/reader"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/scanner"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/matcher"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
@@ -151,18 +152,19 @@ func (c Config) buildManager(logger *zap.SugaredLogger, emit emit.Callback, spli
 	}
 
 	readerFactory := reader.Factory{
-		SugaredLogger:   logger.With("component", "fileconsumer"),
-		FromBeginning:   startAtBeginning,
-		FingerprintSize: int(c.FingerprintSize),
-		MaxLogSize:      int(c.MaxLogSize),
-		Encoding:        enc,
-		SplitFunc:       splitFunc,
-		TrimFunc:        trimFunc,
-		FlushTimeout:    c.FlushPeriod,
-		EmitFunc:        emit,
-		Attributes:      c.Resolver,
-		HeaderConfig:    hCfg,
-		DeleteAtEOF:     c.DeleteAfterRead,
+		SugaredLogger:     logger.With("component", "fileconsumer"),
+		FromBeginning:     startAtBeginning,
+		FingerprintSize:   int(c.FingerprintSize),
+		InitialBufferSize: scanner.DefaultBufferSize,
+		MaxLogSize:        int(c.MaxLogSize),
+		Encoding:          enc,
+		SplitFunc:         splitFunc,
+		TrimFunc:          trimFunc,
+		FlushTimeout:      c.FlushPeriod,
+		EmitFunc:          emit,
+		Attributes:        c.Resolver,
+		HeaderConfig:      hCfg,
+		DeleteAtEOF:       c.DeleteAfterRead,
 	}
 	knownFiles := make([]*fileset.Fileset[*reader.Metadata], 3)
 	for i := 0; i < len(knownFiles); i++ {

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

@@ -28,17 +28,18 @@ const (
 
 type Factory struct {
 	*zap.SugaredLogger
-	HeaderConfig    *header.Config
-	FromBeginning   bool
-	FingerprintSize int
-	MaxLogSize      int
-	Encoding        encoding.Encoding
-	SplitFunc       bufio.SplitFunc
-	TrimFunc        trim.Func
-	FlushTimeout    time.Duration
-	EmitFunc        emit.Callback
-	Attributes      attrs.Resolver
-	DeleteAtEOF     bool
+	HeaderConfig      *header.Config
+	FromBeginning     bool
+	FingerprintSize   int
+	InitialBufferSize int
+	MaxLogSize        int
+	Encoding          encoding.Encoding
+	SplitFunc         bufio.SplitFunc
+	TrimFunc          trim.Func
+	FlushTimeout      time.Duration
+	EmitFunc          emit.Callback
+	Attributes        attrs.Resolver
+	DeleteAtEOF       bool
 }
 
 func (f *Factory) NewFingerprint(file *os.File) (*fingerprint.Fingerprint, error) {
@@ -58,16 +59,21 @@ func (f *Factory) NewReader(file *os.File, fp *fingerprint.Fingerprint) (*Reader
 }
 
 func (f *Factory) NewReaderFromMetadata(file *os.File, m *Metadata) (r *Reader, err error) {
+	// Trim the fingerprint if user has reconfigured fingerprint_size
+	if len(m.Fingerprint.FirstBytes) > f.FingerprintSize {
+		m.Fingerprint.FirstBytes = m.Fingerprint.FirstBytes[:f.FingerprintSize]
+	}
 	r = &Reader{
-		Metadata:        m,
-		logger:          f.SugaredLogger.With("path", file.Name()),
-		file:            file,
-		fileName:        file.Name(),
-		fingerprintSize: f.FingerprintSize,
-		maxLogSize:      f.MaxLogSize,
-		decoder:         decode.New(f.Encoding),
-		lineSplitFunc:   f.SplitFunc,
-		deleteAtEOF:     f.DeleteAtEOF,
+		Metadata:          m,
+		logger:            f.SugaredLogger.With("path", file.Name()),
+		file:              file,
+		fileName:          file.Name(),
+		fingerprintSize:   f.FingerprintSize,
+		initialBufferSize: f.InitialBufferSize,
+		maxLogSize:        f.MaxLogSize,
+		decoder:           decode.New(f.Encoding),
+		lineSplitFunc:     f.SplitFunc,
+		deleteAtEOF:       f.DeleteAtEOF,
 	}
 
 	if !f.FromBeginning {

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

@@ -16,6 +16,7 @@ import (
 	"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/split"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/trim"
@@ -28,13 +29,14 @@ const (
 
 func testFactory(t *testing.T, opts ...testFactoryOpt) (*Factory, *emittest.Sink) {
 	cfg := &testFactoryCfg{
-		fromBeginning:      true,
-		fingerprintSize:    fingerprint.DefaultSize,
-		maxLogSize:         defaultMaxLogSize,
-		encoding:           unicode.UTF8,
-		trimFunc:           trim.Whitespace,
-		flushPeriod:        defaultFlushPeriod,
-		sinkCallBufferSize: 100,
+		fromBeginning:     true,
+		fingerprintSize:   fingerprint.DefaultSize,
+		initialBufferSize: scanner.DefaultBufferSize,
+		maxLogSize:        defaultMaxLogSize,
+		encoding:          unicode.UTF8,
+		trimFunc:          trim.Whitespace,
+		flushPeriod:       defaultFlushPeriod,
+		sinkChanSize:      100,
 		attributes: attrs.Resolver{
 			IncludeFileName: true,
 		},
@@ -46,33 +48,35 @@ func testFactory(t *testing.T, opts ...testFactoryOpt) (*Factory, *emittest.Sink
 	splitFunc, err := cfg.splitCfg.Func(cfg.encoding, false, cfg.maxLogSize)
 	require.NoError(t, err)
 
-	sink := emittest.NewSink(emittest.WithCallBuffer(cfg.sinkCallBufferSize))
+	sink := emittest.NewSink(emittest.WithCallBuffer(cfg.sinkChanSize))
 	return &Factory{
-		SugaredLogger:   testutil.Logger(t),
-		FromBeginning:   cfg.fromBeginning,
-		FingerprintSize: cfg.fingerprintSize,
-		MaxLogSize:      cfg.maxLogSize,
-		Encoding:        cfg.encoding,
-		SplitFunc:       splitFunc,
-		TrimFunc:        cfg.trimFunc,
-		FlushTimeout:    cfg.flushPeriod,
-		EmitFunc:        sink.Callback,
-		Attributes:      cfg.attributes,
+		SugaredLogger:     testutil.Logger(t),
+		FromBeginning:     cfg.fromBeginning,
+		FingerprintSize:   cfg.fingerprintSize,
+		InitialBufferSize: cfg.initialBufferSize,
+		MaxLogSize:        cfg.maxLogSize,
+		Encoding:          cfg.encoding,
+		SplitFunc:         splitFunc,
+		TrimFunc:          cfg.trimFunc,
+		FlushTimeout:      cfg.flushPeriod,
+		EmitFunc:          sink.Callback,
+		Attributes:        cfg.attributes,
 	}, sink
 }
 
 type testFactoryOpt func(*testFactoryCfg)
 
 type testFactoryCfg struct {
-	fromBeginning      bool
-	fingerprintSize    int
-	maxLogSize         int
-	encoding           encoding.Encoding
-	splitCfg           split.Config
-	trimFunc           trim.Func
-	flushPeriod        time.Duration
-	sinkCallBufferSize int
-	attributes         attrs.Resolver
+	fromBeginning     bool
+	fingerprintSize   int
+	initialBufferSize int
+	maxLogSize        int
+	encoding          encoding.Encoding
+	splitCfg          split.Config
+	trimFunc          trim.Func
+	flushPeriod       time.Duration
+	sinkChanSize      int
+	attributes        attrs.Resolver
 }
 
 func withFingerprintSize(size int) testFactoryOpt {
@@ -87,9 +91,15 @@ func withSplitConfig(cfg split.Config) testFactoryOpt {
 	}
 }
 
-func withMaxLogSize(maxLogSize int) testFactoryOpt {
+func withInitialBufferSize(size int) testFactoryOpt {
 	return func(c *testFactoryCfg) {
-		c.maxLogSize = maxLogSize
+		c.initialBufferSize = size
+	}
+}
+
+func withMaxLogSize(size int) testFactoryOpt {
+	return func(c *testFactoryCfg) {
+		c.maxLogSize = size
 	}
 }
 
@@ -99,9 +109,9 @@ func withFlushPeriod(flushPeriod time.Duration) testFactoryOpt {
 	}
 }
 
-func withSinkBufferSize(n int) testFactoryOpt {
+func withSinkChanSize(n int) testFactoryOpt {
 	return func(c *testFactoryCfg) {
-		c.sinkCallBufferSize = n
+		c.sinkChanSize = n
 	}
 }