[stanza] Remove unnecessary slice allocation to track errors (even nil) (open-telemetry#39367)

The most inefficient usages are in ProcessBatch and ProcessBatchWith,
where we allocate a slice of errors even if return value is nil.

Changed all usages to make sure we don't repeat that mistake and
establish a healthier pattern.

Signed-off-by: Bogdan Drutu <[email protected]>

Author: bogdandrutu

.chloggen/stanza-error.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: 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: Remove unnecessary slice allocation to track errors (even nil)
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39367]
+
+# (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: []

pkg/stanza/fileconsumer/internal/checkpoint/checkpoint.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 
 	"go.opentelemetry.io/collector/extension/xextension/storage"
+	"go.uber.org/multierr"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/reader"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
@@ -32,19 +33,19 @@ func SaveKey(ctx context.Context, persister operator.Persister, rmds []*reader.M
 		return fmt.Errorf("encode num files: %w", err)
 	}
 
-	var errs []error
+	var errs error
 	// Encode each known file
 	for _, rmd := range rmds {
 		if err := enc.Encode(rmd); err != nil {
-			errs = append(errs, fmt.Errorf("encode metadata: %w", err))
+			errs = multierr.Append(errs, fmt.Errorf("encode metadata: %w", err))
 		}
 	}
 	ops = append(ops, storage.SetOperation(key, buf.Bytes()))
 	if err := persister.Batch(ctx, ops...); err != nil {
-		errs = append(errs, fmt.Errorf("persist known files: %w", err))
+		errs = multierr.Append(errs, fmt.Errorf("persist known files: %w", err))
 	}
 
-	return errors.Join(errs...)
+	return errs
 }
 
 // Load loads the most recent set of files to the database
@@ -71,7 +72,7 @@ func LoadKey(ctx context.Context, persister operator.Persister, key string) ([]*
 	}
 
 	// Decode each of the known files
-	var errs []error
+	var errs error
 	rmds := make([]*reader.Metadata, 0, knownFileCount)
 	for i := 0; i < knownFileCount; i++ {
 		rmd := new(reader.Metadata)
@@ -92,13 +93,13 @@ func LoadKey(ctx context.Context, persister operator.Persister, key string) ([]*
 				}
 				delete(rmd.FileAttributes, "HeaderAttributes")
 			default:
-				errs = append(errs, errors.New("migrate header attributes: unexpected format"))
+				errs = multierr.Append(errs, errors.New("migrate header attributes: unexpected format"))
 			}
 		}
 
 		// This reader won't be used for anything other than metadata reference, so just wrap the metadata
 		rmds = append(rmds, rmd)
 	}
 
-	return rmds, errors.Join(errs...)
+	return rmds, errs
 }

pkg/stanza/fileconsumer/matcher/internal/finder/finder.go

@@ -4,12 +4,12 @@
 package finder // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/matcher/internal/finder"
 
 import (
-	"errors"
 	"fmt"
 	"maps"
 	"slices"
 
 	"github.com/bmatcuk/doublestar/v4"
+	"go.uber.org/multierr"
 )
 
 func Validate(globs []string) error {
@@ -30,7 +30,7 @@ func FindFiles(includes []string, excludes []string) ([]string, error) {
 	for _, include := range includes {
 		matches, err := doublestar.FilepathGlob(include, doublestar.WithFilesOnly(), doublestar.WithFailOnIOErrors())
 		if err != nil {
-			errs = errors.Join(errs, fmt.Errorf("find files with '%s' pattern: %w", include, err))
+			errs = multierr.Append(errs, fmt.Errorf("find files with '%s' pattern: %w", include, err))
 			// the same pattern could cause an IO error due to one file or directory,
 			// but also could still find files without `doublestar.WithFailOnIOErrors()`.
 			matches, _ = doublestar.FilepathGlob(include, doublestar.WithFilesOnly())

pkg/stanza/fileconsumer/matcher/matcher.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"go.opentelemetry.io/collector/featuregate"
+	"go.uber.org/multierr"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/matcher/internal/filter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/matcher/internal/finder"
@@ -173,11 +174,9 @@ type Matcher struct {
 func (m Matcher) MatchFiles() ([]string, error) {
 	var errs error
 	files, err := finder.FindFiles(m.include, m.exclude)
-	if err != nil {
-		errs = errors.Join(errs, err)
-	}
+	errs = multierr.Append(errs, err)
 	if len(files) == 0 {
-		return files, errors.Join(errors.New("no files match the configured criteria"), errs)
+		return files, multierr.Append(errors.New("no files match the configured criteria"), errs)
 	}
 	if len(m.filterOpts) == 0 {
 		return files, errs
@@ -200,7 +199,7 @@ func (m Matcher) MatchFiles() ([]string, error) {
 	for _, groupedFiles := range groups {
 		groupResult, err := filter.Filter(groupedFiles, m.regex, m.filterOpts...)
 		if len(groupResult) == 0 {
-			return groupResult, errors.Join(err, errs)
+			return groupResult, multierr.Append(err, errs)
 		}
 		result = append(result, groupResult...)
 	}

pkg/stanza/operator/helper/transformer.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/expr-lang/expr/vm"
 	"go.opentelemetry.io/collector/component"
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 
@@ -78,11 +79,11 @@ func (t *TransformerOperator) CanProcess() bool {
 }
 
 func (t *TransformerOperator) ProcessBatchWith(ctx context.Context, entries []*entry.Entry, process ProcessFunction) error {
-	var errs []error
+	var errs error
 	for i := range entries {
-		errs = append(errs, process(ctx, entries[i]))
+		errs = multierr.Append(errs, process(ctx, entries[i]))
 	}
-	return errors.Join(errs...)
+	return errs
 }
 
 // ProcessWith will process an entry with a transform function.

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

@@ -5,9 +5,9 @@ package file // import "github.com/open-telemetry/opentelemetry-collector-contri
 
 import (
 	"context"
-	"errors"
 	"fmt"
 
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
@@ -40,22 +40,22 @@ func (i *Input) Stop() error {
 }
 
 func (i *Input) emitBatch(ctx context.Context, tokens [][]byte, attributes map[string]any, lastRecordNumber int64) error {
-	entries, conversionError := i.convertTokens(tokens, attributes, lastRecordNumber)
-	if conversionError != nil {
-		conversionError = fmt.Errorf("convert tokens: %w", conversionError)
+	var errs error
+	entries, err := i.convertTokens(tokens, attributes, lastRecordNumber)
+	if err != nil {
+		errs = multierr.Append(errs, fmt.Errorf("convert tokens: %w", err))
 	}
 
-	consumeError := i.WriteBatch(ctx, entries)
-	if consumeError != nil {
-		consumeError = fmt.Errorf("consume entries: %w", consumeError)
+	if err = i.WriteBatch(ctx, entries); err != nil {
+		errs = multierr.Append(errs, fmt.Errorf("consume entries: %w", err))
 	}
 
-	return errors.Join(conversionError, consumeError)
+	return errs
 }
 
 func (i *Input) convertTokens(tokens [][]byte, attributes map[string]any, lastRecordNumber int64) ([]*entry.Entry, error) {
 	entries := make([]*entry.Entry, 0, len(tokens))
-	var errs []error
+	var errs error
 
 	for tokenIndex, token := range tokens {
 		if len(token) == 0 {
@@ -64,7 +64,7 @@ func (i *Input) convertTokens(tokens [][]byte, attributes map[string]any, lastRe
 
 		ent, err := i.NewEntry(i.toBody(token))
 		if err != nil {
-			errs = append(errs, fmt.Errorf("create entry: %w", err))
+			errs = multierr.Append(errs, fmt.Errorf("create entry: %w", err))
 			continue
 		}
 
@@ -82,5 +82,5 @@ func (i *Input) convertTokens(tokens [][]byte, attributes map[string]any, lastRe
 
 		entries = append(entries, ent)
 	}
-	return entries, errors.Join(errs...)
+	return entries, errs
 }

pkg/stanza/operator/input/windows/input.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"go.opentelemetry.io/collector/component"
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
 	"golang.org/x/sys/windows"
 
@@ -165,18 +166,18 @@ func (i *Input) Stop() error {
 
 	var errs error
 	if err := i.subscription.Close(); err != nil {
-		errs = errors.Join(errs, fmt.Errorf("failed to close subscription: %w", err))
+		errs = multierr.Append(errs, fmt.Errorf("failed to close subscription: %w", err))
 	}
 
 	if err := i.bookmark.Close(); err != nil {
-		errs = errors.Join(errs, fmt.Errorf("failed to close bookmark: %w", err))
+		errs = multierr.Append(errs, fmt.Errorf("failed to close bookmark: %w", err))
 	}
 
 	if err := i.publisherCache.evictAll(); err != nil {
-		errs = errors.Join(errs, fmt.Errorf("failed to close publishers: %w", err))
+		errs = multierr.Append(errs, fmt.Errorf("failed to close publishers: %w", err))
 	}
 
-	return errors.Join(errs, i.stopRemoteSession())
+	return multierr.Append(errs, i.stopRemoteSession())
 }
 
 // readOnInterval will read events with respect to the polling interval until it reaches the end of the channel.
@@ -272,7 +273,7 @@ func (i *Input) renderDeepAndSend(ctx context.Context, event Event, publisher Pu
 	if err == nil {
 		return i.sendEvent(ctx, deepEvent)
 	}
-	return errors.Join(
+	return multierr.Append(
 		fmt.Errorf("render deep event: %w", err),
 		i.renderSimpleAndSend(ctx, event),
 	)
@@ -297,7 +298,7 @@ func (i *Input) processEventWithRenderingInfo(ctx context.Context, event Event)
 
 	publisher, err := i.publisherCache.get(providerName)
 	if err != nil {
-		return errors.Join(
+		return multierr.Append(
 			fmt.Errorf("open event source for provider %q: %w", providerName, err),
 			i.renderSimpleAndSend(ctx, event),
 		)

pkg/stanza/operator/input/windows/publishercache.go

@@ -6,7 +6,7 @@
 package windows // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/input/windows"
 
 import (
-	"errors"
+	"go.uber.org/multierr"
 )
 
 type publisherCache struct {
@@ -43,9 +43,7 @@ func (c *publisherCache) evictAll() error {
 	var errs error
 	for _, publisher := range c.cache {
 		if publisher.Valid() {
-			if err := publisher.Close(); err != nil {
-				errs = errors.Join(errs, err)
-			}
+			errs = multierr.Append(errs, publisher.Close())
 		}
 	}
 

pkg/stanza/operator/output/file/output.go

@@ -6,11 +6,11 @@ package file // import "github.com/open-telemetry/opentelemetry-collector-contri
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"os"
 	"sync"
 	"text/template"
 
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
@@ -54,11 +54,11 @@ func (o *Output) Stop() error {
 }
 
 func (o *Output) ProcessBatch(ctx context.Context, entries []*entry.Entry) error {
-	var errs []error
+	var errs error
 	for i := range entries {
-		errs = append(errs, o.Process(ctx, entries[i]))
+		errs = multierr.Append(errs, o.Process(ctx, entries[i]))
 	}
-	return errors.Join(errs...)
+	return errs
 }
 
 // Process will write an entry to the output file.

pkg/stanza/operator/output/stdout/output.go

@@ -6,9 +6,9 @@ package stdout // import "github.com/open-telemetry/opentelemetry-collector-cont
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"sync"
 
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
@@ -23,11 +23,11 @@ type Output struct {
 }
 
 func (o *Output) ProcessBatch(ctx context.Context, entries []*entry.Entry) error {
-	var errs []error
+	var errs error
 	for i := range entries {
-		errs = append(errs, o.Process(ctx, entries[i]))
+		errs = multierr.Append(errs, o.Process(ctx, entries[i]))
 	}
-	return errors.Join(errs...)
+	return errs
 }
 
 // Process will log entries received.

pkg/stanza/operator/parser/container/parser.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/goccy/go-json"
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/timeutils"
@@ -155,21 +156,19 @@ func (p *Parser) Stop() error {
 		// nothing is started return
 		return nil
 	}
-	var stopErrs []error
-	err := p.recombineParser.Stop()
-	if err != nil {
-		stopErrs = append(stopErrs, fmt.Errorf("unable to stop the internal recombine operator: %w", err))
+	var errs error
+	if err := p.recombineParser.Stop(); err != nil {
+		errs = multierr.Append(errs, fmt.Errorf("unable to stop the internal recombine operator: %w", err))
 	}
 	// the recombineParser will call the Process of the criLogEmitter synchronously so the entries will be first
 	// written to the channel before the Stop of the recombineParser returns. Then since the criLogEmitter handles
 	// the entries synchronously it is safe to call its Stop.
 	// After criLogEmitter is stopped the crioConsumer will consume the remaining messages and return.
-	err = p.criLogEmitter.Stop()
-	if err != nil {
-		stopErrs = append(stopErrs, fmt.Errorf("unable to stop the internal LogEmitter: %w", err))
+	if err := p.criLogEmitter.Stop(); err != nil {
+		errs = multierr.Append(errs, fmt.Errorf("unable to stop the internal LogEmitter: %w", err))
 	}
 	p.criConsumers.Wait()
-	return errors.Join(stopErrs...)
+	return errs
 }
 
 // detectFormat will detect the container log format