[pkg/stanza] Improve error logs produced by transformer processors (#37285)

#### Description

This improves the error messages of the log transformers processors to
include the log file path and the original log record, effectively
allowing users to quickly debug and investigate deeper the reason for
the problem **without** having to use the debug exporter or using the
highest verbosity level.

Here's an example of one of these failures: 

> ```2025-01-17T15:38:13.637Z error helper/transformer.go:114 Failed to
process entry {"kind": "receiver", "name": "filelog", "data_type":
"logs", "operator_id": "move5", "operator_type": "move",
"log.file.path":
"/var/log/pods/kube-system_kindnet-jxpz6_0784c9f9-ec2b-4829-aeb3-263ec66ef953/kindnet-cni/19.log",
"entry.timestamp": "2025-01-17T15:38:07.645938111Z", "error": "move:
field does not exist: attributes.uid", "action": "send"}```

This log line could be even more helpful if the it included the regex
pattern that was matched against the entry. I'm not sure about adding it
now and it could be added in the future if desired.

A small note on the name of the log keys: they could have shorter names,
like `body` and `file_path` but I think we could also use the attribute
names from the semantic conventions for logs. I have no preference
though and I'm happy to change it if needed.

#### Testing

- Ran it locally in a kind cluster.
- Some unit tests were updated to ensure the `log.record.original` key
is present in the logs.

Author: douglascamata

.chloggen/improve-log-transformers-logs.yaml

@@ -0,0 +1,29 @@
+# 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. filelogreceiver)
+component: pkg/stanza
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add entry's timestamp and attributes to errors logs from log transformers processors
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37285]
+
+# (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: |
+  When a log transformer processor fails to process an log entry it will include entry's timestamp and attributes in its own logs.
+  With this information the user can more easily identify the log file and find the entry that's having issues.
+
+# 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

@@ -94,10 +94,23 @@ func (t *TransformerOperator) ProcessWith(ctx context.Context, entry *entry.Entr
 
 // HandleEntryError will handle an entry error using the on_error strategy.
 func (t *TransformerOperator) HandleEntryError(ctx context.Context, entry *entry.Entry, err error) error {
+	if entry == nil {
+		return fmt.Errorf("got a nil entry, this should not happen and is potentially a bug")
+	}
+
+	logFields := []zap.Field{
+		zap.Any("error", err),
+		zap.Any("action", t.OnError),
+		zap.Any("entry.timestamp", entry.Timestamp),
+	}
+	for attrName, attrValue := range entry.Attributes {
+		logFields = append(logFields, zap.Any(attrName, attrValue))
+	}
+
 	if t.OnError == SendOnErrorQuiet || t.OnError == DropOnErrorQuiet {
-		t.Logger().Debug("Failed to process entry", zap.Any("error", err), zap.Any("action", t.OnError))
+		t.Logger().Debug("Failed to process entry", logFields...)
 	} else {
-		t.Logger().Error("Failed to process entry", zap.Any("error", err), zap.Any("action", t.OnError))
+		t.Logger().Error("Failed to process entry", logFields...)
 	}
 	if t.OnError == SendOnError || t.OnError == SendOnErrorQuiet {
 		writeErr := t.Write(ctx, entry)

pkg/stanza/operator/helper/transformer_test.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -17,6 +18,7 @@ import (
 	"go.uber.org/zap/zaptest/observer"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/attrs"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
 )
@@ -91,6 +93,9 @@ func TestTransformerDropOnError(t *testing.T) {
 	}
 	ctx := context.Background()
 	testEntry := entry.New()
+	now := time.Now()
+	testEntry.Timestamp = now
+	testEntry.AddAttribute(attrs.LogFilePath, "/test/file")
 	transform := func(_ *entry.Entry) error {
 		return fmt.Errorf("Failure")
 	}
@@ -104,8 +109,10 @@ func TestTransformerDropOnError(t *testing.T) {
 		{
 			Entry: zapcore.Entry{Level: zap.ErrorLevel, Message: "Failed to process entry"},
 			Context: []zapcore.Field{
-				{Key: "error", Type: 26, Interface: fmt.Errorf("Failure")},
+				{Key: "error", Type: zapcore.ErrorType, Interface: fmt.Errorf("Failure")},
 				zap.Any("action", "drop"),
+				zap.Any("entry.timestamp", now),
+				zap.Any(attrs.LogFilePath, "/test/file"),
 			},
 		},
 	}
@@ -136,6 +143,9 @@ func TestTransformerDropOnErrorQuiet(t *testing.T) {
 	}
 	ctx := context.Background()
 	testEntry := entry.New()
+	now := time.Now()
+	testEntry.Timestamp = now
+	testEntry.AddAttribute(attrs.LogFilePath, "/test/file")
 	transform := func(_ *entry.Entry) error {
 		return fmt.Errorf("Failure")
 	}
@@ -151,6 +161,8 @@ func TestTransformerDropOnErrorQuiet(t *testing.T) {
 			Context: []zapcore.Field{
 				{Key: "error", Type: 26, Interface: fmt.Errorf("Failure")},
 				zap.Any("action", "drop_quiet"),
+				zap.Any("entry.timestamp", now),
+				zap.Any(attrs.LogFilePath, "/test/file"),
 			},
 		},
 	}
@@ -181,6 +193,9 @@ func TestTransformerSendOnError(t *testing.T) {
 	}
 	ctx := context.Background()
 	testEntry := entry.New()
+	now := time.Now()
+	testEntry.Timestamp = now
+	testEntry.AddAttribute(attrs.LogFilePath, "/test/file")
 	transform := func(_ *entry.Entry) error {
 		return fmt.Errorf("Failure")
 	}
@@ -196,6 +211,8 @@ func TestTransformerSendOnError(t *testing.T) {
 			Context: []zapcore.Field{
 				{Key: "error", Type: 26, Interface: fmt.Errorf("Failure")},
 				zap.Any("action", "send"),
+				zap.Any("entry.timestamp", now),
+				zap.Any(attrs.LogFilePath, "/test/file"),
 			},
 		},
 	}
@@ -226,6 +243,9 @@ func TestTransformerSendOnErrorQuiet(t *testing.T) {
 	}
 	ctx := context.Background()
 	testEntry := entry.New()
+	now := time.Now()
+	testEntry.Timestamp = now
+	testEntry.AddAttribute(attrs.LogFilePath, "/test/file")
 	transform := func(_ *entry.Entry) error {
 		return fmt.Errorf("Failure")
 	}
@@ -241,6 +261,8 @@ func TestTransformerSendOnErrorQuiet(t *testing.T) {
 			Context: []zapcore.Field{
 				{Key: "error", Type: 26, Interface: fmt.Errorf("Failure")},
 				zap.Any("action", "send_quiet"),
+				zap.Any("entry.timestamp", now),
+				zap.Any(attrs.LogFilePath, "/test/file"),
 			},
 		},
 	}

pkg/stanza/operator/transformer/router/transformer.go

@@ -36,27 +36,41 @@ func (t *Transformer) CanProcess() bool {
 
 // Process will route incoming entries based on matching expressions
 func (t *Transformer) Process(ctx context.Context, entry *entry.Entry) error {
+	if entry == nil {
+		return fmt.Errorf("got a nil entry, this should not happen and is potentially a bug")
+	}
+
 	env := helper.GetExprEnv(entry)
 	defer helper.PutExprEnv(env)
 
+	logFields := []zap.Field{
+		zap.Any("entry.timestamp", entry.Timestamp),
+	}
+	for attrName, attrValue := range entry.Attributes {
+		logFields = append(logFields, zap.Any(attrName, attrValue))
+	}
+
 	for _, route := range t.routes {
 		matches, err := vm.Run(route.Expression, env)
 		if err != nil {
-			t.Logger().Warn("Running expression returned an error", zap.Error(err))
+			logFields = append(logFields, zap.Any("error", err))
+			t.Logger().Warn("Running expression returned an error", logFields...)
 			continue
 		}
 
 		// we compile the expression with "AsBool", so this should be safe
 		if matches.(bool) {
 			if err = route.Attribute(entry); err != nil {
-				t.Logger().Error("Failed to label entry", zap.Error(err))
+				logFields = append(logFields, zap.Any("error", err))
+				t.Logger().Error("Failed to label entry", logFields...)
 				return err
 			}
 
 			for _, output := range route.OutputOperators {
 				err = output.Process(ctx, entry)
+				logFields = append(logFields, zap.Any("error", err))
 				if err != nil {
-					t.Logger().Error("Failed to process entry", zap.Error(err))
+					t.Logger().Error("Failed to process entry", logFields...)
 				}
 			}
 			break