[chore][fileconsumer] Extract splitter, flusher, multiline, encoding into dedicated package

.chloggen/pkg-stanza-encoding.yaml

@@ -7,7 +7,7 @@ change_type: deprecation
 component: pkg/stanza
 
 # A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
-note: Deprecate encoding related elements of helper pacakge, in favor of new decode package
+note: Deprecate encoding related elements of helper pacakge, in favor of new decoder package
 
 # Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
 issues: [26019]

.chloggen/pkg-stanza-tokenize.yaml

@@ -0,0 +1,40 @@
+# 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 tokenization related elements of helper pacakge, in favor of new tokenize package
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [25914]
+
+# (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: Includes the following deprecations |
+  - Flusher
+  - FlusherConfig
+  - NewFlusherConfig
+  - Multiline
+  - MultilineConfig
+  - NewMultilineConfig
+  - NewLineStartSplitFunc
+  - NewLineEndSplitFunc
+  - NewNewlineSplitFunc
+  - Splitter
+  - SplitterConfig
+  - NewSplitterConfig
+  - SplitNone
+
+# 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/config.go

@@ -20,6 +20,7 @@ import (
 	"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"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/tokenize"
 )
 
 const (
@@ -49,7 +50,7 @@ func NewConfig() *Config {
 		IncludeFileNameResolved: false,
 		IncludeFilePathResolved: false,
 		PollInterval:            200 * time.Millisecond,
-		Splitter:                helper.NewSplitterConfig(),
+		Splitter:                tokenize.NewSplitterConfig(),
 		StartAt:                 "end",
 		FingerprintSize:         fingerprint.DefaultSize,
 		MaxLogSize:              defaultMaxLogSize,
@@ -61,19 +62,19 @@ func NewConfig() *Config {
 // Config is the configuration of a file input operator
 type Config struct {
 	matcher.Criteria        `mapstructure:",squash"`
-	IncludeFileName         bool                  `mapstructure:"include_file_name,omitempty"`
-	IncludeFilePath         bool                  `mapstructure:"include_file_path,omitempty"`
-	IncludeFileNameResolved bool                  `mapstructure:"include_file_name_resolved,omitempty"`
-	IncludeFilePathResolved bool                  `mapstructure:"include_file_path_resolved,omitempty"`
-	PollInterval            time.Duration         `mapstructure:"poll_interval,omitempty"`
-	StartAt                 string                `mapstructure:"start_at,omitempty"`
-	FingerprintSize         helper.ByteSize       `mapstructure:"fingerprint_size,omitempty"`
-	MaxLogSize              helper.ByteSize       `mapstructure:"max_log_size,omitempty"`
-	MaxConcurrentFiles      int                   `mapstructure:"max_concurrent_files,omitempty"`
-	MaxBatches              int                   `mapstructure:"max_batches,omitempty"`
-	DeleteAfterRead         bool                  `mapstructure:"delete_after_read,omitempty"`
-	Splitter                helper.SplitterConfig `mapstructure:",squash,omitempty"`
-	Header                  *HeaderConfig         `mapstructure:"header,omitempty"`
+	IncludeFileName         bool                    `mapstructure:"include_file_name,omitempty"`
+	IncludeFilePath         bool                    `mapstructure:"include_file_path,omitempty"`
+	IncludeFileNameResolved bool                    `mapstructure:"include_file_name_resolved,omitempty"`
+	IncludeFilePathResolved bool                    `mapstructure:"include_file_path_resolved,omitempty"`
+	PollInterval            time.Duration           `mapstructure:"poll_interval,omitempty"`
+	StartAt                 string                  `mapstructure:"start_at,omitempty"`
+	FingerprintSize         helper.ByteSize         `mapstructure:"fingerprint_size,omitempty"`
+	MaxLogSize              helper.ByteSize         `mapstructure:"max_log_size,omitempty"`
+	MaxConcurrentFiles      int                     `mapstructure:"max_concurrent_files,omitempty"`
+	MaxBatches              int                     `mapstructure:"max_batches,omitempty"`
+	DeleteAfterRead         bool                    `mapstructure:"delete_after_read,omitempty"`
+	Splitter                tokenize.SplitterConfig `mapstructure:",squash,omitempty"`
+	Header                  *HeaderConfig           `mapstructure:"header,omitempty"`
 }
 
 type HeaderConfig struct {

pkg/stanza/fileconsumer/config_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/operatortest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/parser/regex"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/tokenize"
 )
 
 func TestUnmarshal(t *testing.T) {
@@ -279,7 +280,7 @@ func TestUnmarshal(t *testing.T) {
 				Name: "multiline_line_start_string",
 				Expect: func() *mockOperatorConfig {
 					cfg := NewConfig()
-					newSplit := helper.NewSplitterConfig()
+					newSplit := tokenize.NewSplitterConfig()
 					newSplit.Multiline.LineStartPattern = "Start"
 					cfg.Splitter = newSplit
 					return newMockOperatorConfig(cfg)
@@ -289,7 +290,7 @@ func TestUnmarshal(t *testing.T) {
 				Name: "multiline_line_start_special",
 				Expect: func() *mockOperatorConfig {
 					cfg := NewConfig()
-					newSplit := helper.NewSplitterConfig()
+					newSplit := tokenize.NewSplitterConfig()
 					newSplit.Multiline.LineStartPattern = "%"
 					cfg.Splitter = newSplit
 					return newMockOperatorConfig(cfg)
@@ -299,7 +300,7 @@ func TestUnmarshal(t *testing.T) {
 				Name: "multiline_line_end_string",
 				Expect: func() *mockOperatorConfig {
 					cfg := NewConfig()
-					newSplit := helper.NewSplitterConfig()
+					newSplit := tokenize.NewSplitterConfig()
 					newSplit.Multiline.LineEndPattern = "Start"
 					cfg.Splitter = newSplit
 					return newMockOperatorConfig(cfg)
@@ -309,7 +310,7 @@ func TestUnmarshal(t *testing.T) {
 				Name: "multiline_line_end_special",
 				Expect: func() *mockOperatorConfig {
 					cfg := NewConfig()
-					newSplit := helper.NewSplitterConfig()
+					newSplit := tokenize.NewSplitterConfig()
 					newSplit.Multiline.LineEndPattern = "%"
 					cfg.Splitter = newSplit
 					return newMockOperatorConfig(cfg)
@@ -451,8 +452,8 @@ func TestBuild(t *testing.T) {
 		{
 			"MultilineConfiguredStartAndEndPatterns",
 			func(f *Config) {
-				f.Splitter = helper.NewSplitterConfig()
-				f.Splitter.Multiline = helper.MultilineConfig{
+				f.Splitter = tokenize.NewSplitterConfig()
+				f.Splitter.Multiline = tokenize.MultilineConfig{
 					LineEndPattern:   "Exists",
 					LineStartPattern: "Exists",
 				}
@@ -463,8 +464,8 @@ func TestBuild(t *testing.T) {
 		{
 			"MultilineConfiguredStartPattern",
 			func(f *Config) {
-				f.Splitter = helper.NewSplitterConfig()
-				f.Splitter.Multiline = helper.MultilineConfig{
+				f.Splitter = tokenize.NewSplitterConfig()
+				f.Splitter.Multiline = tokenize.MultilineConfig{
 					LineStartPattern: "START.*",
 				}
 			},
@@ -474,8 +475,8 @@ func TestBuild(t *testing.T) {
 		{
 			"MultilineConfiguredEndPattern",
 			func(f *Config) {
-				f.Splitter = helper.NewSplitterConfig()
-				f.Splitter.Multiline = helper.MultilineConfig{
+				f.Splitter = tokenize.NewSplitterConfig()
+				f.Splitter.Multiline = tokenize.MultilineConfig{
 					LineEndPattern: "END.*",
 				}
 			},
@@ -493,8 +494,8 @@ func TestBuild(t *testing.T) {
 		{
 			"LineStartAndEnd",
 			func(f *Config) {
-				f.Splitter = helper.NewSplitterConfig()
-				f.Splitter.Multiline = helper.MultilineConfig{
+				f.Splitter = tokenize.NewSplitterConfig()
+				f.Splitter.Multiline = tokenize.MultilineConfig{
 					LineStartPattern: ".*",
 					LineEndPattern:   ".*",
 				}
@@ -505,17 +506,17 @@ func TestBuild(t *testing.T) {
 		{
 			"NoLineStartOrEnd",
 			func(f *Config) {
-				f.Splitter = helper.NewSplitterConfig()
-				f.Splitter.Multiline = helper.MultilineConfig{}
+				f.Splitter = tokenize.NewSplitterConfig()
+				f.Splitter.Multiline = tokenize.MultilineConfig{}
 			},
 			require.NoError,
 			func(t *testing.T, f *Manager) {},
 		},
 		{
 			"InvalidLineStartRegex",
 			func(f *Config) {
-				f.Splitter = helper.NewSplitterConfig()
-				f.Splitter.Multiline = helper.MultilineConfig{
+				f.Splitter = tokenize.NewSplitterConfig()
+				f.Splitter.Multiline = tokenize.MultilineConfig{
 					LineStartPattern: "(",
 				}
 			},
@@ -525,8 +526,8 @@ func TestBuild(t *testing.T) {
 		{
 			"InvalidLineEndRegex",
 			func(f *Config) {
-				f.Splitter = helper.NewSplitterConfig()
-				f.Splitter.Multiline = helper.MultilineConfig{
+				f.Splitter = tokenize.NewSplitterConfig()
+				f.Splitter.Multiline = tokenize.MultilineConfig{
 					LineEndPattern: "(",
 				}
 			},

pkg/stanza/fileconsumer/file_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/matcher"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/tokenize"
 )
 
 func TestCleanStop(t *testing.T) {
@@ -545,7 +546,7 @@ func TestNoNewline(t *testing.T) {
 	tempDir := t.TempDir()
 	cfg := NewConfig().includeDir(tempDir)
 	cfg.StartAt = "beginning"
-	cfg.Splitter = helper.NewSplitterConfig()
+	cfg.Splitter = tokenize.NewSplitterConfig()
 	cfg.Splitter.Flusher.Period = time.Nanosecond
 	operator, emitCalls := buildTestManager(t, cfg)
 

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

@@ -14,8 +14,8 @@ import (
 	"golang.org/x/text/encoding"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/pipeline"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/tokenize"
 )
 
 type Config struct {
@@ -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 := helper.NewNewlineSplitFunc(enc, false, func(b []byte) []byte {
+	splitFunc, err := tokenize.NewNewlineSplitFunc(enc, false, func(b []byte) []byte {
 		return bytes.Trim(b, "\r\n")
 	})
 	if err != nil {

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

@@ -6,18 +6,18 @@ package splitter // import "github.com/open-telemetry/opentelemetry-collector-co
 import (
 	"bufio"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/tokenize"
 )
 
 type customFactory struct {
-	Flusher  helper.FlusherConfig
+	Flusher  tokenize.FlusherConfig
 	Splitter bufio.SplitFunc
 }
 
 var _ Factory = (*customFactory)(nil)
 
 func NewCustomFactory(
-	flusher helper.FlusherConfig,
+	flusher tokenize.FlusherConfig,
 	splitter bufio.SplitFunc) Factory {
 	return &customFactory{
 		Flusher:  flusher,

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

@@ -9,12 +9,12 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/tokenize"
 )
 
 func TestCustomFactory(t *testing.T) {
 	type fields struct {
-		Flusher  helper.FlusherConfig
+		Flusher  tokenize.FlusherConfig
 		Splitter bufio.SplitFunc
 	}
 	type args struct {
@@ -29,7 +29,7 @@ func TestCustomFactory(t *testing.T) {
 		{
 			name: "default configuration",
 			fields: fields{
-				Flusher: helper.NewFlusherConfig(),
+				Flusher: tokenize.NewFlusherConfig(),
 				Splitter: func(data []byte, atEOF bool) (advance int, token []byte, err error) {
 					return len(data), data, nil
 				},

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

@@ -7,16 +7,16 @@ import (
 	"bufio"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/decoder"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/tokenize"
 )
 
 type multilineFactory struct {
-	helper.SplitterConfig
+	tokenize.SplitterConfig
 }
 
 var _ Factory = (*multilineFactory)(nil)
 
-func NewMultilineFactory(splitter helper.SplitterConfig) Factory {
+func NewMultilineFactory(splitter tokenize.SplitterConfig) Factory {
 	return &multilineFactory{
 		SplitterConfig: splitter,
 	}

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

@@ -8,7 +8,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/tokenize"
 )
 
 func TestMultilineBuild(t *testing.T) {
@@ -17,24 +17,24 @@ func TestMultilineBuild(t *testing.T) {
 	}
 	tests := []struct {
 		name           string
-		splitterConfig helper.SplitterConfig
+		splitterConfig tokenize.SplitterConfig
 		args           args
 		wantErr        bool
 	}{
 		{
 			name:           "default configuration",
-			splitterConfig: helper.NewSplitterConfig(),
+			splitterConfig: tokenize.NewSplitterConfig(),
 			args: args{
 				maxLogSize: 1024,
 			},
 			wantErr: false,
 		},
 		{
 			name: "eoncoding error",
-			splitterConfig: helper.SplitterConfig{
+			splitterConfig: tokenize.SplitterConfig{
 				Encoding:  "error",
-				Flusher:   helper.NewFlusherConfig(),
-				Multiline: helper.NewMultilineConfig(),
+				Flusher:   tokenize.NewFlusherConfig(),
+				Multiline: tokenize.NewMultilineConfig(),
 			},
 			args: args{
 				maxLogSize: 1024,
@@ -43,10 +43,10 @@ func TestMultilineBuild(t *testing.T) {
 		},
 		{
 			name: "Multiline  error",
-			splitterConfig: helper.SplitterConfig{
+			splitterConfig: tokenize.SplitterConfig{
 				Encoding: "utf-8",
-				Flusher:  helper.NewFlusherConfig(),
-				Multiline: helper.MultilineConfig{
+				Flusher:  tokenize.NewFlusherConfig(),
+				Multiline: tokenize.MultilineConfig{
 					LineStartPattern: "START",
 					LineEndPattern:   "END",
 				},