[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 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