[pkg/stanza] Ensure filter operator does not split batches of entries (#45161)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Updates the `filter` operator to process batches of entries without
splitting them. The parser now writes all entries as a single batch,
rather than processing them individually.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes #42391

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Added unit tests to ensure the batches are not split.

---------

Signed-off-by: Paulo Dias <paulodias.gm@gmail.com>
Co-authored-by: Andrzej Stencel <andrzej.stencel@elastic.co>

Author: paulojmdias

.chloggen/fix_42391.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. receiver/filelog)
+component: "pkg/stanza"
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Ensure filter operator does not split batches of entries"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [42391]
+
+# (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: [user]

pkg/stanza/README.md

@@ -93,6 +93,7 @@ Operators that support batching:
 - `assign_keys`
 - `container`
 - `copy`
+- `filter`
 - `flatten`
 - `json_array_parser`
 - `json_parser`
@@ -114,7 +115,6 @@ Operators that support batching:
 Operators that do not support batching:
 
 - `csv_parser`
-- `filter`
 - `router`
 
 ### FAQ

pkg/stanza/operator/transformer/filter/transformer.go

@@ -6,6 +6,7 @@ package filter // import "github.com/open-telemetry/opentelemetry-collector-cont
 import (
 	"context"
 	"crypto/rand"
+	"errors"
 	"math/big"
 
 	"github.com/expr-lang/expr/vm"
@@ -23,7 +24,42 @@ type Transformer struct {
 }
 
 func (t *Transformer) ProcessBatch(ctx context.Context, entries []*entry.Entry) error {
-	return t.ProcessBatchWith(ctx, entries, t.Process)
+	filteredEntries := make([]*entry.Entry, 0, len(entries))
+	var errs []error
+	for _, ent := range entries {
+		env := helper.GetExprEnv(ent)
+		matches, err := vm.Run(t.expression, env)
+		helper.PutExprEnv(env)
+
+		if err != nil {
+			t.Logger().Error("Running expressing returned an error", zap.Error(err))
+			continue
+		}
+
+		filtered, ok := matches.(bool)
+		if !ok {
+			t.Logger().Error("Expression did not compile as a boolean")
+			continue
+		}
+
+		if !filtered {
+			filteredEntries = append(filteredEntries, ent)
+			continue
+		}
+
+		i, err := randInt(rand.Reader, upperBound)
+		if err != nil {
+			errs = append(errs, err)
+			continue
+		}
+
+		if i.Cmp(t.dropCutoff) >= 0 {
+			filteredEntries = append(filteredEntries, ent)
+		}
+	}
+
+	errs = append(errs, t.WriteBatch(ctx, filteredEntries))
+	return errors.Join(errs...)
 }
 
 // Process will drop incoming entries that match the filter expression

pkg/stanza/operator/transformer/filter/transformer_test.go

@@ -185,8 +185,11 @@ func TestTransformer(t *testing.T) {
 
 			filtered := true
 			mockOutput := testutil.NewMockOperator("output")
-			mockOutput.On("Process", mock.Anything, mock.Anything).Return(nil).Run(func(_ mock.Arguments) {
-				filtered = false
+			mockOutput.On("ProcessBatch", mock.Anything, mock.Anything).Return(nil).Run(func(args mock.Arguments) {
+				entries := args.Get(1).([]*entry.Entry)
+				if len(entries) > 0 {
+					filtered = false
+				}
 			})
 
 			op.(*Transformer).OutputOperators = []operator.Operator{mockOutput}
@@ -208,8 +211,9 @@ func TestFilterDropRatio(t *testing.T) {
 
 	processedEntries := 0
 	mockOutput := testutil.NewMockOperator("output")
-	mockOutput.On("Process", mock.Anything, mock.Anything).Return(nil).Run(func(_ mock.Arguments) {
-		processedEntries++
+	mockOutput.On("ProcessBatch", mock.Anything, mock.Anything).Return(nil).Run(func(args mock.Arguments) {
+		entries := args.Get(1).([]*entry.Entry)
+		processedEntries += len(entries)
 	})
 
 	filterOperator, ok := op.(*Transformer)
@@ -243,3 +247,36 @@ func TestFilterDropRatio(t *testing.T) {
 
 	require.Equal(t, 10, processedEntries)
 }
+
+func TestFilterDoesNotSplitBatches(t *testing.T) {
+	cfg := NewConfigWithID("test")
+	cfg.Expression = `body.message == "drop_me"`
+	set := componenttest.NewNopTelemetrySettings()
+	op, err := cfg.Build(set)
+	require.NoError(t, err)
+
+	mockOutput := testutil.NewMockOperator("output")
+	mockOutput.On("ProcessBatch", mock.Anything, mock.Anything).Return(nil)
+
+	filterOperator := op.(*Transformer)
+	filterOperator.OutputOperators = []operator.Operator{mockOutput}
+
+	testEntries := []*entry.Entry{
+		{Body: map[string]any{"message": "keep_me"}},
+		{Body: map[string]any{"message": "keep_me_too"}},
+		{Body: map[string]any{"message": "drop_me"}},
+		{Body: map[string]any{"message": "also_keep"}},
+	}
+
+	err = filterOperator.ProcessBatch(t.Context(), testEntries)
+	require.NoError(t, err)
+
+	// Verify that ProcessBatch was called exactly once, proving the batch was not split
+	mockOutput.AssertNumberOfCalls(t, "ProcessBatch", 1)
+
+	// Verify that 3 entries were passed (one was filtered out)
+	calls := mockOutput.Calls
+	require.Len(t, calls, 1)
+	passedEntries := calls[0].Arguments.Get(1).([]*entry.Entry)
+	require.Len(t, passedEntries, 3)
+}