[pkg/stanza] Extract Encoding from SplitterConfig (#26511)

.chloggen/pkg-stanza-encoding-2.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: breaking
+
+# 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: Move tokenize.SplitterConfig.Encoding to fileconsumer.Config.Encoding
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [26511]
+
+# (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/config.go

@@ -26,6 +26,7 @@ import (
 const (
 	defaultMaxLogSize         = 1024 * 1024
 	defaultMaxConcurrentFiles = 1024
+	defaultEncoding           = "utf-8"
 )
 
 var allowFileDeletion = featuregate.GlobalRegistry().MustRegister(
@@ -51,6 +52,7 @@ func NewConfig() *Config {
 		IncludeFilePathResolved: false,
 		PollInterval:            200 * time.Millisecond,
 		Splitter:                tokenize.NewSplitterConfig(),
+		Encoding:                defaultEncoding,
 		StartAt:                 "end",
 		FingerprintSize:         fingerprint.DefaultSize,
 		MaxLogSize:              defaultMaxLogSize,
@@ -74,6 +76,7 @@ type Config struct {
 	MaxBatches              int                     `mapstructure:"max_batches,omitempty"`
 	DeleteAfterRead         bool                    `mapstructure:"delete_after_read,omitempty"`
 	Splitter                tokenize.SplitterConfig `mapstructure:",squash,omitempty"`
+	Encoding                string                  `mapstructure:"encoding,omitempty"`
 	Header                  *HeaderConfig           `mapstructure:"header,omitempty"`
 }
 
@@ -88,8 +91,13 @@ func (c Config) Build(logger *zap.SugaredLogger, emit emit.Callback) (*Manager,
 		return nil, err
 	}
 
+	enc, err := decode.LookupEncoding(c.Encoding)
+	if err != nil {
+		return nil, err
+	}
+
 	// Ensure that splitter is buildable
-	factory := splitter.NewMultilineFactory(c.Splitter, int(c.MaxLogSize))
+	factory := splitter.NewMultilineFactory(c.Splitter, enc, int(c.MaxLogSize))
 	if _, err := factory.Build(); err != nil {
 		return nil, err
 	}
@@ -130,13 +138,13 @@ func (c Config) buildManager(logger *zap.SugaredLogger, emit emit.Callback, fact
 		return nil, fmt.Errorf("invalid start_at location '%s'", c.StartAt)
 	}
 
+	enc, err := decode.LookupEncoding(c.Encoding)
+	if err != nil {
+		return nil, fmt.Errorf("failed to find encoding: %w", err)
+	}
+
 	var hCfg *header.Config
 	if c.Header != nil {
-		enc, err := decode.LookupEncoding(c.Splitter.Encoding)
-		if err != nil {
-			return nil, fmt.Errorf("failed to create encoding: %w", err)
-		}
-
 		hCfg, err = header.NewConfig(c.Header.Pattern, c.Header.MetadataOperators, enc)
 		if err != nil {
 			return nil, fmt.Errorf("failed to build header config: %w", err)
@@ -148,11 +156,6 @@ func (c Config) buildManager(logger *zap.SugaredLogger, emit emit.Callback, fact
 		return nil, err
 	}
 
-	enc, err := decode.LookupEncoding(c.Splitter.Encoding)
-	if err != nil {
-		return nil, err
-	}
-
 	return &Manager{
 		SugaredLogger: logger.With("component", "fileconsumer"),
 		cancel:        func() {},
@@ -220,7 +223,7 @@ func (c Config) validate() error {
 		return errors.New("`max_batches` must not be negative")
 	}
 
-	enc, err := decode.LookupEncoding(c.Splitter.Encoding)
+	enc, err := decode.LookupEncoding(c.Encoding)
 	if err != nil {
 		return err
 	}

pkg/stanza/fileconsumer/config_test.go

@@ -368,15 +368,15 @@ func TestUnmarshal(t *testing.T) {
 				Name: "encoding_lower",
 				Expect: func() *mockOperatorConfig {
 					cfg := NewConfig()
-					cfg.Splitter.Encoding = "utf-16le"
+					cfg.Encoding = "utf-16le"
 					return newMockOperatorConfig(cfg)
 				}(),
 			},
 			{
 				Name: "encoding_upper",
 				Expect: func() *mockOperatorConfig {
 					cfg := NewConfig()
-					cfg.Splitter.Encoding = "UTF-16lE"
+					cfg.Encoding = "UTF-16lE"
 					return newMockOperatorConfig(cfg)
 				}(),
 			},
@@ -486,7 +486,7 @@ func TestBuild(t *testing.T) {
 		{
 			"InvalidEncoding",
 			func(f *Config) {
-				f.Splitter.Encoding = "UTF-3233"
+				f.Encoding = "UTF-3233"
 			},
 			require.Error,
 			nil,
@@ -681,7 +681,7 @@ func TestBuildWithSplitFunc(t *testing.T) {
 		{
 			"InvalidEncoding",
 			func(f *Config) {
-				f.Splitter.Encoding = "UTF-3233"
+				f.Encoding = "UTF-3233"
 			},
 			require.Error,
 			nil,

pkg/stanza/fileconsumer/file_test.go

@@ -337,7 +337,7 @@ func TestReadUsingNopEncoding(t *testing.T) {
 			cfg := NewConfig().includeDir(tempDir)
 			cfg.StartAt = "beginning"
 			cfg.MaxLogSize = 8
-			cfg.Splitter.Encoding = "nop"
+			cfg.Encoding = "nop"
 			operator, emitCalls := buildTestManager(t, cfg)
 
 			// Create a file, then start
@@ -421,7 +421,7 @@ func TestNopEncodingDifferentLogSizes(t *testing.T) {
 			cfg := NewConfig().includeDir(tempDir)
 			cfg.StartAt = "beginning"
 			cfg.MaxLogSize = tc.maxLogSize
-			cfg.Splitter.Encoding = "nop"
+			cfg.Encoding = "nop"
 			operator, emitCalls := buildTestManager(t, cfg)
 
 			// Create a file, then start
@@ -1289,7 +1289,7 @@ func TestEncodings(t *testing.T) {
 			tempDir := t.TempDir()
 			cfg := NewConfig().includeDir(tempDir)
 			cfg.StartAt = "beginning"
-			cfg.Splitter.Encoding = tc.encoding
+			cfg.Encoding = tc.encoding
 			operator, emitCalls := buildTestManager(t, cfg)
 
 			// Populate the file

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

@@ -6,21 +6,32 @@ package splitter // import "github.com/open-telemetry/opentelemetry-collector-co
 import (
 	"bufio"
 
+	"golang.org/x/text/encoding"
+
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/tokenize"
 )
 
 type multilineFactory struct {
 	splitterCfg tokenize.SplitterConfig
+	encoding    encoding.Encoding
 	maxLogSize  int
 }
 
 var _ Factory = (*multilineFactory)(nil)
 
-func NewMultilineFactory(splitterCfg tokenize.SplitterConfig, maxLogSize int) Factory {
-	return &multilineFactory{splitterCfg: splitterCfg, maxLogSize: maxLogSize}
+func NewMultilineFactory(
+	splitterCfg tokenize.SplitterConfig,
+	encoding encoding.Encoding,
+	maxLogSize int,
+) Factory {
+	return &multilineFactory{
+		splitterCfg: splitterCfg,
+		encoding:    encoding,
+		maxLogSize:  maxLogSize,
+	}
 }
 
 // Build builds Multiline Splitter struct
 func (f *multilineFactory) Build() (bufio.SplitFunc, error) {
-	return f.splitterCfg.Build(false, f.maxLogSize)
+	return f.splitterCfg.Build(f.encoding, false, f.maxLogSize)
 }

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

@@ -7,59 +7,44 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"golang.org/x/text/encoding"
+	"golang.org/x/text/encoding/unicode"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/tokenize"
 )
 
 func TestMultilineBuild(t *testing.T) {
-	type args struct {
-		maxLogSize int
-	}
 	tests := []struct {
 		name           string
 		splitterConfig tokenize.SplitterConfig
-		args           args
+		encoding       encoding.Encoding
+		maxLogSize     int
 		wantErr        bool
 	}{
 		{
 			name:           "default configuration",
 			splitterConfig: tokenize.NewSplitterConfig(),
-			args: args{
-				maxLogSize: 1024,
-			},
-			wantErr: false,
-		},
-		{
-			name: "eoncoding error",
-			splitterConfig: tokenize.SplitterConfig{
-				Encoding:  "error",
-				Flusher:   tokenize.NewFlusherConfig(),
-				Multiline: tokenize.NewMultilineConfig(),
-			},
-			args: args{
-				maxLogSize: 1024,
-			},
-			wantErr: true,
+			encoding:       unicode.UTF8,
+			maxLogSize:     1024,
+			wantErr:        false,
 		},
 		{
 			name: "Multiline  error",
 			splitterConfig: tokenize.SplitterConfig{
-				Encoding: "utf-8",
-				Flusher:  tokenize.NewFlusherConfig(),
+				Flusher: tokenize.NewFlusherConfig(),
 				Multiline: tokenize.MultilineConfig{
 					LineStartPattern: "START",
 					LineEndPattern:   "END",
 				},
 			},
-			args: args{
-				maxLogSize: 1024,
-			},
-			wantErr: true,
+			encoding:   unicode.UTF8,
+			maxLogSize: 1024,
+			wantErr:    true,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			factory := NewMultilineFactory(tt.splitterConfig, tt.args.maxLogSize)
+			factory := NewMultilineFactory(tt.splitterConfig, tt.encoding, tt.maxLogSize)
 			got, err := factory.Build()
 			if (err != nil) != tt.wantErr {
 				t.Errorf("Build() error = %v, wantErr %v", err, tt.wantErr)

pkg/stanza/fileconsumer/reader_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/text/encoding/unicode"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/decode"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
@@ -177,10 +178,9 @@ func TestTokenizationTooLongWithLineStartPattern(t *testing.T) {
 	mlc := tokenize.NewMultilineConfig()
 	mlc.LineStartPattern = `\d+-\d+-\d+`
 	f.splitterFactory = splitter.NewMultilineFactory(tokenize.SplitterConfig{
-		Encoding:  "utf-8",
 		Flusher:   tokenize.NewFlusherConfig(),
 		Multiline: mlc,
-	}, 15)
+	}, unicode.UTF8, 15)
 	f.readerConfig.maxLogSize = 15
 
 	temp := openTemp(t, t.TempDir())
@@ -239,7 +239,7 @@ func testReaderFactory(t *testing.T) (*readerFactory, chan *emitParams) {
 
 func testReaderFactoryWithSplitter(t *testing.T, splitterConfig tokenize.SplitterConfig) (*readerFactory, chan *emitParams) {
 	emitChan := make(chan *emitParams, 100)
-	enc, err := decode.LookupEncoding(splitterConfig.Encoding)
+	enc, err := decode.LookupEncoding(defaultEncoding)
 	require.NoError(t, err)
 	return &readerFactory{
 		SugaredLogger: testutil.Logger(t),
@@ -249,7 +249,7 @@ func testReaderFactoryWithSplitter(t *testing.T, splitterConfig tokenize.Splitte
 			emit:            testEmitFunc(emitChan),
 		},
 		fromBeginning:   true,
-		splitterFactory: splitter.NewMultilineFactory(splitterConfig, defaultMaxLogSize),
+		splitterFactory: splitter.NewMultilineFactory(splitterConfig, enc, defaultMaxLogSize),
 		encoding:        enc,
 	}, emitChan
 }

pkg/stanza/operator/input/file/config.go

@@ -47,7 +47,7 @@ func (c Config) Build(logger *zap.SugaredLogger) (operator.Operator, error) {
 	var toBody toBodyFunc = func(token []byte) interface{} {
 		return string(token)
 	}
-	if decode.IsNop(c.Config.Splitter.Encoding) {
+	if decode.IsNop(c.Config.Encoding) {
 		toBody = func(token []byte) interface{} {
 			copied := make([]byte, len(token))
 			copy(copied, token)

pkg/stanza/operator/input/file/config_test.go

@@ -413,7 +413,7 @@ func TestUnmarshal(t *testing.T) {
 				ExpectErr: false,
 				Expect: func() *Config {
 					cfg := NewConfig()
-					cfg.Splitter.Encoding = "utf-16le"
+					cfg.Encoding = "utf-16le"
 					return cfg
 				}(),
 			},
@@ -422,7 +422,7 @@ func TestUnmarshal(t *testing.T) {
 				ExpectErr: false,
 				Expect: func() *Config {
 					cfg := NewConfig()
-					cfg.Splitter.Encoding = "UTF-16lE"
+					cfg.Encoding = "UTF-16lE"
 					return cfg
 				}(),
 			},
@@ -510,7 +510,7 @@ func TestBuild(t *testing.T) {
 		{
 			"InvalidEncoding",
 			func(f *Config) {
-				f.Splitter.Encoding = "UTF-3233"
+				f.Encoding = "UTF-3233"
 			},
 			require.Error,
 			nil,

pkg/stanza/operator/input/file/file_test.go

@@ -133,7 +133,7 @@ func TestReadUsingNopEncoding(t *testing.T) {
 		t.Run(tc.testName, func(t *testing.T) {
 			operator, logReceived, tempDir := newTestFileOperator(t, func(cfg *Config) {
 				cfg.MaxLogSize = 8
-				cfg.Splitter.Encoding = "nop"
+				cfg.Encoding = "nop"
 			})
 			// Create a file, then start
 			temp := openTemp(t, tempDir)

pkg/stanza/tokenize/splitter.go

@@ -6,13 +6,13 @@ package tokenize // import "github.com/open-telemetry/opentelemetry-collector-co
 import (
 	"bufio"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/decode"
+	"golang.org/x/text/encoding"
+
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/trim"
 )
 
 // SplitterConfig consolidates MultilineConfig and FlusherConfig
 type SplitterConfig struct {
-	Encoding         string          `mapstructure:"encoding,omitempty"`
 	Flusher          FlusherConfig   `mapstructure:",squash,omitempty"`
 	Multiline        MultilineConfig `mapstructure:"multiline,omitempty"`
 	PreserveLeading  bool            `mapstructure:"preserve_leading_whitespaces,omitempty"`
@@ -22,18 +22,13 @@ type SplitterConfig struct {
 // NewSplitterConfig returns default SplitterConfig
 func NewSplitterConfig() SplitterConfig {
 	return SplitterConfig{
-		Encoding:  "utf-8",
 		Multiline: NewMultilineConfig(),
 		Flusher:   FlusherConfig{Period: DefaultFlushPeriod},
 	}
 }
 
 // Build builds bufio.SplitFunc based on the config
-func (c *SplitterConfig) Build(flushAtEOF bool, maxLogSize int) (bufio.SplitFunc, error) {
-	enc, err := decode.LookupEncoding(c.Encoding)
-	if err != nil {
-		return nil, err
-	}
+func (c *SplitterConfig) Build(enc encoding.Encoding, flushAtEOF bool, maxLogSize int) (bufio.SplitFunc, error) {
 	trimFunc := trim.Whitespace(c.PreserveLeading, c.PreserveTrailing)
 	splitFunc, err := c.Multiline.Build(enc, flushAtEOF, maxLogSize, trimFunc)
 	if err != nil {