[chore][pkg/stanza] Simplify tokenize package (#26040)

This PR contains several simplifications within and relating to the new tokenization package: - `tokenize.Multiline` now returns a `bufio.splitFunc` without concern for a flush func. A flush func is applied later if necessary. - Remove `tokenize.Splitter`. `SplitterConfig` can now generate a `bufio.SplitFunc` directly. - Unexport `tokenize.Flusher`. `FlusherConfig` can now wrap a `bufio.SplitFunc` directly. (It does still create a struct to maintain state, but then wraps the split func with a receiver that interacts with this state.) - `fileconsumer`'s internal `splitter` package was also simplified. It may make sense to merge this into `tokenize` in a future PR, but this PR is large enough.
open-telemetry · Aug 30, 2023 · f6f1a24 · f6f1a24
1 parent 5645f1e
commit f6f1a24
Show file tree

Hide file tree

Showing 15 changed files with 107 additions and 158 deletions.
diff --git a/pkg/stanza/fileconsumer/config.go b/pkg/stanza/fileconsumer/config.go
@@ -89,8 +89,8 @@ func (c Config) Build(logger *zap.SugaredLogger, emit emit.Callback) (*Manager,
 	}
 
 	// Ensure that splitter is buildable
-	factory := splitter.NewMultilineFactory(c.Splitter)
-	if _, err := factory.Build(int(c.MaxLogSize)); err != nil {
+	factory := splitter.NewMultilineFactory(c.Splitter, int(c.MaxLogSize))
+	if _, err := factory.Build(); err != nil {
 		return nil, err
 	}
 
@@ -109,7 +109,7 @@ func (c Config) BuildWithSplitFunc(logger *zap.SugaredLogger, emit emit.Callback
 
 	// Ensure that splitter is buildable
 	factory := splitter.NewCustomFactory(c.Splitter.Flusher, splitFunc)
-	if _, err := factory.Build(int(c.MaxLogSize)); err != nil {
+	if _, err := factory.Build(); err != nil {
 		return nil, err
 	}
 

diff --git a/pkg/stanza/fileconsumer/internal/header/config.go b/pkg/stanza/fileconsumer/internal/header/config.go
@@ -69,7 +69,7 @@ func NewConfig(matchRegex string, metadataOperators []operator.Config, enc encod
 		return nil, fmt.Errorf("failed to compile `pattern`: %w", err)
 	}
 
-	splitFunc, err := tokenize.NewNewlineSplitFunc(enc, false, func(b []byte) []byte {
+	splitFunc, err := tokenize.NewlineSplitFunc(enc, false, func(b []byte) []byte {
 		return bytes.Trim(b, "\r\n")
 	})
 	if err != nil {

diff --git a/pkg/stanza/fileconsumer/internal/splitter/custom.go b/pkg/stanza/fileconsumer/internal/splitter/custom.go
@@ -10,22 +10,17 @@ import (
 )
 
 type customFactory struct {
-	Flusher  tokenize.FlusherConfig
-	Splitter bufio.SplitFunc
+	flusherCfg tokenize.FlusherConfig
+	splitFunc  bufio.SplitFunc
 }
 
 var _ Factory = (*customFactory)(nil)
 
-func NewCustomFactory(
-	flusher tokenize.FlusherConfig,
-	splitter bufio.SplitFunc) Factory {
-	return &customFactory{
-		Flusher:  flusher,
-		Splitter: splitter,
-	}
+func NewCustomFactory(flusherCfg tokenize.FlusherConfig, splitFunc bufio.SplitFunc) Factory {
+	return &customFactory{flusherCfg: flusherCfg, splitFunc: splitFunc}
 }
 
 // Build builds Multiline Splitter struct
-func (factory *customFactory) Build(_ int) (bufio.SplitFunc, error) {
-	return factory.Flusher.Build().SplitFunc(factory.Splitter), nil
+func (f *customFactory) Build() (bufio.SplitFunc, error) {
+	return f.flusherCfg.Wrap(f.splitFunc), nil
 }
diff --git a/pkg/stanza/fileconsumer/internal/splitter/custom_test.go b/pkg/stanza/fileconsumer/internal/splitter/custom_test.go
@@ -17,13 +17,9 @@ func TestCustomFactory(t *testing.T) {
 		Flusher  tokenize.FlusherConfig
 		Splitter bufio.SplitFunc
 	}
-	type args struct {
-		maxLogSize int
-	}
 	tests := []struct {
 		name    string
 		fields  fields
-		args    args
 		wantErr bool
 	}{
 		{
@@ -34,16 +30,13 @@ func TestCustomFactory(t *testing.T) {
 					return len(data), data, nil
 				},
 			},
-			args: args{
-				maxLogSize: 1024,
-			},
 			wantErr: false,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			factory := NewCustomFactory(tt.fields.Flusher, tt.fields.Splitter)
-			got, err := factory.Build(tt.args.maxLogSize)
+			got, err := factory.Build()
 			if (err != nil) != tt.wantErr {
 				t.Errorf("Build() error = %v, wantErr %v", err, tt.wantErr)
 				return

diff --git a/pkg/stanza/fileconsumer/internal/splitter/factory.go b/pkg/stanza/fileconsumer/internal/splitter/factory.go
@@ -8,5 +8,5 @@ import (
 )
 
 type Factory interface {
-	Build(maxLogSize int) (bufio.SplitFunc, error)
+	Build() (bufio.SplitFunc, error)
 }
diff --git a/pkg/stanza/fileconsumer/internal/splitter/multiline.go b/pkg/stanza/fileconsumer/internal/splitter/multiline.go
@@ -6,32 +6,21 @@ package splitter // import "github.com/open-telemetry/opentelemetry-collector-co
 import (
 	"bufio"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/decoder"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/tokenize"
 )
 
 type multilineFactory struct {
-	tokenize.SplitterConfig
+	splitterCfg tokenize.SplitterConfig
+	maxLogSize  int
 }
 
 var _ Factory = (*multilineFactory)(nil)
 
-func NewMultilineFactory(splitter tokenize.SplitterConfig) Factory {
-	return &multilineFactory{
-		SplitterConfig: splitter,
-	}
+func NewMultilineFactory(splitterCfg tokenize.SplitterConfig, maxLogSize int) Factory {
+	return &multilineFactory{splitterCfg: splitterCfg, maxLogSize: maxLogSize}
 }
 
 // Build builds Multiline Splitter struct
-func (factory *multilineFactory) Build(maxLogSize int) (bufio.SplitFunc, error) {
-	enc, err := decoder.LookupEncoding(factory.Encoding)
-	if err != nil {
-		return nil, err
-	}
-	flusher := factory.Flusher.Build()
-	splitter, err := factory.Multiline.Build(enc, false, factory.PreserveLeadingWhitespaces, factory.PreserveTrailingWhitespaces, flusher, maxLogSize)
-	if err != nil {
-		return nil, err
-	}
-	return splitter, nil
+func (f *multilineFactory) Build() (bufio.SplitFunc, error) {
+	return f.splitterCfg.Build(false, f.maxLogSize)
 }
diff --git a/pkg/stanza/fileconsumer/internal/splitter/multiline_test.go b/pkg/stanza/fileconsumer/internal/splitter/multiline_test.go
@@ -59,8 +59,8 @@ func TestMultilineBuild(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			factory := NewMultilineFactory(tt.splitterConfig)
-			got, err := factory.Build(tt.args.maxLogSize)
+			factory := NewMultilineFactory(tt.splitterConfig, tt.args.maxLogSize)
+			got, err := factory.Build()
 			if (err != nil) != tt.wantErr {
 				t.Errorf("Build() error = %v, wantErr %v", err, tt.wantErr)
 				return

diff --git a/pkg/stanza/fileconsumer/reader_factory.go b/pkg/stanza/fileconsumer/reader_factory.go
@@ -29,7 +29,7 @@ type readerFactory struct {
 }
 
 func (f *readerFactory) newReader(file *os.File, fp *fingerprint.Fingerprint) (*reader, error) {
-	lineSplitFunc, err := f.splitterFactory.Build(f.readerConfig.maxLogSize)
+	lineSplitFunc, err := f.splitterFactory.Build()
 	if err != nil {
 		return nil, err
 	}
@@ -44,7 +44,7 @@ func (f *readerFactory) copy(old *reader, newFile *os.File) (*reader, error) {
 	var err error
 	lineSplitFunc := old.lineSplitFunc
 	if lineSplitFunc == nil {
-		lineSplitFunc, err = f.splitterFactory.Build(f.readerConfig.maxLogSize)
+		lineSplitFunc, err = f.splitterFactory.Build()
 		if err != nil {
 			return nil, err
 		}

diff --git a/pkg/stanza/fileconsumer/reader_test.go b/pkg/stanza/fileconsumer/reader_test.go
@@ -180,7 +180,7 @@ func TestTokenizationTooLongWithLineStartPattern(t *testing.T) {
 		Encoding:  "utf-8",
 		Flusher:   tokenize.NewFlusherConfig(),
 		Multiline: mlc,
-	})
+	}, 15)
 	f.readerConfig.maxLogSize = 15
 
 	temp := openTemp(t, t.TempDir())
@@ -249,7 +249,7 @@ func testReaderFactoryWithSplitter(t *testing.T, splitterConfig tokenize.Splitte
 			emit:            testEmitFunc(emitChan),
 		},
 		fromBeginning:   true,
-		splitterFactory: splitter.NewMultilineFactory(splitterConfig),
+		splitterFactory: splitter.NewMultilineFactory(splitterConfig, defaultMaxLogSize),
 		encoding:        enc,
 	}, emitChan
 }

diff --git a/pkg/stanza/operator/input/tcp/tcp.go b/pkg/stanza/operator/input/tcp/tcp.go
@@ -83,7 +83,7 @@ type BaseConfig struct {
 type MultiLineBuilderFunc func(enc encoding.Encoding) (bufio.SplitFunc, error)
 
 func (c Config) defaultMultilineBuilder(enc encoding.Encoding) (bufio.SplitFunc, error) {
-	splitFunc, err := c.Multiline.Build(enc, true, c.PreserveLeadingWhitespaces, c.PreserveTrailingWhitespaces, nil, int(c.MaxLogSize))
+	splitFunc, err := c.Multiline.Build(enc, true, c.PreserveLeadingWhitespaces, c.PreserveTrailingWhitespaces, int(c.MaxLogSize))
 	if err != nil {
 		return nil, err
 	}

diff --git a/pkg/stanza/operator/input/udp/udp.go b/pkg/stanza/operator/input/udp/udp.go
@@ -91,7 +91,7 @@ func (c Config) Build(logger *zap.SugaredLogger) (operator.Operator, error) {
 	}
 
 	// Build multiline
-	splitFunc, err := c.Multiline.Build(enc, true, c.PreserveLeadingWhitespaces, c.PreserveTrailingWhitespaces, nil, MaxUDPSize)
+	splitFunc, err := c.Multiline.Build(enc, true, c.PreserveLeadingWhitespaces, c.PreserveTrailingWhitespaces, MaxUDPSize)
 	if err != nil {
 		return nil, err
 	}

diff --git a/pkg/stanza/tokenize/flusher.go b/pkg/stanza/tokenize/flusher.go
@@ -8,6 +8,8 @@ import (
 	"time"
 )
 
+const DefaultFlushPeriod = 500 * time.Millisecond
+
 // FlusherConfig is a configuration of Flusher helper
 type FlusherConfig struct {
 	Period time.Duration `mapstructure:"force_flush_period"`
@@ -17,21 +19,22 @@ type FlusherConfig struct {
 func NewFlusherConfig() FlusherConfig {
 	return FlusherConfig{
 		// Empty or `0s` means that we will never force flush
-		Period: time.Millisecond * 500,
+		Period: DefaultFlushPeriod,
 	}
 }
 
-// Build creates Flusher from configuration
-func (c *FlusherConfig) Build() *Flusher {
-	return &Flusher{
+// Wrap a bufio.SplitFunc with a flusher
+func (c *FlusherConfig) Wrap(splitFunc bufio.SplitFunc) bufio.SplitFunc {
+	f := &flusher{
 		lastDataChange:     time.Now(),
 		forcePeriod:        c.Period,
 		previousDataLength: 0,
 	}
+	return f.splitFunc(splitFunc)
 }
 
-// Flusher keeps information about flush state
-type Flusher struct {
+// flusher keeps information about flush state
+type flusher struct {
 	// forcePeriod defines time from last flush which should pass before setting force to true.
 	// Never forces if forcePeriod is set to 0
 	forcePeriod time.Duration
@@ -45,7 +48,7 @@ type Flusher struct {
 	previousDataLength int
 }
 
-func (f *Flusher) UpdateDataChangeTime(length int) {
+func (f *flusher) updateDataChangeTime(length int) {
 	// Skip if length is greater than 0 and didn't changed
 	if length > 0 && length == f.previousDataLength {
 		return
@@ -58,17 +61,17 @@ func (f *Flusher) UpdateDataChangeTime(length int) {
 }
 
 // Flushed reset data length
-func (f *Flusher) Flushed() {
-	f.UpdateDataChangeTime(0)
+func (f *flusher) flushed() {
+	f.updateDataChangeTime(0)
 }
 
 // ShouldFlush returns true if data should be forcefully flushed
-func (f *Flusher) ShouldFlush() bool {
+func (f *flusher) shouldFlush() bool {
 	// Returns true if there is f.forcePeriod after f.lastDataChange and data length is greater than 0
 	return f.forcePeriod > 0 && time.Since(f.lastDataChange) > f.forcePeriod && f.previousDataLength > 0
 }
 
-func (f *Flusher) SplitFunc(splitFunc bufio.SplitFunc) bufio.SplitFunc {
+func (f *flusher) splitFunc(splitFunc bufio.SplitFunc) bufio.SplitFunc {
 	return func(data []byte, atEOF bool) (advance int, token []byte, err error) {
 		advance, token, err = splitFunc(data, atEOF)
 
@@ -80,21 +83,21 @@ func (f *Flusher) SplitFunc(splitFunc bufio.SplitFunc) bufio.SplitFunc {
 		// Return token
 		if token != nil {
 			// Inform flusher that we just flushed
-			f.Flushed()
+			f.flushed()
 			return
 		}
 
 		// If there is no token, force flush eventually
-		if f.ShouldFlush() {
+		if f.shouldFlush() {
 			// Inform flusher that we just flushed
-			f.Flushed()
+			f.flushed()
 			token = trimWhitespacesFunc(data)
 			advance = len(data)
 			return
 		}
 
 		// Inform flusher that we didn't flushed
-		f.UpdateDataChangeTime(len(data))
+		f.updateDataChangeTime(len(data))
 		return
 	}
 }