[pkg/stanza] Move flush state into reader package

.chloggen/pkg-stanza-metadata-flush.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: deprecation
+
+# 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: Deprecate 'flush.WithPeriod'. Use 'flush.WithFunc' instead.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [27843]
+
+# (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/fileconsumer/benchmark_test.go

@@ -127,6 +127,20 @@ func BenchmarkFileInput(b *testing.B) {
 				return cfg
 			},
 		},
+		{
+			name: "NoFlush",
+			paths: []string{
+				"file0.log",
+			},
+			config: func() *Config {
+				cfg := NewConfig()
+				cfg.Include = []string{
+					"file*.log",
+				}
+				cfg.FlushPeriod = 0
+				return cfg
+			},
+		},
 	}
 
 	for _, bench := range cases {

pkg/stanza/fileconsumer/config.go

@@ -18,7 +18,6 @@ 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/splitter"
 	"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"
@@ -113,23 +112,18 @@ func (c Config) Build(logger *zap.SugaredLogger, emit emit.Callback) (*Manager,
 		trimFunc = c.TrimConfig.Func()
 	}
 
-	// Ensure that splitter is buildable
-	factory := splitter.NewFactory(splitFunc, trimFunc, c.FlushPeriod, int(c.MaxLogSize))
-	return c.buildManager(logger, emit, factory)
+	return c.buildManager(logger, emit, splitFunc, trimFunc)
 }
 
 // BuildWithSplitFunc will build a file input operator with customized splitFunc function
 func (c Config) BuildWithSplitFunc(logger *zap.SugaredLogger, emit emit.Callback, splitFunc bufio.SplitFunc) (*Manager, error) {
 	if err := c.validate(); err != nil {
 		return nil, err
 	}
-
-	// Ensure that splitter is buildable
-	factory := splitter.NewFactory(splitFunc, c.TrimConfig.Func(), c.FlushPeriod, int(c.MaxLogSize))
-	return c.buildManager(logger, emit, factory)
+	return c.buildManager(logger, emit, splitFunc, c.TrimConfig.Func())
 }
 
-func (c Config) buildManager(logger *zap.SugaredLogger, emit emit.Callback, factory splitter.Factory) (*Manager, error) {
+func (c Config) buildManager(logger *zap.SugaredLogger, emit emit.Callback, splitFunc bufio.SplitFunc, trimFunc trim.Func) (*Manager, error) {
 	if emit == nil {
 		return nil, fmt.Errorf("must provide emit function")
 	}
@@ -175,11 +169,13 @@ func (c Config) buildManager(logger *zap.SugaredLogger, emit emit.Callback, fact
 				IncludeFileNameResolved: c.IncludeFileNameResolved,
 				IncludeFilePathResolved: c.IncludeFilePathResolved,
 				DeleteAtEOF:             c.DeleteAfterRead,
+				FlushTimeout:            c.FlushPeriod,
 			},
-			FromBeginning:   startAtBeginning,
-			SplitterFactory: factory,
-			Encoding:        enc,
-			HeaderConfig:    hCfg,
+			FromBeginning: startAtBeginning,
+			Encoding:      enc,
+			SplitFunc:     splitFunc,
+			TrimFunc:      trimFunc,
+			HeaderConfig:  hCfg,
 		},
 		fileMatcher:   fileMatcher,
 		roller:        newRoller(),

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

@@ -8,6 +8,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"time"
 
 	"go.uber.org/zap"
 	"golang.org/x/text/encoding"
@@ -16,53 +17,65 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/decode"
 	"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/splitter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/util"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/flush"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/trim"
 )
 
 type Factory struct {
 	*zap.SugaredLogger
-	Config          *Config
-	FromBeginning   bool
-	SplitterFactory splitter.Factory
-	Encoding        encoding.Encoding
-	HeaderConfig    *header.Config
+	Config        *Config
+	FromBeginning bool
+	Encoding      encoding.Encoding
+	HeaderConfig  *header.Config
+	SplitFunc     bufio.SplitFunc
+	TrimFunc      trim.Func
 }
 
 func (f *Factory) NewReader(file *os.File, fp *fingerprint.Fingerprint) (*Reader, error) {
-	return f.build(file, &Metadata{
+	m := &Metadata{
 		Fingerprint:    fp,
 		FileAttributes: map[string]any{},
-	}, f.SplitterFactory.SplitFunc())
+	}
+	if f.Config.FlushTimeout > 0 {
+		m.FlushState = &flush.State{LastDataChange: time.Now()}
+	}
+	return f.build(file, m)
 }
 
 // copy creates a deep copy of a reader
 func (f *Factory) Copy(old *Reader, newFile *os.File) (*Reader, error) {
-	lineSplitFunc := old.lineSplitFunc
-	if lineSplitFunc == nil {
-		lineSplitFunc = f.SplitterFactory.SplitFunc()
-	}
 	return f.build(newFile, &Metadata{
 		Fingerprint:     old.Fingerprint.Copy(),
 		Offset:          old.Offset,
 		FileAttributes:  util.MapCopy(old.FileAttributes),
 		HeaderFinalized: old.HeaderFinalized,
-	}, lineSplitFunc)
+		FlushState: &flush.State{
+			LastDataChange: old.FlushState.LastDataChange,
+			LastDataLength: old.FlushState.LastDataLength,
+		},
+	})
 }
 
 func (f *Factory) NewFingerprint(file *os.File) (*fingerprint.Fingerprint, error) {
 	return fingerprint.New(file, f.Config.FingerprintSize)
 }
 
-func (f *Factory) build(file *os.File, m *Metadata, lineSplitFunc bufio.SplitFunc) (r *Reader, err error) {
+func (f *Factory) build(file *os.File, m *Metadata) (r *Reader, err error) {
 	r = &Reader{
-		Config:        f.Config,
-		Metadata:      m,
-		file:          file,
-		fileName:      file.Name(),
-		logger:        f.SugaredLogger.With("path", file.Name()),
-		decoder:       decode.New(f.Encoding),
-		lineSplitFunc: lineSplitFunc,
+		Config:   f.Config,
+		Metadata: m,
+		file:     file,
+		fileName: file.Name(),
+		logger:   f.SugaredLogger.With("path", file.Name()),
+		decoder:  decode.New(f.Encoding),
+	}
+
+	if m.FlushState == nil {
+		r.lineSplitFunc = trim.WithFunc(trim.ToLength(f.SplitFunc, f.Config.MaxLogSize), f.TrimFunc)
+	} else {
+		flushFunc := m.FlushState.Func(f.SplitFunc, f.Config.FlushTimeout)
+		r.lineSplitFunc = trim.WithFunc(trim.ToLength(flushFunc, f.Config.MaxLogSize), f.TrimFunc)
 	}
 
 	if !f.FromBeginning {

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

@@ -9,6 +9,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"time"
 
 	"go.uber.org/zap"
 
@@ -18,6 +19,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/scanner"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/flush"
 )
 
 type Config struct {
@@ -29,13 +31,15 @@ type Config struct {
 	IncludeFileNameResolved bool
 	IncludeFilePathResolved bool
 	DeleteAtEOF             bool
+	FlushTimeout            time.Duration
 }
 
 type Metadata struct {
 	Fingerprint     *fingerprint.Fingerprint
 	Offset          int64
 	FileAttributes  map[string]any
 	HeaderFinalized bool
+	FlushState      *flush.State
 }
 
 // Reader manages a single file

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

@@ -5,43 +5,19 @@ package splitter // import "github.com/open-telemetry/opentelemetry-collector-co
 
 import (
 	"bufio"
-	"time"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/flush"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/trim"
 )
 
-type Factory interface {
-	SplitFunc() bufio.SplitFunc
-}
-
-type factory struct {
-	splitFunc   bufio.SplitFunc
-	trimFunc    trim.Func
-	flushPeriod time.Duration
-	maxLength   int
-}
-
-var _ Factory = (*factory)(nil)
-
-func NewFactory(splitFunc bufio.SplitFunc, trimFunc trim.Func, flushPeriod time.Duration, maxLength int) Factory {
-	return &factory{
-		splitFunc:   splitFunc,
-		trimFunc:    trimFunc,
-		flushPeriod: flushPeriod,
-		maxLength:   maxLength,
-	}
-}
-
-// SplitFunc builds a bufio.SplitFunc based on the configuration
-func (f *factory) SplitFunc() bufio.SplitFunc {
-	// First apply the base splitFunc.
-	// If no token is found, we may still flush one based on timing.
-	// If a token is emitted for any reason, we must then apply trim rules.
-	// We must trim to max length _before_ trimming whitespace because otherwise we
-	// cannot properly keep track of the number of bytes to advance.
-	// For instance, if we have advance: 5, token: []byte(" foo "):
-	//   Trimming whitespace first would result in advance: 5, token: []byte("foo")
-	//   Then if we trim to max length of 2, we don't know whether or not to reduce advance.
-	return trim.WithFunc(trim.ToLength(flush.WithPeriod(f.splitFunc, f.flushPeriod), f.maxLength), f.trimFunc)
+// Func builds a bufio.SplitFunc based on the configuration
+// First apply the base splitFunc.
+// If a token is emitted for any reason, we must then apply trim rules.
+// We must trim to max length _before_ trimming whitespace because otherwise we
+// cannot properly keep track of the number of bytes to advance.
+// For instance, if we have advance: 5, token: []byte(" foo "):
+//
+//	Trimming whitespace first would result in advance: 5, token: []byte("foo")
+//	Then if we trim to max length of 2, we don't know whether or not to reduce advance.
+func Func(splitFunc bufio.SplitFunc, maxLength int, trimFunc trim.Func) bufio.SplitFunc {
+	return trim.WithFunc(trim.ToLength(splitFunc, maxLength), trimFunc)
 }

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

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/flush"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/split/splittest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/trim"
 )
@@ -134,7 +135,13 @@ func TestFactorySplitFunc(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
-		factory := NewFactory(tc.baseFunc, tc.trimFunc, tc.flushPeriod, tc.maxLength)
-		t.Run(tc.name, splittest.New(factory.SplitFunc(), tc.input, tc.steps...))
+		var splitFunc bufio.SplitFunc
+		if tc.flushPeriod > 0 {
+			s := &flush.State{LastDataChange: time.Now()}
+			splitFunc = Func(s.Func(tc.baseFunc, tc.flushPeriod), tc.maxLength, tc.trimFunc)
+		} else {
+			splitFunc = Func(tc.baseFunc, tc.maxLength, tc.trimFunc)
+		}
+		t.Run(tc.name, splittest.New(splitFunc, tc.input, tc.steps...))
 	}
 }

pkg/stanza/fileconsumer/reader_test.go

@@ -15,7 +15,6 @@ 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/splitter"
 	"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"
@@ -237,10 +236,12 @@ func testReaderFactory(t *testing.T, sCfg split.Config, maxLogSize int, flushPer
 			FingerprintSize: fingerprint.DefaultSize,
 			MaxLogSize:      maxLogSize,
 			Emit:            testEmitFunc(emitChan),
+			FlushTimeout:    flushPeriod,
 		},
-		FromBeginning:   true,
-		SplitterFactory: splitter.NewFactory(splitFunc, trim.Whitespace, flushPeriod, maxLogSize),
-		Encoding:        enc,
+		FromBeginning: true,
+		Encoding:      enc,
+		SplitFunc:     splitFunc,
+		TrimFunc:      trim.Whitespace,
 	}, emitChan
 }