jason810496 commented on code in PR #71882:
URL: https://github.com/apache/airflow/pull/71882#discussion_r3838803389
##########
go-sdk/pkg/execution/logger.go:
##########
@@ -75,20 +87,121 @@ 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
+ }
+ return newSocketLogHandler(
+ writer,
+ newLogLevelFilter(defaultLevel,
parseNamespaceLogLevels(os.Getenv(namespaceLevelsEnv))),
+ )
+}
+
+func newSocketLogHandler(writer io.Writer, filter *logLevelFilter)
*SocketLogHandler {
shared := &socketLogHandlerShared{}
if writer != nil {
shared.writer = writer
shared.connected = true
}
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":
+ return slog.LevelError, true
+ case "CRITICAL", "FATAL":
+ return criticalLogLevel, true
+ default:
+ return 0, false
+ }
+}
+
Review Comment:
Addressed in f9f1be3a61, good catch.
##########
go-sdk/pkg/execution/logger_test.go:
##########
@@ -64,7 +65,126 @@ 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,
+ "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) {
+ assert.Equal(t, map[string]slog.Level{
+ "example": slog.LevelWarn,
+ "example.detail": slog.LevelDebug,
+ }, parseNamespaceLogLevels(
+ "example=INFO, malformed example.detail=DEBUG example=WARNING
=ERROR empty= unknown=VERBOSE",
+ ))
+}
+
+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.noisy.child").Info("namespace filtered")
+ logger.WithGroup("example.noisy.child").Warn("namespace warning")
+ logger.WithGroup("unrelated").Warn("unrelated filtered")
+ logger.Log(context.Background(), slog.LevelDebug+1, "unsupported level")
Review Comment:
Addressed in 4b109d02d9, thanks.
##########
go-sdk/pkg/execution/logger.go:
##########
@@ -75,20 +87,121 @@ 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
+ }
+ return newSocketLogHandler(
+ writer,
+ newLogLevelFilter(defaultLevel,
parseNamespaceLogLevels(os.Getenv(namespaceLevelsEnv))),
+ )
+}
+
+func newSocketLogHandler(writer io.Writer, filter *logLevelFilter)
*SocketLogHandler {
shared := &socketLogHandlerShared{}
if writer != nil {
shared.writer = writer
shared.connected = true
}
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":
+ 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 {
+ levels := make(map[string]slog.Level)
+ 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 {
+ continue
+ }
Review Comment:
Agreed, Java-SDK handles the malformed as well. Addressed in cfcf8f91a8.
##########
go-sdk/pkg/execution/logger.go:
##########
@@ -134,6 +251,11 @@ func (h *SocketLogHandler) Handle(_ context.Context, r
slog.Record) error {
return true
})
+ if loggerName != "" {
+ entry["logger"] = loggerName
+ }
+ entry["level"] = getAirflowLogLevelName(r.Level)
Review Comment:
Added test to cover it in 87a758b8ce.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
