add WriterOperator::WriteBatch and Operator::ProcessBatch methods

The File input's `emitBatch` function now calls `ProcessBatch` instead of `Process`.

The added `ProcessBatch` method will make each Stanza operator
capable of accepting a batch of entries.

At this stage, all the implementations of `ProcessBatch` just call `Process` in a loop.

Author: andrzej-stencel

pkg/stanza/adapter/mocks_test.go

@@ -50,6 +50,10 @@ func (o *UnstartableOperator) Start(_ operator.Persister) error {
 	return errors.New("something very unusual happened")
 }
 
+func (o *UnstartableOperator) ProcessBatch(_ context.Context, _ []entry.Entry) error {
+	return nil
+}
+
 // Process will return nil
 func (o *UnstartableOperator) Process(_ context.Context, _ *entry.Entry) error {
 	return nil

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

@@ -30,6 +30,13 @@ func newPipelineOutput(set component.TelemetrySettings) *pipelineOutput {
 	}
 }
 
+func (e *pipelineOutput) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	for i := range entries {
+		_ = e.Process(ctx, &entries[i])
+	}
+	return nil
+}
+
 // Drop the entry if logChan is full, in order to avoid this operator blocking.
 // This protects against a case where an operator could return an error, but continue propagating a log entry,
 // leaving an unexpected entry in the output channel.

pkg/stanza/operator/helper/emitter.go

@@ -94,6 +94,15 @@ func (e *LogEmitter) Stop() error {
 	return nil
 }
 
+// ProcessBatch emits the entries to the consumerFunc
+func (e *LogEmitter) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	for i := range entries {
+		_ = e.Process(ctx, &entries[i])
+	}
+
+	return nil
+}
+
 // Process will emit an entry to the output channel
 func (e *LogEmitter) Process(ctx context.Context, ent *entry.Entry) error {
 	if oldBatch := e.appendEntry(ent); len(oldBatch) > 0 {

pkg/stanza/operator/helper/input.go

@@ -82,6 +82,15 @@ func (i *InputOperator) CanProcess() bool {
 	return false
 }
 
+// ProcessBatch will always return an error if called.
+func (i *InputOperator) ProcessBatch(_ context.Context, _ []entry.Entry) error {
+	i.Logger().Error("Operator received a batch of entries, but can not process")
+	return errors.NewError(
+		"Operator can not process logs.",
+		"Ensure that operator is not configured to receive logs from other operators",
+	)
+}
+
 // Process will always return an error if called.
 func (i *InputOperator) Process(_ context.Context, _ *entry.Entry) error {
 	i.Logger().Error("Operator received an entry, but can not process")

pkg/stanza/operator/helper/writer.go

@@ -47,6 +47,25 @@ type WriterOperator struct {
 	OutputOperators []operator.Operator
 }
 
+// Write writes a batch of entries to the outputs of the operator.
+// A batch is a collection of entries that are sent in one go.
+func (w *WriterOperator) WriteBatch(ctx context.Context, entries []entry.Entry) error {
+	for i, op := range w.OutputOperators {
+		if i == len(w.OutputOperators)-1 {
+			return op.ProcessBatch(ctx, entries)
+		}
+		copyOfEntries := make([]entry.Entry, 0, len(entries))
+		for i := range entries {
+			copyOfEntries = append(copyOfEntries, *entries[i].Copy())
+		}
+		err := op.ProcessBatch(ctx, copyOfEntries)
+		if err != nil {
+			w.Logger().Error("Failed to process entries", zap.Error(err))
+		}
+	}
+	return nil
+}
+
 // Write will write an entry to the outputs of the operator.
 func (w *WriterOperator) Write(ctx context.Context, e *entry.Entry) error {
 	for i, op := range w.OutputOperators {

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

@@ -39,33 +39,46 @@ func (i *Input) Stop() error {
 }
 
 func (i *Input) emitBatch(ctx context.Context, tokens []emit.Token) error {
+	entries, conversionError := i.convertTokens(tokens)
+	if conversionError != nil {
+		conversionError = fmt.Errorf("convert tokens: %w", conversionError)
+	}
+
+	consumeError := i.WriteBatch(ctx, entries)
+	if consumeError != nil {
+		consumeError = fmt.Errorf("consume entries: %w", consumeError)
+	}
+
+	return errors.Join(conversionError, consumeError)
+}
+
+func (i *Input) convertTokens(tokens []emit.Token) ([]entry.Entry, error) {
+	entries := make([]entry.Entry, 0, len(tokens))
 	var errs []error
 	for _, token := range tokens {
-		err := i.emit(ctx, token)
+		if len(token.Body) == 0 {
+			continue
+		}
+		entry, err := i.convertToken(token)
 		if err != nil {
 			errs = append(errs, err)
+			continue
 		}
+		entries = append(entries, *entry)
 	}
-	if len(errs) > 0 {
-		return errors.Join(errs...)
-	}
-	return nil
+	return entries, errors.Join(errs...)
 }
 
-func (i *Input) emit(ctx context.Context, token emit.Token) error {
-	if len(token.Body) == 0 {
-		return nil
-	}
-
+func (i *Input) convertToken(token emit.Token) (*entry.Entry, error) {
 	ent, err := i.NewEntry(i.toBody(token.Body))
 	if err != nil {
-		return fmt.Errorf("create entry: %w", err)
+		return nil, fmt.Errorf("create entry: %w", err)
 	}
 
 	for k, v := range token.Attributes {
 		if err := ent.Set(entry.NewAttributeField(k), v); err != nil {
 			i.Logger().Error("set attribute", zap.Error(err))
 		}
 	}
-	return i.Write(ctx, ent)
+	return ent, nil
 }

pkg/stanza/operator/operator.go

@@ -36,6 +36,8 @@ type Operator interface {
 
 	// CanProcess indicates if the operator will process entries from other operators.
 	CanProcess() bool
+	// Process processes a batch of entries from an operator.
+	ProcessBatch(context.Context, []entry.Entry) error
 	// Process will process an entry from an operator.
 	Process(context.Context, *entry.Entry) error
 	// Logger returns the operator's logger

pkg/stanza/operator/output/drop/output.go

@@ -15,6 +15,10 @@ type Output struct {
 	helper.OutputOperator
 }
 
+func (o *Output) ProcessBatch(_ context.Context, _ []entry.Entry) error {
+	return nil
+}
+
 // Process will drop the incoming entry.
 func (o *Output) Process(_ context.Context, _ *entry.Entry) error {
 	return nil

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

@@ -6,6 +6,7 @@ package file // import "github.com/open-telemetry/opentelemetry-collector-contri
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"os"
 	"sync"
 	"text/template"
@@ -52,6 +53,14 @@ func (o *Output) Stop() error {
 	return nil
 }
 
+func (o *Output) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	var errs []error
+	for i := range entries {
+		errs = append(errs, o.Process(ctx, &entries[i]))
+	}
+	return errors.Join(errs...)
+}
+
 // Process will write an entry to the output file.
 func (o *Output) Process(_ context.Context, entry *entry.Entry) error {
 	o.mux.Lock()

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

@@ -6,6 +6,7 @@ package stdout // import "github.com/open-telemetry/opentelemetry-collector-cont
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"sync"
 
 	"go.uber.org/zap"
@@ -21,6 +22,14 @@ type Output struct {
 	mux     sync.Mutex
 }
 
+func (o *Output) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	var errs []error
+	for i := range entries {
+		errs = append(errs, o.Process(ctx, &entries[i]))
+	}
+	return errors.Join(errs...)
+}
+
 // Process will log entries received.
 func (o *Output) Process(_ context.Context, entry *entry.Entry) error {
 	o.mux.Lock()

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

@@ -68,6 +68,14 @@ type Parser struct {
 	timeLayout              string
 }
 
+func (p *Parser) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	var errs []error
+	for i := range entries {
+		errs = append(errs, p.Process(ctx, &entries[i]))
+	}
+	return errors.Join(errs...)
+}
+
 // Process will parse an entry of Container logs
 func (p *Parser) Process(ctx context.Context, entry *entry.Entry) (err error) {
 	format := p.format

pkg/stanza/operator/parser/csv/parser.go

@@ -4,6 +4,7 @@ package csv // import "github.com/open-telemetry/opentelemetry-collector-contrib
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 
@@ -27,6 +28,14 @@ type Parser struct {
 
 type parseFunc func(any) (any, error)
 
+func (p *Parser) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	var errs []error
+	for i := range entries {
+		errs = append(errs, p.Process(ctx, &entries[i]))
+	}
+	return errors.Join(errs...)
+}
+
 // Process will parse an entry for csv.
 func (p *Parser) Process(ctx context.Context, e *entry.Entry) error {
 	// Static parse function

pkg/stanza/operator/parser/json/parser.go

@@ -5,6 +5,7 @@ package json // import "github.com/open-telemetry/opentelemetry-collector-contri
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 
@@ -21,6 +22,14 @@ type Parser struct {
 	parseInts bool
 }
 
+func (p *Parser) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	var errs []error
+	for i := range entries {
+		errs = append(errs, p.Process(ctx, &entries[i]))
+	}
+	return errors.Join(errs...)
+}
+
 // Process will parse an entry for JSON.
 func (p *Parser) Process(ctx context.Context, entry *entry.Entry) error {
 	return p.ParserOperator.ProcessWith(ctx, entry, p.parse)

pkg/stanza/operator/parser/jsonarray/parser.go

@@ -21,6 +21,14 @@ type Parser struct {
 
 type parseFunc func(any) (any, error)
 
+func (p *Parser) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	var errs []error
+	for i := range entries {
+		errs = append(errs, p.Process(ctx, &entries[i]))
+	}
+	return errors.Join(errs...)
+}
+
 // Process will parse an entry for json array.
 func (p *Parser) Process(ctx context.Context, e *entry.Entry) error {
 	return p.ParserOperator.ProcessWith(ctx, e, p.parse)

pkg/stanza/operator/parser/keyvalue/parser.go

@@ -5,6 +5,7 @@ package keyvalue // import "github.com/open-telemetry/opentelemetry-collector-co
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/parseutils"
@@ -19,6 +20,14 @@ type Parser struct {
 	pairDelimiter string
 }
 
+func (p *Parser) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	var errs []error
+	for i := range entries {
+		errs = append(errs, p.Process(ctx, &entries[i]))
+	}
+	return errors.Join(errs...)
+}
+
 // Process will parse an entry for key value pairs.
 func (p *Parser) Process(ctx context.Context, entry *entry.Entry) error {
 	return p.ParserOperator.ProcessWith(ctx, entry, p.parse)

pkg/stanza/operator/parser/regex/parser.go

@@ -5,6 +5,7 @@ package regex // import "github.com/open-telemetry/opentelemetry-collector-contr
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"regexp"
 
@@ -26,6 +27,14 @@ func (p *Parser) Stop() error {
 	return nil
 }
 
+func (p *Parser) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	var errs []error
+	for i := range entries {
+		errs = append(errs, p.Process(ctx, &entries[i]))
+	}
+	return errors.Join(errs...)
+}
+
 // Process will parse an entry for regex.
 func (p *Parser) Process(ctx context.Context, entry *entry.Entry) error {
 	return p.ParserOperator.ProcessWith(ctx, entry, p.parse)

pkg/stanza/operator/parser/scope/parser.go

@@ -5,6 +5,7 @@ package scope // import "github.com/open-telemetry/opentelemetry-collector-contr
 
 import (
 	"context"
+	"errors"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
@@ -16,6 +17,14 @@ type Parser struct {
 	helper.ScopeNameParser
 }
 
+func (p *Parser) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	var errs []error
+	for i := range entries {
+		errs = append(errs, p.Process(ctx, &entries[i]))
+	}
+	return errors.Join(errs...)
+}
+
 // Process will parse logger name from an entry.
 func (p *Parser) Process(ctx context.Context, entry *entry.Entry) error {
 	return p.ProcessWith(ctx, entry, p.Parse)

pkg/stanza/operator/parser/severity/parser.go

@@ -5,6 +5,7 @@ package severity // import "github.com/open-telemetry/opentelemetry-collector-co
 
 import (
 	"context"
+	"errors"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
@@ -16,6 +17,14 @@ type Parser struct {
 	helper.SeverityParser
 }
 
+func (p *Parser) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	var errs []error
+	for i := range entries {
+		errs = append(errs, p.Process(ctx, &entries[i]))
+	}
+	return errors.Join(errs...)
+}
+
 // Process will parse severity from an entry.
 func (p *Parser) Process(ctx context.Context, entry *entry.Entry) error {
 	return p.ProcessWith(ctx, entry, p.Parse)

pkg/stanza/operator/parser/syslog/parser.go

@@ -6,6 +6,7 @@ package syslog // import "github.com/open-telemetry/opentelemetry-collector-cont
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"regexp"
 	"time"
@@ -36,6 +37,14 @@ type Parser struct {
 	maxOctets                    int
 }
 
+func (p *Parser) ProcessBatch(ctx context.Context, entries []entry.Entry) error {
+	var errs []error
+	for i := range entries {
+		errs = append(errs, p.Process(ctx, &entries[i]))
+	}
+	return errors.Join(errs...)
+}
+
 // Process will parse an entry field as syslog.
 func (p *Parser) Process(ctx context.Context, entry *entry.Entry) error {
 	// if pri header is missing and this is an expected behavior then facility and severity values should be skipped.