[pkg/stanza] Fix syslog parser operator logging errors at ERROR level when on_error is set to quiet mode

Signed-off-by: Paulo Dias <[email protected]>

Author: paulojmdias

.chloggen/fix_stanza_parser_syslog.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 syslog parser operator 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/parser/syslog/parser.go

@@ -47,7 +47,11 @@ func (p *Parser) ProcessBatch(ctx context.Context, entries []*entry.Entry) error
 	for _, ent := range entries {
 		skip, err := p.Skip(ctx, ent)
 		if err != nil {
-			errs = append(errs, p.HandleEntryErrorWithWrite(ctx, ent, err, write))
+			if handleErr := p.HandleEntryErrorWithWrite(ctx, ent, err, write); handleErr != nil {
+				if p.OnError != helper.DropOnErrorQuiet && p.OnError != helper.SendOnErrorQuiet {
+					errs = append(errs, handleErr)
+				}
+			}
 			continue
 		}
 		if skip {
@@ -61,7 +65,11 @@ func (p *Parser) ProcessBatch(ctx context.Context, entries []*entry.Entry) error
 			var bytes []byte
 			bytes, err = toBytes(ent.Body)
 			if err != nil {
-				errs = append(errs, p.HandleEntryErrorWithWrite(ctx, ent, err, write))
+				if handleErr := p.HandleEntryErrorWithWrite(ctx, ent, err, write); handleErr != nil {
+					if p.OnError != helper.DropOnErrorQuiet && p.OnError != helper.SendOnErrorQuiet {
+						errs = append(errs, handleErr)
+					}
+				}
 				continue
 			}
 			if p.shouldSkipPriorityValues(bytes) {
@@ -77,7 +85,11 @@ func (p *Parser) ProcessBatch(ctx context.Context, entries []*entry.Entry) error
 		}
 
 		if err = callback(ent); err != nil {
-			errs = append(errs, p.HandleEntryErrorWithWrite(ctx, ent, err, write))
+			if handleErr := p.HandleEntryErrorWithWrite(ctx, ent, err, write); handleErr != nil {
+				if p.OnError != helper.DropOnErrorQuiet && p.OnError != helper.SendOnErrorQuiet {
+					errs = append(errs, handleErr)
+				}
+			}
 			continue
 		}
 
@@ -94,7 +106,11 @@ func (p *Parser) Process(ctx context.Context, entry *entry.Entry) error {
 	if !p.enableOctetCounting && p.allowSkipPriHeader {
 		bytes, err := toBytes(entry.Body)
 		if err != nil {
-			return err
+			handleErr := p.HandleEntryError(ctx, entry, err)
+			if p.OnError == helper.DropOnErrorQuiet || p.OnError == helper.SendOnErrorQuiet {
+				return nil
+			}
+			return handleErr
 		}
 
 		if p.shouldSkipPriorityValues(bytes) {

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

@@ -172,3 +172,133 @@ func TestSyslogParserDoesNotSplitBatches(t *testing.T) {
 	}))
 	output.AssertNumberOfCalls(t, "ProcessBatch", 1)
 }
+
+func TestSyslogQuietModeProcessBatch(t *testing.T) {
+	testCases := []struct {
+		name        string
+		onError     string
+		expectError bool
+	}{
+		{
+			name:        "DropOnErrorQuiet_ReturnsNoError",
+			onError:     "drop_quiet",
+			expectError: false,
+		},
+		{
+			name:        "SendOnErrorQuiet_ReturnsNoError",
+			onError:     "send_quiet",
+			expectError: false,
+		},
+		{
+			name:        "DropOnError_ReturnsError",
+			onError:     "drop",
+			expectError: true,
+		},
+		{
+			name:        "SendOnError_ReturnsError",
+			onError:     "send",
+			expectError: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			cfg := basicConfig()
+			cfg.Protocol = syslog.RFC5424
+			cfg.OnError = tc.onError
+
+			set := componenttest.NewNopTelemetrySettings()
+			op, err := cfg.Build(set)
+			require.NoError(t, err)
+
+			output := &testutil.Operator{}
+			output.On("ID").Return("fake")
+			output.On("CanProcess").Return(true)
+			output.On("ProcessBatch", mock.Anything, mock.Anything).Return(nil)
+			require.NoError(t, op.SetOutputs([]operator.Operator{output}))
+
+			ctx := t.Context()
+
+			// Create entries with invalid syslog messages that will cause parse errors
+			entry1 := entry.New()
+			entry1.Body = "invalid syslog message 1"
+
+			entry2 := entry.New()
+			entry2.Body = "invalid syslog message 2"
+
+			entry3 := entry.New()
+			entry3.Body = "invalid syslog message 3"
+
+			testEntries := []*entry.Entry{entry1, entry2, entry3}
+
+			err = op.ProcessBatch(ctx, testEntries)
+			if tc.expectError {
+				require.Error(t, err, "expected error in non-quiet mode")
+			} else {
+				require.NoError(t, err, "expected no error in quiet mode")
+			}
+		})
+	}
+}
+
+// TestSyslogQuietModeProcess tests quiet mode handling for toBytes error
+// in the Process method when allowSkipPriHeader is enabled
+func TestSyslogQuietModeProcess(t *testing.T) {
+	testCases := []struct {
+		name        string
+		onError     string
+		expectError bool
+	}{
+		{
+			name:        "DropOnErrorQuiet_ReturnsNoError",
+			onError:     "drop_quiet",
+			expectError: false,
+		},
+		{
+			name:        "SendOnErrorQuiet_ReturnsNoError",
+			onError:     "send_quiet",
+			expectError: false,
+		},
+		{
+			name:        "DropOnError_ReturnsError",
+			onError:     "drop",
+			expectError: true,
+		},
+		{
+			name:        "SendOnError_ReturnsError",
+			onError:     "send",
+			expectError: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			cfg := basicConfig()
+			cfg.Protocol = syslog.RFC3164
+			cfg.AllowSkipPriHeader = true // Enable this to trigger the toBytes path
+			cfg.OnError = tc.onError
+
+			set := componenttest.NewNopTelemetrySettings()
+			op, err := cfg.Build(set)
+			require.NoError(t, err)
+
+			output := &testutil.Operator{}
+			output.On("ID").Return("fake")
+			output.On("CanProcess").Return(true)
+			output.On("Process", mock.Anything, mock.Anything).Return(nil)
+			require.NoError(t, op.SetOutputs([]operator.Operator{output}))
+
+			// Create entry with non-string body that will cause toBytes error
+			e := entry.New()
+			e.Body = 12345 // Non-string body causes toBytes to fail
+			e.ObservedTimestamp = time.Now()
+
+			err = op.Process(t.Context(), e)
+			if tc.expectError {
+				require.Error(t, err, "expected error in non-quiet mode")
+			} else {
+				require.NoError(t, err, "expected no error in quiet mode")
+			}
+		})
+	}
+}