[chore][pkg/stanza] Simplify tokenize package (open-telemetry#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.
ClickHouse · 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
  }
 }