[pkg/stanza] Improve error logs produced by transformer processors

.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 log file path and original record 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 the log file path and the entry's body in its own logs.
+  This makes it easier to debug why certain log entries are causing issues in the processors. 
+
+# 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/fileconsumer/attrs/attrs.go

@@ -18,6 +18,7 @@ const (
 	LogFileOwnerName      = "log.file.owner.name"
 	LogFileOwnerGroupName = "log.file.owner.group.name"
 	LogFileRecordNumber   = "log.file.record_number"
+	LogRecordOriginal     = "log.record.original"
 )
 
 type Resolver struct {

pkg/stanza/operator/helper/transformer.go

@@ -6,13 +6,15 @@ package helper // import "github.com/open-telemetry/opentelemetry-collector-cont
 import (
 	"context"
 	"fmt"
+	"slices"
 
 	"github.com/expr-lang/expr/vm"
 	"go.opentelemetry.io/collector/component"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/errors"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/attrs"
 )
 
 // NewTransformerConfig creates a new transformer config with default values
@@ -94,10 +96,22 @@ 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 {
+	logFields := []zap.Field{
+		zap.Any("error", err),
+		zap.Any("action", t.OnError),
+	}
+	if entry != nil {
+		toAddFields := []zap.Field{zap.Any(attrs.LogRecordOriginal, entry.Body)}
+		if logFilePath, ok := entry.Attributes[attrs.LogFilePath]; ok {
+			toAddFields = slices.Insert(toAddFields, 0, zap.Any(attrs.LogFilePath, logFilePath))
+		}
+		logFields = slices.Insert(logFields, 0, toAddFields...)
+	}
+
 	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

@@ -17,6 +17,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 +92,8 @@ func TestTransformerDropOnError(t *testing.T) {
 	}
 	ctx := context.Background()
 	testEntry := entry.New()
+	testEntry.Body = "test"
+	testEntry.AddAttribute(attrs.LogFilePath, "/test/file")
 	transform := func(_ *entry.Entry) error {
 		return fmt.Errorf("Failure")
 	}
@@ -104,7 +107,9 @@ 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")},
+				zap.Any(attrs.LogFilePath, "/test/file"),
+				zap.Any(attrs.LogRecordOriginal, "test"),
+				{Key: "error", Type: zapcore.ErrorType, Interface: fmt.Errorf("Failure")},
 				zap.Any("action", "drop"),
 			},
 		},
@@ -136,6 +141,8 @@ func TestTransformerDropOnErrorQuiet(t *testing.T) {
 	}
 	ctx := context.Background()
 	testEntry := entry.New()
+	testEntry.Body = "test"
+	testEntry.AddAttribute(attrs.LogFilePath, "/test/file")
 	transform := func(_ *entry.Entry) error {
 		return fmt.Errorf("Failure")
 	}
@@ -149,6 +156,8 @@ func TestTransformerDropOnErrorQuiet(t *testing.T) {
 		{
 			Entry: zapcore.Entry{Level: zap.DebugLevel, Message: "Failed to process entry"},
 			Context: []zapcore.Field{
+				zap.Any(attrs.LogFilePath, "/test/file"),
+				zap.Any(attrs.LogRecordOriginal, "test"),
 				{Key: "error", Type: 26, Interface: fmt.Errorf("Failure")},
 				zap.Any("action", "drop_quiet"),
 			},
@@ -181,6 +190,8 @@ func TestTransformerSendOnError(t *testing.T) {
 	}
 	ctx := context.Background()
 	testEntry := entry.New()
+	testEntry.Body = "test"
+	testEntry.AddAttribute(attrs.LogFilePath, "/test/file")
 	transform := func(_ *entry.Entry) error {
 		return fmt.Errorf("Failure")
 	}
@@ -194,6 +205,8 @@ func TestTransformerSendOnError(t *testing.T) {
 		{
 			Entry: zapcore.Entry{Level: zap.ErrorLevel, Message: "Failed to process entry"},
 			Context: []zapcore.Field{
+				zap.Any(attrs.LogFilePath, "/test/file"),
+				zap.Any(attrs.LogRecordOriginal, "test"),
 				{Key: "error", Type: 26, Interface: fmt.Errorf("Failure")},
 				zap.Any("action", "send"),
 			},
@@ -226,6 +239,8 @@ func TestTransformerSendOnErrorQuiet(t *testing.T) {
 	}
 	ctx := context.Background()
 	testEntry := entry.New()
+	testEntry.Body = "test"
+	testEntry.AddAttribute(attrs.LogFilePath, "/test/file")
 	transform := func(_ *entry.Entry) error {
 		return fmt.Errorf("Failure")
 	}
@@ -239,6 +254,8 @@ func TestTransformerSendOnErrorQuiet(t *testing.T) {
 		{
 			Entry: zapcore.Entry{Level: zap.DebugLevel, Message: "Failed to process entry"},
 			Context: []zapcore.Field{
+				zap.Any(attrs.LogFilePath, "/test/file"),
+				zap.Any(attrs.LogRecordOriginal, "test"),
 				{Key: "error", Type: 26, Interface: fmt.Errorf("Failure")},
 				zap.Any("action", "send_quiet"),
 			},

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

@@ -6,11 +6,13 @@ package router // import "github.com/open-telemetry/opentelemetry-collector-cont
 import (
 	"context"
 	"fmt"
+	"slices"
 
 	"github.com/expr-lang/expr/vm"
 	"go.uber.org/zap"
 
 	"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/operator/helper"
 )
@@ -39,24 +41,33 @@ func (t *Transformer) Process(ctx context.Context, entry *entry.Entry) error {
 	env := helper.GetExprEnv(entry)
 	defer helper.PutExprEnv(env)
 
+	logger := t.Logger()
+	if entry != nil {
+		toAddFields := []zap.Field{zap.Any(attrs.LogRecordOriginal, entry.Body)}
+		if logFilePath, ok := entry.Attributes[attrs.LogFilePath]; ok {
+			toAddFields = slices.Insert(toAddFields, 0, zap.Any(attrs.LogFilePath, logFilePath))
+		}
+		logger = logger.With(toAddFields...)
+	}
+
 	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))
+			logger.Warn("Running expression returned an error", zap.Error(err))
 			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))
+				logger.Error("Failed to label entry", zap.Error(err))
 				return err
 			}
 
 			for _, output := range route.OutputOperators {
 				err = output.Process(ctx, entry)
 				if err != nil {
-					t.Logger().Error("Failed to process entry", zap.Error(err))
+					logger.Error("Failed to process entry", zap.Error(err))
 				}
 			}
 			break