[pkg/stanza] Fix transformer operators logging errors at ERROR level when on_error is set to quiet mode (#45366)

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

Fix transformer operators logging errors at ERROR level when `on_error`
is set to quiet mode (`send_quiet` or `drop_quiet`)

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

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

Updated unit tests and added tests to all operators, which should
support `on_error` configuration.

---------

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

Author: paulojmdias

.chloggen/fix_42646.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: "bug_fix"
+
+# 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: "Fix transformer operators logging errors at ERROR level when `on_error` is set to quiet mode"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [42646]
+
+# (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/operator/helper/transformer.go

@@ -106,7 +106,12 @@ func (t *TransformerOperator) ProcessBatchWithTransform(ctx context.Context, ent
 		}
 
 		if err = transform(ent); err != nil {
-			errs = append(errs, t.HandleEntryErrorWithWrite(ctx, ent, err, write))
+			if handleErr := t.HandleEntryErrorWithWrite(ctx, ent, err, write); handleErr != nil {
+				// Only append error if not in quiet mode
+				if !t.isQuietMode() {
+					errs = append(errs, handleErr)
+				}
+			}
 			continue
 		}
 
@@ -130,7 +135,12 @@ func (t *TransformerOperator) ProcessWith(ctx context.Context, entry *entry.Entr
 	}
 
 	if err := transform(entry); err != nil {
-		return t.HandleEntryError(ctx, entry, err)
+		handleErr := t.HandleEntryError(ctx, entry, err)
+		// Return nil for quiet modes to prevent error from bubbling up
+		if t.isQuietMode() {
+			return nil
+		}
+		return handleErr
 	}
 	return t.Write(ctx, entry)
 }
@@ -145,7 +155,7 @@ func (t *TransformerOperator) HandleEntryErrorWithWrite(ctx context.Context, ent
 		return errors.New("got a nil entry, this should not happen and is potentially a bug")
 	}
 
-	if t.OnError == SendOnErrorQuiet || t.OnError == DropOnErrorQuiet {
+	if t.isQuietMode() {
 		// No need to construct the zap attributes if logging not enabled at debug level.
 		if t.Logger().Core().Enabled(zapcore.DebugLevel) {
 			t.Logger().Debug("Failed to process entry", zapAttributes(entry, t.OnError, err)...)
@@ -162,6 +172,11 @@ func (t *TransformerOperator) HandleEntryErrorWithWrite(ctx context.Context, ent
 	return err
 }
 
+// isQuietMode returns true if the operator is configured to use quiet mode
+func (t *TransformerOperator) isQuietMode() bool {
+	return t.OnError == DropOnErrorQuiet || t.OnError == SendOnErrorQuiet
+}
+
 func (t *TransformerOperator) Skip(_ context.Context, entry *entry.Entry) (bool, error) {
 	if t.IfExpr == nil {
 		return false, nil

pkg/stanza/operator/helper/transformer_test.go

@@ -151,7 +151,7 @@ func TestTransformerDropOnErrorQuiet(t *testing.T) {
 	}
 
 	err := transformer.ProcessWith(ctx, testEntry, transform)
-	require.Error(t, err)
+	require.NoError(t, err)
 	output.AssertNotCalled(t, "Process", mock.Anything, mock.Anything)
 
 	// Test output logs
@@ -251,7 +251,7 @@ func TestTransformerSendOnErrorQuiet(t *testing.T) {
 	}
 
 	err := transformer.ProcessWith(ctx, testEntry, transform)
-	require.Error(t, err)
+	require.NoError(t, err)
 	output.AssertCalled(t, "Process", mock.Anything, mock.Anything)
 
 	// Test output logs
@@ -380,6 +380,284 @@ func TestTransformerDoesNotSplitBatches(t *testing.T) {
 	output.AssertNumberOfCalls(t, "ProcessBatch", 1)
 }
 
+// TestBatchProcessingAllEntriesFailQuietMode tests that when all entries fail in a batch with quiet mode,
+// no errors are returned and entries are handled according to the quiet mode (drop or send).
+func TestBatchProcessingAllEntriesFailQuietMode(t *testing.T) {
+	testCases := []struct {
+		name             string
+		onError          string
+		expectProcessed  bool
+		expectError      bool
+		expectedLogLevel zapcore.Level
+	}{
+		{
+			name:             "DropOnErrorQuiet",
+			onError:          DropOnErrorQuiet,
+			expectProcessed:  false,
+			expectError:      false,
+			expectedLogLevel: zapcore.DebugLevel,
+		},
+		{
+			name:             "SendOnErrorQuiet",
+			onError:          SendOnErrorQuiet,
+			expectProcessed:  true,
+			expectError:      false,
+			expectedLogLevel: zapcore.DebugLevel,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			output := &testutil.Operator{}
+			output.On("ID").Return("test-output")
+			output.On("ProcessBatch", mock.Anything, mock.Anything).Return(nil)
+
+			obs, logs := observer.New(zapcore.DebugLevel)
+			set := componenttest.NewNopTelemetrySettings()
+			set.Logger = zap.New(obs)
+
+			transformer := TransformerOperator{
+				OnError: tc.onError,
+				WriterOperator: WriterOperator{
+					BasicOperator: BasicOperator{
+						OperatorID:   "test-id",
+						OperatorType: "test-type",
+						set:          set,
+					},
+					OutputOperators: []operator.Operator{output},
+					OutputIDs:       []string{"test-output"},
+				},
+			}
+
+			ctx := t.Context()
+			testEntry1 := entry.New()
+			testEntry2 := entry.New()
+			testEntry3 := entry.New()
+			testEntries := []*entry.Entry{testEntry1, testEntry2, testEntry3}
+
+			// Transform function that always fails
+			transform := func(_ *entry.Entry) error {
+				return errors.New("transform failure")
+			}
+
+			err := transformer.ProcessBatchWithTransform(ctx, testEntries, transform)
+
+			if tc.expectError {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err, "expected no error in quiet mode")
+			}
+
+			// Verify that logs were created at the expected level
+			require.Equal(t, 3, logs.Len(), "expected 3 log entries for 3 failed entries")
+			for _, logEntry := range logs.All() {
+				require.Equal(t, tc.expectedLogLevel, logEntry.Level, "expected log level to be %v", tc.expectedLogLevel)
+			}
+
+			// Verify ProcessBatch was called
+			output.AssertCalled(t, "ProcessBatch", mock.Anything, mock.Anything)
+
+			// Check if entries were sent or dropped
+			calls := output.Calls
+			if len(calls) > 0 {
+				processedEntries := calls[0].Arguments.Get(1).([]*entry.Entry)
+				if tc.expectProcessed {
+					require.Len(t, processedEntries, 3, "expected all 3 entries to be sent in SendOnErrorQuiet mode")
+				} else {
+					require.Empty(t, processedEntries, "expected 0 entries to be sent in DropOnErrorQuiet mode")
+				}
+			}
+		})
+	}
+}
+
+// TestBatchProcessingMixedSuccessFailureQuietMode tests batch processing with a mix of successful and failed entries in quiet mode.
+func TestBatchProcessingMixedSuccessFailureQuietMode(t *testing.T) {
+	testCases := []struct {
+		name                   string
+		onError                string
+		expectError            bool
+		expectedProcessedCount int
+		expectedLogLevel       zapcore.Level
+	}{
+		{
+			name:                   "DropOnErrorQuiet",
+			onError:                DropOnErrorQuiet,
+			expectError:            false,
+			expectedProcessedCount: 2, // Only successful entries
+			expectedLogLevel:       zapcore.DebugLevel,
+		},
+		{
+			name:                   "SendOnErrorQuiet",
+			onError:                SendOnErrorQuiet,
+			expectError:            false,
+			expectedProcessedCount: 4, // All entries (2 successful + 2 failed)
+			expectedLogLevel:       zapcore.DebugLevel,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			output := &testutil.Operator{}
+			output.On("ID").Return("test-output")
+			output.On("ProcessBatch", mock.Anything, mock.Anything).Return(nil)
+
+			obs, logs := observer.New(zapcore.DebugLevel)
+			set := componenttest.NewNopTelemetrySettings()
+			set.Logger = zap.New(obs)
+
+			transformer := TransformerOperator{
+				OnError: tc.onError,
+				WriterOperator: WriterOperator{
+					BasicOperator: BasicOperator{
+						OperatorID:   "test-id",
+						OperatorType: "test-type",
+						set:          set,
+					},
+					OutputOperators: []operator.Operator{output},
+					OutputIDs:       []string{"test-output"},
+				},
+			}
+
+			ctx := t.Context()
+			testEntry1 := entry.New()
+			testEntry1.Body = "success1"
+			testEntry2 := entry.New()
+			testEntry2.Body = "fail1"
+			testEntry3 := entry.New()
+			testEntry3.Body = "success2"
+			testEntry4 := entry.New()
+			testEntry4.Body = "fail2"
+			testEntries := []*entry.Entry{testEntry1, testEntry2, testEntry3, testEntry4}
+
+			// Transform function that fails for entries with "fail" in body
+			transform := func(e *entry.Entry) error {
+				if body, ok := e.Body.(string); ok && body[:4] == "fail" {
+					return errors.New("transform failure")
+				}
+				return nil
+			}
+
+			err := transformer.ProcessBatchWithTransform(ctx, testEntries, transform)
+
+			if tc.expectError {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err, "expected no error in quiet mode")
+			}
+
+			// Verify that logs were created for failed entries only
+			require.Equal(t, 2, logs.Len(), "expected 2 log entries for 2 failed entries")
+			for _, logEntry := range logs.All() {
+				require.Equal(t, tc.expectedLogLevel, logEntry.Level, "expected log level to be %v", tc.expectedLogLevel)
+			}
+
+			// Verify ProcessBatch was called
+			output.AssertCalled(t, "ProcessBatch", mock.Anything, mock.Anything)
+
+			// Check the number of processed entries
+			calls := output.Calls
+			if len(calls) > 0 {
+				processedEntries := calls[0].Arguments.Get(1).([]*entry.Entry)
+				require.Len(t, processedEntries, tc.expectedProcessedCount,
+					"expected %d entries to be processed", tc.expectedProcessedCount)
+			}
+		})
+	}
+}
+
+// TestBatchProcessingAllEntriesFailNonQuietMode tests that errors are properly returned in non-quiet mode.
+func TestBatchProcessingAllEntriesFailNonQuietMode(t *testing.T) {
+	testCases := []struct {
+		name             string
+		onError          string
+		expectProcessed  bool
+		expectError      bool
+		expectedLogLevel zapcore.Level
+	}{
+		{
+			name:             "DropOnError",
+			onError:          DropOnError,
+			expectProcessed:  false,
+			expectError:      true,
+			expectedLogLevel: zapcore.ErrorLevel,
+		},
+		{
+			name:             "SendOnError",
+			onError:          SendOnError,
+			expectProcessed:  true,
+			expectError:      true,
+			expectedLogLevel: zapcore.ErrorLevel,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			output := &testutil.Operator{}
+			output.On("ID").Return("test-output")
+			output.On("ProcessBatch", mock.Anything, mock.Anything).Return(nil)
+
+			obs, logs := observer.New(zapcore.WarnLevel)
+			set := componenttest.NewNopTelemetrySettings()
+			set.Logger = zap.New(obs)
+
+			transformer := TransformerOperator{
+				OnError: tc.onError,
+				WriterOperator: WriterOperator{
+					BasicOperator: BasicOperator{
+						OperatorID:   "test-id",
+						OperatorType: "test-type",
+						set:          set,
+					},
+					OutputOperators: []operator.Operator{output},
+					OutputIDs:       []string{"test-output"},
+				},
+			}
+
+			ctx := t.Context()
+			testEntry1 := entry.New()
+			testEntry2 := entry.New()
+			testEntry3 := entry.New()
+			testEntries := []*entry.Entry{testEntry1, testEntry2, testEntry3}
+
+			// Transform function that always fails
+			transform := func(_ *entry.Entry) error {
+				return errors.New("transform failure")
+			}
+
+			err := transformer.ProcessBatchWithTransform(ctx, testEntries, transform)
+
+			if tc.expectError {
+				require.Error(t, err, "expected error in non-quiet mode")
+				// Verify we get multiple errors joined together
+				require.Contains(t, err.Error(), "transform failure")
+			} else {
+				require.NoError(t, err)
+			}
+
+			// Verify that logs were created at the expected level
+			require.Equal(t, 3, logs.Len(), "expected 3 log entries for 3 failed entries")
+			for _, logEntry := range logs.All() {
+				require.Equal(t, tc.expectedLogLevel, logEntry.Level)
+			}
+
+			// Verify ProcessBatch was called
+			output.AssertCalled(t, "ProcessBatch", mock.Anything, mock.Anything)
+
+			// Check if entries were sent or dropped
+			calls := output.Calls
+			if len(calls) > 0 {
+				processedEntries := calls[0].Arguments.Get(1).([]*entry.Entry)
+				if tc.expectProcessed {
+					require.Len(t, processedEntries, 3, "expected all 3 entries to be sent in SendOnError mode")
+				} else {
+					require.Empty(t, processedEntries, "expected 0 entries to be sent in DropOnError mode")
+				}
+			}
+		})
+	}
+}
+
 func TestTransformerIf(t *testing.T) {
 	cases := []struct {
 		name        string