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 577496acc3b Honor Airflow log levels in Go task logging (#71882)
577496acc3b is described below
commit 577496acc3b99767c24e4146e02f1a0c308d6a95
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Sun Sep 6 21:05:46 2026 -0500
Honor Airflow log levels in Go task logging (#71882)
* Honor Airflow log levels in Go task logging
Coordinator-mode Go tasks currently transmit records below Airflow's
configured thresholds, diverging from Python task behavior and overstating
logging compatibility.
* Normalize custom Go task log levels
Custom slog levels should remain visible in task logs instead of being
silently discarded when they do not exactly match an Airflow severity.
* Use slog groups for Go task log namespaces
Group-scoped loggers provide an idiomatic namespace that allows Airflow to
apply namespace thresholds before constructing each record.
* Align Go namespace log levels with Python tasks
Shared namespace logging configuration should filter Go and Python task
records consistently.
* Report invalid Go namespace log levels
Configuration typos must be visible instead of silently changing which Go
task records reach Airflow.
* Verify custom Go task log level normalization
Filtering behavior alone cannot protect the level name that Airflow
receives for custom slog levels.
* Protect Go task log metadata from user attributes
Airflow's level and logger fields must remain authoritative when task code
uses the same attribute names.
---
go-sdk/README.md | 11 +++
go-sdk/pkg/execution/integration_test.go | 69 ++++++++++++++
go-sdk/pkg/execution/logger.go | 146 ++++++++++++++++++++++++++--
go-sdk/pkg/execution/logger_test.go | 158 ++++++++++++++++++++++++++++++-
go-sdk/pkg/execution/server.go | 2 +-
5 files changed, 378 insertions(+), 8 deletions(-)
diff --git a/go-sdk/README.md b/go-sdk/README.md
index 360e185a9a5..8fb6564c9e7 100644
--- a/go-sdk/README.md
+++ b/go-sdk/README.md
@@ -194,6 +194,17 @@ func extract(ctx sdk.TIRunContext, log *slog.Logger) (any,
error) {
`TryNumber`; `ctx.DagRun()` returns `DagID`, `RunID`, and the `*time.Time`
fields `LogicalDate`,
`DataIntervalStart`, and `DataIntervalEnd` (nil when the run has no such
value, e.g. a manual trigger).
+### Task logging
+
+In coordinator mode, the injected logger filters records using Airflow's
configured `[logging] logging_level` before sending them to the supervisor.
Airflow also propagates `[logging] namespace_levels`; use a group-scoped logger
to set the namespace:
+
+```go
+databaseLog := log.WithGroup("example.database")
+databaseLog.Debug("query complete", "rows", 42)
+```
+
+Namespace levels use longest dotted-prefix matching, so an `example=DEBUG`
override also applies to `example.database` unless a more specific override
takes precedence.
+
## Deployment
A Python task runner executes the Go task directly, with no separate Go worker
process to run on the host.
diff --git a/go-sdk/pkg/execution/integration_test.go
b/go-sdk/pkg/execution/integration_test.go
index da6b10784fe..482165d887d 100644
--- a/go-sdk/pkg/execution/integration_test.go
+++ b/go-sdk/pkg/execution/integration_test.go
@@ -20,10 +20,12 @@ package execution
import (
"bytes"
"context"
+ "encoding/json"
"errors"
"io"
"log/slog"
"net"
+ "strings"
"testing"
"time"
@@ -660,6 +662,73 @@ func TestServeStartupDetailsEndToEnd(t *testing.T) {
}
}
+func TestServeUsesSupervisorLogLevelEnvironment(t *testing.T) {
+ t.Setenv(loggingLevelEnv, "ERROR")
+ t.Setenv(namespaceLevelsEnv, "example=DEBUG")
+
+ commAddr, logsAddr, commCh, logsCh, cleanup := startSupervisor(t)
+ defer cleanup()
+
+ provider := &fakeProvider{
+ register: func(r bundlev1.Registry) error {
+ r.AddDag("dag1").AddTaskWithName("logging", func(logger
*slog.Logger) error {
+ logger.Info("global filtered")
+
logger.WithGroup("example.child").Debug("namespace debug")
+ logger.WithGroup("unrelated").Warn("unrelated
filtered")
+ logger.Error("global error")
+ return nil
+ })
+ return nil
+ },
+ }
+
+ done := make(chan error, 1)
+ go func() { done <- Serve(provider, commAddr, logsAddr) }()
+
+ commConn := <-commCh
+ defer commConn.Close()
+ logsConn := <-logsCh
+ defer logsConn.Close()
+ require.NoError(t, commConn.SetDeadline(time.Now().Add(10*time.Second)))
+ require.NoError(t, logsConn.SetDeadline(time.Now().Add(10*time.Second)))
+
+ payload, err := encodeRequest(0, map[string]any{
+ "type": "StartupDetails",
+ "ti": map[string]any{
+ "id": "550e8400-e29b-41d4-a716-446655440000",
+ "dag_id": "dag1",
+ "task_id": "logging",
+ "run_id": "run1",
+ "try_number": 1,
+ },
+ "bundle_info": map[string]any{"name": "fake", "version": "1.0"},
+ })
+ require.NoError(t, err)
+ require.NoError(t, writeFrame(commConn, payload))
+
+ frame, err := readFrame(commConn)
+ require.NoError(t, err)
+ require.True(t, isNilRaw(frame.Err))
+ assert.Equal(t, "SucceedTask", peekBodyType(frame.Body))
+
+ select {
+ case err := <-done:
+ require.NoError(t, err)
+ case <-time.After(2 * time.Second):
+ t.Fatal("Serve did not return after task completion")
+ }
+
+ output, err := io.ReadAll(logsConn)
+ require.NoError(t, err)
+ var events []string
+ for line := range strings.Lines(string(output)) {
+ var entry map[string]any
+ require.NoError(t, json.Unmarshal([]byte(line), &entry))
+ events = append(events, entry["event"].(string))
+ }
+ assert.Equal(t, []string{"namespace debug", "global error"}, events)
+}
+
// TestServeClientRoundTripEndToEnd drives a task that calls back into the
// supervisor mid-execution, so the comm dispatcher's request/response
// multiplexing is exercised against the real Serve rather than only the
diff --git a/go-sdk/pkg/execution/logger.go b/go-sdk/pkg/execution/logger.go
index 05de45c01f7..bb554523c40 100644
--- a/go-sdk/pkg/execution/logger.go
+++ b/go-sdk/pkg/execution/logger.go
@@ -22,9 +22,21 @@ import (
"encoding/json"
"io"
"log/slog"
+ "os"
"strings"
"sync"
"time"
+ "unicode"
+)
+
+const (
+ loggingLevelEnv = "AIRFLOW__LOGGING__LOGGING_LEVEL"
+ namespaceLevelsEnv = "AIRFLOW__LOGGING__NAMESPACE_LEVELS"
+
+ // NOTSET sits below every supported record level. CRITICAL sits above
ERROR
+ // because slog has no built-in equivalents for either Airflow level.
+ notsetLogLevel = slog.Level(-100)
+ criticalLogLevel = slog.LevelError + 4
)
// SocketLogHandler is an slog.Handler that streams structured JSON log lines
@@ -44,7 +56,7 @@ import (
// updating the supervisor side in lockstep.
type SocketLogHandler struct {
shared *socketLogHandlerShared
- level slog.Level
+ filter *logLevelFilter
// attrs is the list of attributes accumulated via WithAttrs. Each
entry's
// key has already been qualified with whatever groups were active at
the
// WithAttrs call site, so a later WithGroup does NOT retroactively
prefix
@@ -75,9 +87,35 @@ type socketLogHandlerShared struct {
var _ slog.Handler = (*SocketLogHandler)(nil)
+type logLevelFilter struct {
+ defaultLevel slog.Level
+ namespaceLevels map[string]slog.Level
+}
+
// NewSocketLogHandler creates a new handler. If writer is nil, messages are
// buffered until Connect() is called.
func NewSocketLogHandler(writer io.Writer, level slog.Level) *SocketLogHandler
{
+ return newSocketLogHandler(writer, newLogLevelFilter(level, nil))
+}
+
+func newSocketLogHandlerFromEnv(writer io.Writer) *SocketLogHandler {
+ defaultLevel, ok := parseLogLevel(os.Getenv(loggingLevelEnv))
+ if !ok {
+ defaultLevel = slog.LevelInfo
+ }
+ namespaceLevels, invalidEntries :=
parseNamespaceLogLevels(os.Getenv(namespaceLevelsEnv))
+ handler := newSocketLogHandler(
+ writer,
+ newLogLevelFilter(defaultLevel, namespaceLevels),
+ )
+ logger := slog.New(handler)
+ for _, invalidEntry := range invalidEntries {
+ logger.Error("Ignoring invalid namespace_levels entry",
"entry", invalidEntry)
+ }
+ return handler
+}
+
+func newSocketLogHandler(writer io.Writer, filter *logLevelFilter)
*SocketLogHandler {
shared := &socketLogHandlerShared{}
if writer != nil {
shared.writer = writer
@@ -85,10 +123,93 @@ func NewSocketLogHandler(writer io.Writer, level
slog.Level) *SocketLogHandler {
}
return &SocketLogHandler{
shared: shared,
- level: level,
+ filter: filter,
+ }
+}
+
+func newLogLevelFilter(
+ defaultLevel slog.Level,
+ namespaceLevels map[string]slog.Level,
+) *logLevelFilter {
+ return &logLevelFilter{
+ defaultLevel: defaultLevel,
+ namespaceLevels: namespaceLevels,
+ }
+}
+
+func parseLogLevel(value string) (slog.Level, bool) {
+ switch strings.ToUpper(strings.TrimSpace(value)) {
+ case "NOTSET":
+ return notsetLogLevel, true
+ case "DEBUG":
+ return slog.LevelDebug, true
+ case "INFO":
+ return slog.LevelInfo, true
+ case "WARN", "WARNING":
+ return slog.LevelWarn, true
+ case "ERROR", "EXCEPTION":
+ return slog.LevelError, true
+ case "CRITICAL", "FATAL":
+ return criticalLogLevel, true
+ default:
+ return 0, false
+ }
+}
+
+func getAirflowLogLevelName(level slog.Level) string {
+ switch {
+ case level < slog.LevelDebug:
+ return "notset"
+ case level < slog.LevelInfo:
+ return "debug"
+ case level < slog.LevelWarn:
+ return "info"
+ case level < slog.LevelError:
+ return "warning"
+ case level < criticalLogLevel:
+ return "error"
+ default:
+ return "critical"
}
}
+func parseNamespaceLogLevels(value string) (map[string]slog.Level, []string) {
+ levels := make(map[string]slog.Level)
+ var invalidEntries []string
+ entries := strings.FieldsFunc(value, func(r rune) bool {
+ return r == ',' || unicode.IsSpace(r)
+ })
+ for _, entry := range entries {
+ loggerName, levelName, ok := strings.Cut(entry, "=")
+ loggerName = strings.TrimSpace(loggerName)
+ level, validLevel := parseLogLevel(levelName)
+ if !ok || loggerName == "" || !validLevel {
+ invalidEntries = append(invalidEntries, entry)
+ continue
+ }
+ levels[loggerName] = level
+ }
+ return levels, invalidEntries
+}
+
+func (f *logLevelFilter) getLevel(loggerName string) slog.Level {
+ for namespace := loggerName; namespace != ""; {
+ if level, ok := f.namespaceLevels[namespace]; ok {
+ return level
+ }
+ separator := strings.LastIndexByte(namespace, '.')
+ if separator == -1 {
+ break
+ }
+ namespace = namespace[:separator]
+ }
+ return f.defaultLevel
+}
+
+func (f *logLevelFilter) isEnabled(loggerName string, level slog.Level) bool {
+ return level >= f.getLevel(loggerName)
+}
+
// Connect sets the writer and flushes any buffered log messages.
func (h *SocketLogHandler) Connect(w io.Writer) {
h.shared.mu.Lock()
@@ -104,15 +225,19 @@ func (h *SocketLogHandler) Connect(w io.Writer) {
}
func (h *SocketLogHandler) Enabled(_ context.Context, level slog.Level) bool {
- return level >= h.level
+ return h.filter.isEnabled(h.getLoggerName(), level)
}
func (h *SocketLogHandler) Handle(_ context.Context, r slog.Record) error {
+ loggerName := h.getLoggerName()
+ if !h.filter.isEnabled(loggerName, r.Level) {
+ return nil
+ }
+
entry := make(map[string]any)
// Set standard fields.
entry["event"] = r.Message
- entry["level"] = strings.ToLower(r.Level.String())
if !r.Time.IsZero() {
entry["timestamp"] = r.Time.Format(time.RFC3339Nano)
}
@@ -134,6 +259,11 @@ func (h *SocketLogHandler) Handle(_ context.Context, r
slog.Record) error {
return true
})
+ if loggerName != "" {
+ entry["logger"] = loggerName
+ }
+ entry["level"] = getAirflowLogLevelName(r.Level)
+
line, err := json.Marshal(entry)
if err != nil {
return err
@@ -170,7 +300,7 @@ func (h *SocketLogHandler) WithAttrs(attrs []slog.Attr)
slog.Handler {
}
return &SocketLogHandler{
shared: h.shared,
- level: h.level,
+ filter: h.filter,
attrs: newAttrs,
groups: h.groups,
}
@@ -182,12 +312,16 @@ func (h *SocketLogHandler) WithGroup(name string)
slog.Handler {
}
return &SocketLogHandler{
shared: h.shared,
- level: h.level,
+ filter: h.filter,
attrs: h.attrs,
groups: append(append([]string{}, h.groups...), name),
}
}
+func (h *SocketLogHandler) getLoggerName() string {
+ return strings.Join(h.groups, ".")
+}
+
// groupPrefix returns the dotted prefix (including trailing ".") to apply to
// attribute keys for the currently-active groups, or "" if no group is
// active.
diff --git a/go-sdk/pkg/execution/logger_test.go
b/go-sdk/pkg/execution/logger_test.go
index ee39fef69f0..19a67ad59dd 100644
--- a/go-sdk/pkg/execution/logger_test.go
+++ b/go-sdk/pkg/execution/logger_test.go
@@ -19,6 +19,7 @@ package execution
import (
"bytes"
+ "context"
"encoding/json"
"errors"
"log/slog"
@@ -64,7 +65,146 @@ func TestSocketLogHandlerLevelFiltering(t *testing.T) {
var entry map[string]any
require.NoError(t, json.Unmarshal([]byte(lines[0]), &entry))
assert.Equal(t, "should appear", entry["event"])
- assert.Equal(t, "warn", entry["level"])
+ assert.Equal(t, "warning", entry["level"])
+}
+
+func TestParseLogLevel(t *testing.T) {
+ tests := map[string]slog.Level{
+ "notset": notsetLogLevel,
+ "DEBUG": slog.LevelDebug,
+ " info ": slog.LevelInfo,
+ "WARN": slog.LevelWarn,
+ "warning": slog.LevelWarn,
+ "ERROR": slog.LevelError,
+ "exception": slog.LevelError,
+ "critical": criticalLogLevel,
+ "FATAL": criticalLogLevel,
+ }
+ for value, expected := range tests {
+ level, ok := parseLogLevel(value)
+ assert.True(t, ok, value)
+ assert.Equal(t, expected, level, value)
+ }
+
+ _, ok := parseLogLevel("verbose")
+ assert.False(t, ok)
+}
+
+func TestGetAirflowLogLevelName(t *testing.T) {
+ tests := map[slog.Level]string{
+ notsetLogLevel: "notset",
+ slog.LevelDebug - 1: "notset",
+ slog.LevelDebug: "debug",
+ slog.LevelDebug + 1: "debug",
+ slog.LevelInfo - 1: "debug",
+ slog.LevelInfo: "info",
+ slog.LevelInfo + 1: "info",
+ slog.LevelWarn - 1: "info",
+ slog.LevelWarn: "warning",
+ slog.LevelWarn + 1: "warning",
+ slog.LevelError - 1: "warning",
+ slog.LevelError: "error",
+ slog.LevelError + 1: "error",
+ criticalLogLevel - 1: "error",
+ criticalLogLevel: "critical",
+ criticalLogLevel + 1: "critical",
+ }
+ for level, expected := range tests {
+ assert.Equal(t, expected, getAirflowLogLevelName(level), level)
+ }
+}
+
+func TestParseNamespaceLogLevels(t *testing.T) {
+ levels, invalidEntries := parseNamespaceLogLevels(
+ "example=INFO, malformed example.detail=DEBUG example=WARNING
=ERROR empty= unknown=VERBOSE",
+ )
+ assert.Equal(t, map[string]slog.Level{
+ "example": slog.LevelWarn,
+ "example.detail": slog.LevelDebug,
+ }, levels)
+ assert.Equal(t, []string{"malformed", "=ERROR", "empty=",
"unknown=VERBOSE"}, invalidEntries)
+}
+
+func TestSocketLogHandlerReportsInvalidNamespaceLevels(t *testing.T) {
+ t.Setenv(loggingLevelEnv, "INFO")
+ invalidEntry := "example=DEB" + "GU"
+ t.Setenv(namespaceLevelsEnv, invalidEntry)
+
+ var buf bytes.Buffer
+ newSocketLogHandlerFromEnv(&buf)
+
+ var entry map[string]any
+ require.NoError(t,
json.Unmarshal([]byte(strings.TrimSpace(buf.String())), &entry))
+ assert.Equal(t, "Ignoring invalid namespace_levels entry",
entry["event"])
+ assert.Equal(t, invalidEntry, entry["entry"])
+ assert.Equal(t, "error", entry["level"])
+}
+
+func TestLogLevelFilterUsesLongestNamespacePrefix(t *testing.T) {
+ filter := newLogLevelFilter(slog.LevelError, map[string]slog.Level{
+ "example": slog.LevelInfo,
+ "example.detail": slog.LevelDebug,
+ })
+
+ assert.Equal(t, slog.LevelDebug,
filter.getLevel("example.detail.child"))
+ assert.Equal(t, slog.LevelInfo, filter.getLevel("example.other"))
+ assert.Equal(t, slog.LevelError, filter.getLevel("exampled"))
+ assert.Equal(t, slog.LevelError, filter.getLevel(""))
+}
+
+func TestSocketLogHandlerUsesEnvironmentLevelFiltering(t *testing.T) {
+ t.Setenv(loggingLevelEnv, "ERROR")
+ t.Setenv(namespaceLevelsEnv, "example=DEBUG, example.noisy=WARNING")
+
+ var buf bytes.Buffer
+ logger := slog.New(newSocketLogHandlerFromEnv(&buf))
+ logger.Info("global filtered")
+ logger.WithGroup("example.detail").Debug("namespace debug")
+ logger.WithGroup("example.detail").Log(context.Background(),
slog.LevelDebug+1, "custom level")
+ logger.WithGroup("example.noisy.child").Info("namespace filtered")
+ logger.WithGroup("example.noisy.child").Warn("namespace warning")
+ logger.WithGroup("unrelated").Warn("unrelated filtered")
+ logger.Error("global error")
+
+ var entries []map[string]any
+ for line := range strings.Lines(buf.String()) {
+ var entry map[string]any
+ require.NoError(t, json.Unmarshal([]byte(line), &entry))
+ entries = append(entries, entry)
+ }
+ require.Len(t, entries, 4)
+ assert.Equal(t, "namespace debug", entries[0]["event"])
+ assert.Equal(t, "example.detail", entries[0]["logger"])
+ assert.Equal(t, "custom level", entries[1]["event"])
+ assert.Equal(t, "debug", entries[1]["level"])
+ assert.Equal(t, "namespace warning", entries[2]["event"])
+ assert.Equal(t, "example.noisy.child", entries[2]["logger"])
+ assert.Equal(t, "global error", entries[3]["event"])
+ assert.NotContains(t, entries[3], "logger")
+}
+
+func TestSocketLogHandlerUsesGroupNamespaceInEnabled(t *testing.T) {
+ filter := newLogLevelFilter(slog.LevelError,
map[string]slog.Level{"example": slog.LevelDebug})
+ handler := newSocketLogHandler(nil, filter)
+
+ assert.True(
+ t,
+
handler.WithGroup("example.detail").Enabled(context.Background(),
slog.LevelDebug),
+ )
+ assert.False(t,
handler.WithGroup("unrelated").Enabled(context.Background(), slog.LevelWarn))
+}
+
+func TestSocketLogHandlerDefaultsInvalidEnvironmentLevelToInfo(t *testing.T) {
+ t.Setenv(loggingLevelEnv, "VERBOSE")
+ t.Setenv(namespaceLevelsEnv, "")
+
+ var buf bytes.Buffer
+ logger := slog.New(newSocketLogHandlerFromEnv(&buf))
+ logger.Debug("filtered")
+ logger.Info("included")
+
+ assert.Contains(t, buf.String(), `"event":"included"`)
+ assert.NotContains(t, buf.String(), "filtered")
}
func TestSocketLogHandlerBuffering(t *testing.T) {
@@ -111,6 +251,7 @@ func TestSocketLogHandlerWithGroup(t *testing.T) {
var entry map[string]any
require.NoError(t,
json.Unmarshal([]byte(strings.TrimSpace(buf.String())), &entry))
assert.Equal(t, "val", entry["grp.key"])
+ assert.Equal(t, "grp", entry["logger"])
}
// TestSocketLogHandlerInlineGroup verifies that an inline slog.Group passed as
@@ -201,6 +342,21 @@ func TestSocketLogHandlerKeyMapping(t *testing.T) {
assert.False(t, hasTime)
}
+func TestSocketLogHandlerStandardFieldsOverrideAttrs(t *testing.T) {
+ var buf bytes.Buffer
+ handler := NewSocketLogHandler(&buf, slog.LevelDebug)
+ logger := slog.New(handler).
+ With("level", "custom", "logger", "custom").
+ WithGroup("example")
+
+ logger.Info("message")
+
+ var entry map[string]any
+ require.NoError(t,
json.Unmarshal([]byte(strings.TrimSpace(buf.String())), &entry))
+ assert.Equal(t, "info", entry["level"])
+ assert.Equal(t, "example", entry["logger"])
+}
+
// TestSocketLogHandlerStringifiesErrorAttr verifies that an attribute whose
// value is a stock `error` is rendered as its .Error() string. Without value
// resolution the underlying struct's unexported fields would be marshaled as
diff --git a/go-sdk/pkg/execution/server.go b/go-sdk/pkg/execution/server.go
index d4a4d43f157..a69c2303db8 100644
--- a/go-sdk/pkg/execution/server.go
+++ b/go-sdk/pkg/execution/server.go
@@ -92,7 +92,7 @@ func Serve(provider bundlev1.BundleProvider, commAddr,
logsAddr string) error {
// Buffer log records until the logs socket is connected. Anything the
// runtime emits between Connect-time and the first frame still gets
// flushed.
- logHandler := NewSocketLogHandler(nil, slog.LevelDebug)
+ logHandler := newSocketLogHandlerFromEnv(nil)
logger := slog.New(logHandler)
slog.SetDefault(logger)
