[pkg/stanza] Fix container parser operator logging errors at ERROR level when on_error is set to quiet mode (#45704)

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

Fix container parser operator 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
Relates to #42646

Assisted-by: Claude Opus 4.7

---------

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

Author: paulojmdias

.chloggen/fix_stanza_transformer_container.yaml

@@ -0,0 +1,32 @@
+# 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 stanza container operator logging errors at ERROR level when `on_error` is set to a 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: |
+  Previously, the `container` operator logged entry-level processing errors at
+  ERROR level even when `on_error` was set to `drop_quiet` or `send_quiet`.
+  These errors are now logged at DEBUG level in quiet modes, matching the
+  documented behavior. Downstream delivery failures continue to propagate so
+  the pipeline can react to them.
+
+# 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/parser.go

@@ -14,6 +14,11 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/stanzaerrors"
 )
 
+// ErrEntryHandled signals that ParseWith already handled the entry
+// (logged and optionally written downstream) and the caller must not
+// write it again or propagate an error.
+var ErrEntryHandled = errors.New("entry handled by parser in quiet mode")
+
 // NewParserConfig creates a new parser config with default values
 func NewParserConfig(operatorID, operatorType string) ParserConfig {
 	return ParserConfig{
@@ -117,7 +122,7 @@ func (p *ParserOperator) ProcessBatchWithCallback(ctx context.Context, entries [
 		}
 
 		if err = p.ParseWith(ctx, ent, parse, write); err != nil {
-			if p.OnError != DropOnErrorQuiet && p.OnError != SendOnErrorQuiet {
+			if !errors.Is(err, ErrEntryHandled) {
 				errs = append(errs, err)
 			}
 			continue
@@ -153,15 +158,13 @@ func (p *ParserOperator) ProcessWithCallback(ctx context.Context, entry *entry.E
 	}
 
 	if err = p.ParseWith(ctx, entry, parse, p.Write); err != nil {
-		if p.OnError == DropOnErrorQuiet || p.OnError == SendOnErrorQuiet {
+		if errors.Is(err, ErrEntryHandled) {
 			return nil
 		}
-
 		return err
 	}
 	if cb != nil {
-		err = cb(entry)
-		if err != nil {
+		if err = cb(entry); err != nil {
 			return p.HandleEntryError(ctx, entry, err)
 		}
 	}
@@ -170,24 +173,35 @@ func (p *ParserOperator) ProcessWithCallback(ctx context.Context, entry *entry.E
 }
 
 // ParseWith will process an entry's field with a parser function.
+// In quiet on_error modes any entry-level error is handled internally and
+// ErrEntryHandled is returned so callers do not write or propagate again.
 func (p *ParserOperator) ParseWith(ctx context.Context, entry *entry.Entry, parse ParseFunction, write WriteFunction) error {
+	// handle translates a nil-in-quiet-mode return from HandleEntryErrorWithWrite
+	// into ErrEntryHandled.
+	handle := func(err error) error {
+		handled := p.HandleEntryErrorWithWrite(ctx, entry, err, write)
+		if handled == nil && p.isQuietMode() {
+			return ErrEntryHandled
+		}
+		return handled
+	}
+
 	value, ok := entry.Get(p.ParseFrom)
 	if !ok {
-		err := stanzaerrors.NewError(
+		return handle(stanzaerrors.NewError(
 			"Entry is missing the expected parse_from field.",
 			"Ensure that all incoming entries contain the parse_from field.",
 			"parse_from", p.ParseFrom.String(),
-		)
-		return p.HandleEntryErrorWithWrite(ctx, entry, err, write)
+		))
 	}
 
 	newValue, err := parse(value)
 	if err != nil {
-		return p.HandleEntryErrorWithWrite(ctx, entry, err, write)
+		return handle(err)
 	}
 
 	if err := entry.Set(p.ParseTo, newValue); err != nil {
-		return p.HandleEntryErrorWithWrite(ctx, entry, fmt.Errorf("set parse_to: %w", err), write)
+		return handle(fmt.Errorf("set parse_to: %w", err))
 	}
 
 	if p.BodyField != nil {
@@ -218,16 +232,16 @@ func (p *ParserOperator) ParseWith(ctx context.Context, entry *entry.Entry, pars
 
 	// Handle parsing errors after attempting to parse all
 	if timeParseErr != nil {
-		return p.HandleEntryErrorWithWrite(ctx, entry, fmt.Errorf("time parser: %w", timeParseErr), write)
+		return handle(fmt.Errorf("time parser: %w", timeParseErr))
 	}
 	if severityParseErr != nil {
-		return p.HandleEntryErrorWithWrite(ctx, entry, fmt.Errorf("severity parser: %w", severityParseErr), write)
+		return handle(fmt.Errorf("severity parser: %w", severityParseErr))
 	}
 	if traceParseErr != nil {
-		return p.HandleEntryErrorWithWrite(ctx, entry, fmt.Errorf("trace parser: %w", traceParseErr), write)
+		return handle(fmt.Errorf("trace parser: %w", traceParseErr))
 	}
 	if scopeNameParserErr != nil {
-		return p.HandleEntryErrorWithWrite(ctx, entry, fmt.Errorf("scope_name parser: %w", scopeNameParserErr), write)
+		return handle(fmt.Errorf("scope_name parser: %w", scopeNameParserErr))
 	}
 	return nil
 }

pkg/stanza/operator/helper/parser_test.go

@@ -703,6 +703,147 @@ func NewTestParserConfig() ParserConfig {
 	return expect
 }
 
+func TestProcessWithCallback_QuietMode(t *testing.T) {
+	// Quiet mode swallows processing errors, but send_quiet still surfaces
+	// downstream write failures so the pipeline can react to delivery errors.
+	// drop_quiet never calls write so its parse failure is fully suppressed.
+	testCases := []struct {
+		name            string
+		onError         string
+		expectWriteWrap bool
+	}{
+		{
+			name:            "SendOnErrorQuiet_WriteFailure_PropagatesWriteError",
+			onError:         SendOnErrorQuiet,
+			expectWriteWrap: true,
+		},
+		{
+			name:    "DropOnErrorQuiet_ParseFailure_Suppressed",
+			onError: DropOnErrorQuiet,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fakeOut := testutil.NewFakeOutputWithProcessError(t)
+			set := componenttest.NewNopTelemetrySettings()
+			set.Logger = zaptest.NewLogger(t)
+			writer := &WriterOperator{
+				BasicOperator: BasicOperator{
+					OperatorID:   "test-id",
+					OperatorType: "test-type",
+					set:          set,
+				},
+				OutputIDs: []string{fakeOut.ID()},
+			}
+			require.NoError(t, writer.SetOutputs([]operator.Operator{fakeOut}))
+
+			parser := ParserOperator{
+				TransformerOperator: TransformerOperator{
+					WriterOperator: *writer,
+					OnError:        tc.onError,
+				},
+				ParseFrom: entry.NewBodyField(),
+				ParseTo:   entry.NewAttributeField(),
+			}
+
+			parse := func(_ any) (any, error) {
+				return nil, errors.New("parse failure")
+			}
+
+			ctx := t.Context()
+			testEntry := entry.New()
+			err := parser.ProcessWithCallback(ctx, testEntry, parse, nil)
+			if tc.expectWriteWrap {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), "failed to send entry after error")
+			} else {
+				require.NoError(t, err, "quiet mode must not propagate the processing error")
+			}
+		})
+	}
+}
+
+// TestProcessWithCallback_QuietMode_SkipAndCallbackPaths exercises the four
+// code paths in helper/parser.go that previously did not honor quiet mode:
+//   - ProcessWithCallback Skip error
+//   - ProcessWithCallback callback error
+//   - ProcessBatchWithCallback Skip error
+//   - ProcessBatchWithCallback callback error
+//
+// All four must be suppressed when the operator is configured in quiet mode.
+func TestProcessWithCallback_QuietMode_SkipAndCallbackPaths(t *testing.T) {
+	parse := func(v any) (any, error) { return v, nil }
+	okCb := func(_ *entry.Entry) error { return nil }
+	failingCb := func(_ *entry.Entry) error { return errors.New("callback failure") }
+
+	// An "if" expression that always errors during evaluation (because the
+	// attribute is missing and types do not match), driving the Skip error
+	// path.
+	badIfExpr, err := ExprCompileBool(`attributes["missing"] + 1 == 2`)
+	require.NoError(t, err)
+
+	for _, onError := range []string{DropOnErrorQuiet, SendOnErrorQuiet} {
+		t.Run("ProcessWithCallback_Skip_"+onError, func(t *testing.T) {
+			writer, _ := writerWithFakeOut(t)
+			parser := ParserOperator{
+				TransformerOperator: TransformerOperator{
+					WriterOperator: *writer,
+					OnError:        onError,
+					IfExpr:         badIfExpr,
+				},
+				ParseFrom: entry.NewBodyField(),
+				ParseTo:   entry.NewAttributeField(),
+			}
+			require.NoError(t, parser.ProcessWithCallback(t.Context(), entry.New(), parse, okCb),
+				"Skip error must be suppressed in quiet mode")
+		})
+
+		t.Run("ProcessWithCallback_Callback_"+onError, func(t *testing.T) {
+			writer, _ := writerWithFakeOut(t)
+			parser := ParserOperator{
+				TransformerOperator: TransformerOperator{
+					WriterOperator: *writer,
+					OnError:        onError,
+				},
+				ParseFrom: entry.NewBodyField(),
+				ParseTo:   entry.NewAttributeField(),
+			}
+			require.NoError(t, parser.ProcessWithCallback(t.Context(), entry.New(), parse, failingCb),
+				"callback error must be suppressed in quiet mode")
+		})
+
+		t.Run("ProcessBatchWithCallback_Skip_"+onError, func(t *testing.T) {
+			writer, _ := writerWithFakeOut(t)
+			parser := ParserOperator{
+				TransformerOperator: TransformerOperator{
+					WriterOperator: *writer,
+					OnError:        onError,
+					IfExpr:         badIfExpr,
+				},
+				ParseFrom: entry.NewBodyField(),
+				ParseTo:   entry.NewAttributeField(),
+			}
+			require.NoError(t, parser.ProcessBatchWithCallback(t.Context(), []*entry.Entry{entry.New()}, parse, okCb),
+				"batch Skip error must be suppressed in quiet mode")
+		})
+
+		t.Run("ProcessBatchWithCallback_Callback_"+onError, func(t *testing.T) {
+			writer, _ := writerWithFakeOut(t)
+			parser := ParserOperator{
+				TransformerOperator: TransformerOperator{
+					WriterOperator: *writer,
+					OnError:        onError,
+				},
+				ParseFrom: entry.NewBodyField(),
+				ParseTo:   entry.NewAttributeField(),
+			}
+			require.NoError(t, parser.ProcessBatchWithCallback(t.Context(), []*entry.Entry{entry.New()}, parse, failingCb),
+				"batch callback error must be suppressed in quiet mode")
+		})
+	}
+}
+
 func writerWithFakeOut(t *testing.T) (*WriterOperator, *testutil.FakeOutput) {
 	fakeOut := testutil.NewFakeOutput(t)
 	set := componenttest.NewNopTelemetrySettings()

pkg/stanza/operator/helper/transformer.go

@@ -106,12 +106,7 @@ func (t *TransformerOperator) ProcessBatchWithTransform(ctx context.Context, ent
 		}
 
 		if err = transform(ent); err != nil {
-			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)
-				}
-			}
+			errs = append(errs, t.HandleEntryErrorWithWrite(ctx, ent, err, write))
 			continue
 		}
 
@@ -135,21 +130,22 @@ func (t *TransformerOperator) ProcessWith(ctx context.Context, entry *entry.Entr
 	}
 
 	if err := transform(entry); err != nil {
-		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.HandleEntryError(ctx, entry, err)
 	}
 	return t.Write(ctx, entry)
 }
 
-// HandleEntryError will handle an entry error using the on_error strategy.
+// HandleEntryError handles an entry error using the on_error strategy.
+// In quiet modes (drop_quiet, send_quiet) the processing error is swallowed,
+// but a downstream write error from send_quiet is still returned so the
+// pipeline can react to delivery failures.
 func (t *TransformerOperator) HandleEntryError(ctx context.Context, entry *entry.Entry, err error) error {
 	return t.HandleEntryErrorWithWrite(ctx, entry, err, t.Write)
 }
 
+// HandleEntryErrorWithWrite is like HandleEntryError but uses the supplied write function.
+// In quiet modes the processing error is swallowed; a downstream write error
+// from send_quiet is still returned.
 func (t *TransformerOperator) HandleEntryErrorWithWrite(ctx context.Context, entry *entry.Entry, err error, write WriteFunction) error {
 	if entry == nil {
 		return errors.New("got a nil entry, this should not happen and is potentially a bug")
@@ -163,12 +159,16 @@ func (t *TransformerOperator) HandleEntryErrorWithWrite(ctx context.Context, ent
 	} else {
 		t.Logger().Error("Failed to process entry", zapAttributes(entry, t.OnError, err)...)
 	}
+
 	if t.OnError == SendOnError || t.OnError == SendOnErrorQuiet {
 		if writeErr := write(ctx, entry); writeErr != nil {
-			err = fmt.Errorf("failed to send entry after error: %w", writeErr)
+			return fmt.Errorf("failed to send entry after error: %w", writeErr)
 		}
 	}
 
+	if t.isQuietMode() {
+		return nil
+	}
 	return err
 }