This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 9761167313d Go SDK: stop log attributes from displacing standard log
fields (#73820)
9761167313d is described below
commit 9761167313d84d1367f3bb4735aa88242ad4c956
Author: Henry Chen <[email protected]>
AuthorDate: Sun Oct 25 00:05:32 2026 +0800
Go SDK: stop log attributes from displacing standard log fields (#73820)
* Go SDK: stop log attributes from displacing standard log fields
A user attribute whose key collides with a standard field replaced that
field instead of losing to it. "event" is the one that hurts: the
supervisor takes a task's log line out of that key, so a Dag logging an
attribute called "event" lost its message entirely, with nothing left in
the line to show the message had ever existed. "timestamp" was displaced
the same way, leaving the supervisor a value it cannot decode as a
datetime.
"level" and "logger" were already immune, for no reason other than being
written after the attributes rather than before them.
* Move a clashing log attribute aside instead of overwriting it
Letting the standard field win the slot still discarded a value the caller
asked to log. Reserving the slot and re-keying the attribute to
"fields.<key>" keeps both: the supervisor finds the message in "event" and
a parsable "timestamp", and the attribute survives under a name that
collides with nothing. logrus resolves the same clash the same way.
Rejecting the attribute outright, as structlog's TypeError does, is not
available here. slog discards a handler's error, so returning one would
drop the whole line, and a panic would surface in production rather than
fail fast at the call site.
---
go-sdk/pkg/execution/logger.go | 36 ++++++++++++++++++++++++++++-------
go-sdk/pkg/execution/logger_test.go | 38 +++++++++++++++++++++++++++++++++++++
2 files changed, 67 insertions(+), 7 deletions(-)
diff --git a/go-sdk/pkg/execution/logger.go b/go-sdk/pkg/execution/logger.go
index bb554523c40..c577e3278a3 100644
--- a/go-sdk/pkg/execution/logger.go
+++ b/go-sdk/pkg/execution/logger.go
@@ -47,7 +47,8 @@ const (
// - "event" for the log message (not "msg")
// - "level" in lowercase (not "INFO"/"ERROR")
// - "timestamp" in RFC3339Nano format (not "time")
-// - Additional attributes are included as top-level fields
+// - Additional attributes are included as top-level fields; one whose key
+// collides with a standard field is moved to "fields.<key>"
//
// Groups are encoded as dotted key prefixes on a flat JSON object
// (`{"grp.key": "val"}`), not as nested objects. The Airflow supervisor's
@@ -236,12 +237,6 @@ func (h *SocketLogHandler) Handle(_ context.Context, r
slog.Record) error {
entry := make(map[string]any)
- // Set standard fields.
- entry["event"] = r.Message
- if !r.Time.IsZero() {
- entry["timestamp"] = r.Time.Format(time.RFC3339Nano)
- }
-
// Apply pre-configured attrs. Keys are already qualified with the
groups
// active at the WithAttrs call site, so the current h.groups is NOT
// applied here — only to record-level attrs below. The stored key
already
@@ -259,6 +254,12 @@ func (h *SocketLogHandler) Handle(_ context.Context, r
slog.Record) error {
return true
})
+ moveClashingAttrs(entry)
+
+ entry["event"] = r.Message
+ if !r.Time.IsZero() {
+ entry["timestamp"] = r.Time.Format(time.RFC3339Nano)
+ }
if loggerName != "" {
entry["logger"] = loggerName
}
@@ -282,6 +283,27 @@ func (h *SocketLogHandler) Handle(_ context.Context, r
slog.Record) error {
return err
}
+// standardLogFields are the keys SocketLogHandler writes itself on every line.
+// The supervisor takes a task's message out of "event" and decodes "timestamp"
+// as a datetime, so an attribute must never end up occupying those slots.
+var standardLogFields = []string{"event", "timestamp", "level", "logger"}
+
+// clashedAttrPrefix qualifies an attribute whose key collides with a standard
+// field, the way logrus does. Letting the standard field simply overwrite it
+// would discard a value the caller asked to log.
+const clashedAttrPrefix = "fields."
+
+// moveClashingAttrs re-keys any attribute sitting in a standard field's slot,
+// so that writing the standard field afterwards keeps both values.
+func moveClashingAttrs(entry map[string]any) {
+ for _, key := range standardLogFields {
+ if value, ok := entry[key]; ok {
+ entry[clashedAttrPrefix+key] = value
+ delete(entry, key)
+ }
+ }
+}
+
func (h *SocketLogHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
if len(attrs) == 0 {
return h
diff --git a/go-sdk/pkg/execution/logger_test.go
b/go-sdk/pkg/execution/logger_test.go
index 19a67ad59dd..39769e3d8f3 100644
--- a/go-sdk/pkg/execution/logger_test.go
+++ b/go-sdk/pkg/execution/logger_test.go
@@ -25,6 +25,7 @@ import (
"log/slog"
"strings"
"testing"
+ "time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -355,6 +356,43 @@ func TestSocketLogHandlerStandardFieldsOverrideAttrs(t
*testing.T) {
require.NoError(t,
json.Unmarshal([]byte(strings.TrimSpace(buf.String())), &entry))
assert.Equal(t, "info", entry["level"])
assert.Equal(t, "example", entry["logger"])
+ assert.Equal(t, "custom", entry["fields.level"])
+ assert.Equal(t, "custom", entry["fields.logger"])
+}
+
+func TestSocketLogHandlerMovesAttrsClashingWithStandardFields(t *testing.T) {
+ for _, tt := range []struct {
+ key string
+ val any
+ want any
+ }{
+ {key: "event", val: "user value", want: "user value"},
+ {key: "timestamp", val: 12345, want: float64(12345)},
+ {key: "level", val: "custom", want: "custom"},
+ {key: "logger", val: "custom", want: "custom"},
+ } {
+ t.Run(tt.key, func(t *testing.T) {
+ var buf bytes.Buffer
+ handler := NewSocketLogHandler(&buf, slog.LevelDebug)
+ logger := slog.New(handler).With(tt.key, tt.val)
+
+ logger.Info("the real message")
+
+ var entry map[string]any
+ require.NoError(t,
json.Unmarshal([]byte(strings.TrimSpace(buf.String())), &entry))
+
+ assert.Equal(t, "the real message", entry["event"])
+ assert.Equal(t, "info", entry["level"])
+
+ ts, ok := entry["timestamp"].(string)
+ require.True(t, ok, "timestamp should stay a string,
got %T", entry["timestamp"])
+ _, err := time.Parse(time.RFC3339Nano, ts)
+ assert.NoError(t, err, "timestamp should stay
RFC3339Nano")
+
+ assert.Equal(t, tt.want, entry["fields."+tt.key])
+ assert.NotEqual(t, tt.want, entry[tt.key], "reserved
slot must not keep the attr")
+ })
+ }
}
// TestSocketLogHandlerStringifiesErrorAttr verifies that an attribute whose
