[pkg/stanza] improve performance (open-telemetry#16027)

* [pkg/stanza] improve performance

.chloggen/pkg-stanza-perf.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# 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: "improve performance"
+
+# One or more tracking issues related to the change
+issues: [16028]
+
+# (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:

pkg/stanza/fileconsumer/reader_test.go

@@ -105,9 +105,7 @@ func testReaderFactory(t *testing.T) (*readerFactory, chan *emitParams) {
 		readerConfig: &readerConfig{
 			fingerprintSize: DefaultFingerprintSize,
 			maxLogSize:      defaultMaxLogSize,
-			emit: func(_ context.Context, attrs *FileAttributes, token []byte) {
-				emitChan <- &emitParams{attrs, token}
-			},
+			emit:            testEmitFunc(emitChan),
 		},
 		fromBeginning: true,
 		splitterFactory: newMultilineSplitterFactory(

pkg/stanza/fileconsumer/util_test.go

@@ -33,6 +33,14 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
 )
 
+func testEmitFunc(emitChan chan *emitParams) EmitFunc {
+	return func(_ context.Context, attrs *FileAttributes, token []byte) {
+		copied := make([]byte, len(token))
+		copy(copied, token)
+		emitChan <- &emitParams{attrs, copied}
+	}
+}
+
 // includeDir is a builder-like helper for quickly setting up a test config
 func (c *Config) includeDir(dir string) *Config {
 	c.Include = append(c.Include, fmt.Sprintf("%s/*", dir))
@@ -56,9 +64,7 @@ func buildTestManager(t *testing.T, cfg *Config) (*Manager, chan *emitParams) {
 }
 
 func buildTestManagerWithEmit(t *testing.T, cfg *Config, emitChan chan *emitParams) *Manager {
-	input, err := cfg.Build(testutil.Logger(t), func(_ context.Context, attrs *FileAttributes, token []byte) {
-		emitChan <- &emitParams{attrs, token}
-	})
+	input, err := cfg.Build(testutil.Logger(t), testEmitFunc(emitChan))
 	require.NoError(t, err)
 	return input
 }

pkg/stanza/operator/helper/encoding.go

@@ -45,27 +45,28 @@ func (c EncodingConfig) Build() (Encoding, error) {
 	}
 
 	return Encoding{
-		Encoding: enc,
+		Encoding:     enc,
+		decodeBuffer: make([]byte, 1<<12),
+		decoder:      enc.NewDecoder(),
 	}, nil
 }
 
 type Encoding struct {
-	Encoding encoding.Encoding
+	Encoding     encoding.Encoding
+	decoder      *encoding.Decoder
+	decodeBuffer []byte
 }
 
-// decode converts the bytes in msgBuf to utf-8 from the configured encoding
+// Decode converts the bytes in msgBuf to utf-8 from the configured encoding
 func (e *Encoding) Decode(msgBuf []byte) ([]byte, error) {
-	decodeBuffer := make([]byte, 1<<12)
-	decoder := e.Encoding.NewDecoder()
-
 	for {
-		decoder.Reset()
-		nDst, _, err := decoder.Transform(decodeBuffer, msgBuf, true)
+		e.decoder.Reset()
+		nDst, _, err := e.decoder.Transform(e.decodeBuffer, msgBuf, true)
 		if err == nil {
-			return decodeBuffer[:nDst], nil
+			return e.decodeBuffer[:nDst], nil
 		}
 		if errors.Is(err, transform.ErrShortDst) {
-			decodeBuffer = make([]byte, len(decodeBuffer)*2)
+			e.decodeBuffer = make([]byte, len(e.decodeBuffer)*2)
 			continue
 		}
 		return nil, fmt.Errorf("transform encoding: %w", err)

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

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