From 496670e8d39ccea7fe21c668d5cb28a5f54caff1 Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Fri, 21 Aug 2026 17:14:46 +0000 Subject: [PATCH 01/11] feat(kernel): forward logs through shared logger Signed-off-by: Vu Anh Phung --- KERNEL_REV | 2 +- doc.go | 18 +-- internal/backend/kernel/cgo.go | 60 +++------- internal/backend/kernel/log_callback.go | 85 ++++++++++++++ internal/backend/kernel/log_callback_test.go | 112 +++++++++++++++++++ internal/backend/kernel/logforward.go | 44 ++++++++ internal/backend/kernel/logforward_test.go | 92 +++++++++++++++ logger/logger.go | 32 +++++- logger/logger_test.go | 66 +++++++++++ 9 files changed, 451 insertions(+), 60 deletions(-) create mode 100644 internal/backend/kernel/log_callback.go create mode 100644 internal/backend/kernel/log_callback_test.go create mode 100644 internal/backend/kernel/logforward.go create mode 100644 internal/backend/kernel/logforward_test.go create mode 100644 logger/logger_test.go diff --git a/KERNEL_REV b/KERNEL_REV index 95cfce81..04e3f21c 100644 --- a/KERNEL_REV +++ b/KERNEL_REV @@ -1 +1 @@ -eff8950428f4e6cc9975c663ec919f334962f7d0 +f7aa79c00b35abb3f8236cf723bda289ea78a3b6 diff --git a/doc.go b/doc.go index 18673400..3f0ac892 100644 --- a/doc.go +++ b/doc.go @@ -195,19 +195,21 @@ level the lines are suppressed with no cost. dbsql.SetLogLevel("debug") // or DATABRICKS_LOG_LEVEL=debug -The same level is mapped into the kernel's internal (Rust) log subscriber, with two -caveats specific to the Rust lines: they go to stderr directly (not affected by -logger.SetLogOutput), and their verbosity is fixed when the first kernel session in -the process is opened — set the level before that first connect to control them. +The same level is mapped into the kernel's internal (Rust) log subscriber. Its +records are forwarded into the driver's logger, so they use the same output as the +Go and Thrift paths, including a local file configured with logger.SetLogOutput. +Changing SetLogOutput later safely retargets all three paths. The Rust verbosity is +fixed when the first kernel session in the process is opened, so set the level +before that first connect. For finer control of the Rust verbosity independent of the driver level, set -DBSQL_KERNEL_DEBUG to any non-empty value: it forces the kernel subscriber on and -defers to RUST_LOG. Filter on the target databricks::sql::kernel (note the colons): +DBSQL_KERNEL_DEBUG to any non-empty value: the callback then defers filtering to +RUST_LOG. Filter on the target databricks::sql::kernel (note the colons): # kernel logs only, at the kernel's own verbosity: - DBSQL_KERNEL_DEBUG=1 RUST_LOG=databricks::sql::kernel=debug ./your_app 2>&1 + DBSQL_KERNEL_DEBUG=1 RUST_LOG=databricks::sql::kernel=debug ./your_app # kernel logs plus its HTTP stack: - DBSQL_KERNEL_DEBUG=1 RUST_LOG=debug ./your_app 2>&1 + DBSQL_KERNEL_DEBUG=1 RUST_LOG=debug ./your_app Supported on the kernel backend: PAT and OAuth (M2M via WithClientCredentials, U2M via the authType=oauthU2M DSN param); reading scalar, nested, and complex-typed diff --git a/internal/backend/kernel/cgo.go b/internal/backend/kernel/cgo.go index 3fc26289..20d5b403 100644 --- a/internal/backend/kernel/cgo.go +++ b/internal/backend/kernel/cgo.go @@ -37,7 +37,6 @@ import ( "context" "fmt" "runtime" - "sync" "unsafe" "github.com/databricks/databricks-sql-go/driverctx" @@ -101,56 +100,25 @@ func klogCtx(ctx context.Context, format string, args ...any) { ).Debug().Msgf("[kernel] "+format, args...) } -// initLoggingOnce guards kernel_init_logging, which is process-wide and -// first-call-wins in the kernel. We install the kernel subscriber lazily on the -// first session open rather than in init(), so a process that never opens a -// kernel session installs nothing. -var initLoggingOnce sync.Once - -// initKernelLogging turns on the kernel's own Rust (tracing) logs and points their -// verbosity at the driver's log level, so DATABRICKS_LOG_LEVEL drives both the Go -// binding lines and the kernel's Rust lines from one knob. The mapped level is -// passed to kernel_init_logging (Go zerolog level → the kernel's OFF/ERROR/WARN/ -// INFO/DEBUG/TRACE string); DBSQL_KERNEL_DEBUG forces the subscriber on with a -// NULL level so the kernel honors RUST_LOG instead (the advanced override for -// tuning kernel-only verbosity). file_path=NULL sends kernel logs to stderr — the -// kernel ABI has no sink hook, so the Rust lines always go to stderr and are NOT -// routed through logger.SetLogOutput (unlike the Go binding lines). +// initKernelLogging routes the kernel's own Rust tracing records into the +// driver's shared logger and maps the driver level into the kernel subscriber. +// Consequently DATABRICKS_LOG_LEVEL controls both paths and SetLogOutput controls +// their common destination. DBSQL_KERNEL_DEBUG keeps its advanced behavior: the +// callback is installed with a NULL level so the kernel honors RUST_LOG instead. // -// Best-effort: an Internal return (e.g. the host already installed a global -// subscriber) is a documented, benign outcome — logged at Warn, never fatal to -// connect. The subscriber installs at whatever level is mapped in; a driver left at -// the default Warn level (benchmarks included, and never having set -// DBSQL_KERNEL_DEBUG) installs it at WARN, so the kernel emits nothing below Warn -// and there is no hot-path cost. +// Best-effort: failure to install the process-global callback is logged at Warn +// and never fails a connection. At the default Warn level the kernel filters out +// lower-level events before crossing cgo. // // Scope caveat: the kernel subscriber is PROCESS-WIDE, first-call-wins, and never // uninstalled — in a long-lived multi-tenant process the first kernel session's -// level/destination applies to ALL subsequent kernel sessions, with no way to -// re-scope or turn it off afterward. That is a kernel-ABI property, not a Go one. -// A direct consequence: the driver level is sampled HERE, once, at the first kernel -// session — a later dbsql.SetLogLevel re-levels the Go binding lines (klog/klogCtx -// re-read GetLevel per call) but NOT the already-installed Rust subscriber. Set the -// level before opening the first kernel connection to govern the Rust logs. +// level applies to all later sessions. The driver level is sampled here, once; +// set it before the first kernel connection. The output is different: the shared +// logger uses a stable writer proxy, so a later SetLogOutput safely retargets both +// Go and forwarded Rust records. func initKernelLogging() { - initLoggingOnce.Do(func() { - // resolveKernelLogArg decides the level (or NULL for the DBSQL_KERNEL_DEBUG - // override, which lets the kernel honor RUST_LOG). The pure decision lives in - // logging_level.go so it's unit-tested without cgo. - var level cStr - if lvl, useNULL := resolveKernelLogArg(); !useNULL { - level = newCStr(lvl) - defer level.free() - } // else level stays {c: nil} → NULL → kernel honors RUST_LOG - if err := call(func() C.KernelStatusCode { - return C.kernel_init_logging(level.c, nil) - }); err != nil { - // The kernel subscriber didn't install (commonly: the host already - // installed a global tracing subscriber). Non-fatal — surface it through - // the shared logger so it's visible without a separate stderr scrape. - logger.Logger.Warn().Msgf("databricks: kernel_init_logging: %v (kernel logs unavailable; proceeding)", err) - } - }) + level, useNULL := resolveKernelLogArg() + installKernelLogCallback(level, useNULL) } // call runs a fallible kernel entry point and, on a non-Success status, reads diff --git a/internal/backend/kernel/log_callback.go b/internal/backend/kernel/log_callback.go new file mode 100644 index 00000000..6d3c5e09 --- /dev/null +++ b/internal/backend/kernel/log_callback.go @@ -0,0 +1,85 @@ +//go:build cgo && databricks_kernel + +package kernel + +/* +#include +#include "databricks_kernel.h" + +// cgo generates this declaration with mutable char pointers; the kernel lends +// read-only strings, so cast once in the C adapter to match KernelLogCallback. +void kernelLogTrampoline(char* level, char* target, char* message, void* user_data); +static KernelLogCallback kernel_log_cb(void) { + return (KernelLogCallback)kernelLogTrampoline; +} +static void* kernel_handle_to_ctx(uintptr_t h) { return (void*)h; } +static void kernel_invoke_log_trampoline_for_test( + char* level, char* target, char* message, void* user_data) { + kernelLogTrampoline(level, target, message, user_data); +} +*/ +import "C" + +import ( + "runtime/cgo" + "sync" + "unsafe" + + "github.com/databricks/databricks-sql-go/logger" +) + +var ( + logCallbackOnce sync.Once + logCallbackHandle cgo.Handle +) + +//export kernelLogTrampoline +func kernelLogTrampoline(level, target, message *C.char, userData unsafe.Pointer) { + // A panic must not cross the C ABI. Treat a malformed handle or sink failure + // as a dropped diagnostic; logging must never fail a query. + defer func() { _ = recover() }() + if userData == nil { + return + } + sink, ok := cgo.Handle(uintptr(userData)).Value().(*logSink) + if !ok || sink == nil { + return + } + sink.forward(C.GoString(level), C.GoString(target), C.GoString(message)) +} + +func invokeLogTrampolineForTest(h cgo.Handle, level, target, message string) { + clevel := C.CString(level) + defer C.free(unsafe.Pointer(clevel)) + ctarget := C.CString(target) + defer C.free(unsafe.Pointer(ctarget)) + cmessage := C.CString(message) + defer C.free(unsafe.Pointer(cmessage)) + ctx := C.kernel_handle_to_ctx(C.uintptr_t(h)) + C.kernel_invoke_log_trampoline_for_test(clevel, ctarget, cmessage, ctx) +} + +func installKernelLogCallback(level string, useNULL bool) { + logCallbackOnce.Do(func() { + // OFF intentionally installs no subscriber and retains no callback state. + if !useNULL && level == "OFF" { + return + } + + logCallbackHandle = cgo.NewHandle(newLogSink()) + ctx := C.kernel_handle_to_ctx(C.uintptr_t(logCallbackHandle)) + var clevel cStr + if !useNULL { + clevel = newCStr(level) + defer clevel.free() + } + if err := call(func() C.KernelStatusCode { + return C.kernel_init_logging_callback(clevel.c, C.kernel_log_cb(), ctx) + }); err != nil { + logCallbackHandle.Delete() + logCallbackHandle = 0 + logger.Logger.Warn().Msgf( + "databricks: kernel_init_logging_callback: %v (kernel logs not forwarded; proceeding)", err) + } + }) +} diff --git a/internal/backend/kernel/log_callback_test.go b/internal/backend/kernel/log_callback_test.go new file mode 100644 index 00000000..6e5249a3 --- /dev/null +++ b/internal/backend/kernel/log_callback_test.go @@ -0,0 +1,112 @@ +//go:build cgo && databricks_kernel + +package kernel + +import ( + "os" + "os/exec" + "path/filepath" + "runtime/cgo" + "strings" + "testing" + "time" + + "github.com/databricks/databricks-sql-go/logger" +) + +const ( + logFileHelperEnv = "DBSQL_KERNEL_LOG_FILE_HELPER" + logFilePathEnv = "DBSQL_KERNEL_LOG_FILE_PATH" + goLogFileProbe = "go local-file logging probe" + rustLogFileProbe = "retry max_wait_ms is below min_wait_ms" +) + +// TestKernelCallbackWritesConfiguredFileEndToEnd proves the user-visible parity +// contract in a fresh process: the same file passed to logger.SetLogOutput gets a +// native Go record and a real Rust tracing record delivered through the C ABI. +// A subprocess is required because the kernel tracing subscriber is process-wide +// and first-call-wins. +func TestKernelCallbackWritesConfiguredFileEndToEnd(t *testing.T) { + if os.Getenv(logFileHelperEnv) == "1" { + path := os.Getenv(logFilePathEnv) + file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600) //nolint:gosec // Parent supplies its temp path. + if err != nil { + t.Fatal(err) + } + if err := logger.SetLogLevel("warn"); err != nil { + t.Fatal(err) + } + logger.SetLogOutput(file) + + initKernelLogging() + logger.Logger.Warn().Msg(goLogFileProbe) + // The C ABI corrects this inverted range and emits a Rust klog::warn!, + // giving the test a deterministic kernel-owned record without a server. + err = trySetRetry(Config{Retry: &RetryConfig{ + MinWait: 5 * time.Second, + MaxWait: time.Second, + MaxRetries: 1, + }}) + if err != nil { + t.Fatal(err) + } + + logger.SetLogOutput(os.Stderr) + if err := file.Sync(); err != nil { + t.Fatal(err) + } + if err := file.Close(); err != nil { + t.Fatal(err) + } + return + } + + logPath := filepath.Join(t.TempDir(), "driver-and-kernel.log") + cmd := exec.Command(os.Args[0], "-test.run=^TestKernelCallbackWritesConfiguredFileEndToEnd$") //nolint:gosec // Re-executes this test binary only. + cmd.Env = append(os.Environ(), logFileHelperEnv+"=1", logFilePathEnv+"="+logPath) + if output, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("logging helper failed: %v\n%s", err, output) + } + contents, err := os.ReadFile(logPath) //nolint:gosec // Test-owned temporary path. + if err != nil { + t.Fatal(err) + } + got := string(contents) + if !strings.Contains(got, goLogFileProbe) { + t.Errorf("local log file is missing Go record: %q", got) + } + if !strings.Contains(got, rustLogFileProbe) { + t.Errorf("local log file is missing Rust kernel record: %q", got) + } +} + +func TestLogCallbackRoundTrip(t *testing.T) { + type record struct{ level, target, message string } + received := make(chan record, 1) + h := cgo.NewHandle(&logSink{observe: func(level, target, message string) { + received <- record{level, target, message} + }}) + defer h.Delete() + + invokeLogTrampolineForTest(h, "debug", "databricks::sql::kernel", "callback probe") + select { + case got := <-received: + want := record{"debug", "databricks::sql::kernel", "callback probe"} + if got != want { + t.Fatalf("callback record = %#v, want %#v", got, want) + } + default: + t.Fatal("callback did not reach the Go sink") + } +} + +func TestLogCallbackPanicDoesNotCrossABI(t *testing.T) { + h := cgo.NewHandle(&logSink{observe: func(string, string, string) { + panic("sink failure") + }}) + defer h.Delete() + + // The trampoline's recovery boundary converts a logger panic into a dropped + // diagnostic instead of allowing it to cross cgo and terminate the process. + invokeLogTrampolineForTest(h, "error", "databricks::sql::kernel", "boom") +} diff --git a/internal/backend/kernel/logforward.go b/internal/backend/kernel/logforward.go new file mode 100644 index 00000000..01547f89 --- /dev/null +++ b/internal/backend/kernel/logforward.go @@ -0,0 +1,44 @@ +package kernel + +import ( + "strings" + + "github.com/databricks/databricks-sql-go/logger" + "github.com/rs/zerolog" +) + +// logSink is the Go destination for kernel tracing records. The logger value is +// immutable and ungated because the kernel already applied its configured level. +// Its writer is the logger package's stable shared proxy, so SetLogOutput can +// safely retarget this snapshot after the first kernel connection. +type logSink struct { + log zerolog.Logger + observe func(level, target, message string) +} + +func newLogSink() *logSink { + return &logSink{log: logger.Logger.Level(zerolog.TraceLevel)} +} + +func (s *logSink) forward(level, target, message string) { + if s == nil { + return + } + if s.observe != nil { + s.observe(level, target, message) + } + switch strings.ToLower(level) { + case "error": + s.log.Error().Str("target", target).Msg(message) + case "warn": + s.log.Warn().Str("target", target).Msg(message) + case "info": + s.log.Info().Str("target", target).Msg(message) + case "debug": + s.log.Debug().Str("target", target).Msg(message) + case "trace": + s.log.Trace().Str("target", target).Msg(message) + default: + s.log.Debug().Str("target", target).Str("kernelLevel", level).Msg(message) + } +} diff --git a/internal/backend/kernel/logforward_test.go b/internal/backend/kernel/logforward_test.go new file mode 100644 index 00000000..56f72cc0 --- /dev/null +++ b/internal/backend/kernel/logforward_test.go @@ -0,0 +1,92 @@ +package kernel + +import ( + "bytes" + "encoding/json" + "os" + "strings" + "testing" + + "github.com/databricks/databricks-sql-go/logger" + "github.com/rs/zerolog" +) + +func TestLogSinkForwardMapsLevels(t *testing.T) { + cases := []struct { + level string + want string + }{ + {"error", "error"}, + {"warn", "warn"}, + {"info", "info"}, + {"debug", "debug"}, + {"trace", "trace"}, + {"future", "debug"}, + } + for _, tc := range cases { + var buf bytes.Buffer + sink := &logSink{log: logger.Logger.Output(&buf).Level(zerolog.TraceLevel)} + sink.forward(tc.level, "databricks::sql::kernel", "hello") + var record map[string]any + if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &record); err != nil { + t.Fatalf("level %q: %v", tc.level, err) + } + if record["level"] != tc.want || record["target"] != "databricks::sql::kernel" || record["message"] != "hello" { + t.Errorf("level %q: record = %#v", tc.level, record) + } + } +} + +func TestLogSinkFollowsLocalFileRetarget(t *testing.T) { + prevLevel := logger.Logger.GetLevel() + t.Cleanup(func() { + logger.SetLogOutput(os.Stderr) + logger.Logger.Logger = logger.Logger.Level(prevLevel) + }) + + first, err := os.CreateTemp(t.TempDir(), "kernel-first-*.log") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := first.Close(); err != nil { + t.Errorf("close first log: %v", err) + } + }) + second, err := os.CreateTemp(t.TempDir(), "kernel-second-*.log") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := second.Close(); err != nil { + t.Errorf("close second log: %v", err) + } + }) + + logger.SetLogOutput(first) + sink := newLogSink() + sink.forward("debug", "databricks::sql::kernel", "kernel first destination") + logger.SetLogOutput(second) + sink.forward("warn", "databricks::sql::kernel", "kernel second destination") + + if err := first.Sync(); err != nil { + t.Fatal(err) + } + if err := second.Sync(); err != nil { + t.Fatal(err) + } + firstBytes, err := os.ReadFile(first.Name()) + if err != nil { + t.Fatal(err) + } + secondBytes, err := os.ReadFile(second.Name()) + if err != nil { + t.Fatal(err) + } + if got := string(firstBytes); !strings.Contains(got, "kernel first destination") || strings.Contains(got, "kernel second destination") { + t.Fatalf("first log contents = %q", got) + } + if got := string(secondBytes); !strings.Contains(got, "kernel second destination") || strings.Contains(got, "kernel first destination") { + t.Fatalf("second log contents = %q", got) + } +} diff --git a/logger/logger.go b/logger/logger.go index 683501a1..ccbde806 100644 --- a/logger/logger.go +++ b/logger/logger.go @@ -4,6 +4,7 @@ import ( "io" "os" "runtime" + "sync" "time" "github.com/mattn/go-isatty" @@ -14,6 +15,27 @@ type DBSQLLogger struct { zerolog.Logger } +// sharedOutput is the process-wide destination behind every driver logger, +// including logger values derived before a later SetLogOutput call. Serializing +// writes keeps Go and kernel-callback records intact when they arrive from +// different goroutines or native kernel threads. +type sharedOutput struct { + mu sync.Mutex + w io.Writer +} + +func (o *sharedOutput) Write(p []byte) (int, error) { + o.mu.Lock() + defer o.mu.Unlock() + return o.w.Write(p) +} + +func (o *sharedOutput) set(w io.Writer) { + o.mu.Lock() + defer o.mu.Unlock() + o.w = w +} + // Track is a simple utility function to use with logger to log a message with a timestamp. // Recommended to use in conjunction with Duration. // @@ -36,15 +58,15 @@ func (l *DBSQLLogger) Duration(msg string, start time.Time) { l.Debug().Msgf("%v elapsed time: %v", msg, time.Since(start)) } -var Logger = &DBSQLLogger{ - zerolog.New(os.Stderr).With().Timestamp().Logger(), -} +var output = &sharedOutput{w: os.Stderr} + +var Logger = &DBSQLLogger{zerolog.New(output).With().Timestamp().Logger()} // Enable pretty printing for interactive terminals and json for production. func init() { // for tty terminal enable pretty logs if isatty.IsTerminal(os.Stdout.Fd()) && runtime.GOOS != "windows" { - Logger = &DBSQLLogger{Logger.Output(zerolog.ConsoleWriter{Out: os.Stderr})} + output.set(zerolog.ConsoleWriter{Out: os.Stderr}) } // by default only log warns or above loglvl := zerolog.WarnLevel @@ -72,7 +94,7 @@ func SetLogLevel(l string) error { // Sets logging output. Default is os.Stderr. If in terminal, pretty logs are enabled. func SetLogOutput(w io.Writer) { - Logger.Logger = Logger.Output(w) + output.set(w) } // Sets log to trace. -1 diff --git a/logger/logger_test.go b/logger/logger_test.go new file mode 100644 index 00000000..ea827b6c --- /dev/null +++ b/logger/logger_test.go @@ -0,0 +1,66 @@ +package logger + +import ( + "os" + "strings" + "testing" + + "github.com/rs/zerolog" +) + +// Existing logger values must follow SetLogOutput. The kernel callback keeps an +// immutable Trace-level logger value for thread safety, so retargeting only works +// when every value writes through the same stable output proxy. +func TestSetLogOutputRetargetsExistingLogger(t *testing.T) { + prevLevel := Logger.GetLevel() + t.Cleanup(func() { + SetLogOutput(os.Stderr) + Logger.Logger = Logger.Level(prevLevel) + }) + + first, err := os.CreateTemp(t.TempDir(), "driver-first-*.log") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := first.Close(); err != nil { + t.Errorf("close first log: %v", err) + } + }) + second, err := os.CreateTemp(t.TempDir(), "driver-second-*.log") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := second.Close(); err != nil { + t.Errorf("close second log: %v", err) + } + }) + + snapshot := Logger.Level(zerolog.TraceLevel) + SetLogOutput(first) + snapshot.Info().Msg("first destination") + SetLogOutput(second) + snapshot.Info().Msg("second destination") + + if err := first.Sync(); err != nil { + t.Fatal(err) + } + if err := second.Sync(); err != nil { + t.Fatal(err) + } + firstBytes, err := os.ReadFile(first.Name()) + if err != nil { + t.Fatal(err) + } + secondBytes, err := os.ReadFile(second.Name()) + if err != nil { + t.Fatal(err) + } + if got := string(firstBytes); !strings.Contains(got, "first destination") || strings.Contains(got, "second destination") { + t.Fatalf("first log contents = %q", got) + } + if got := string(secondBytes); !strings.Contains(got, "second destination") || strings.Contains(got, "first destination") { + t.Fatalf("second log contents = %q", got) + } +} From 83b26554bdd56cd22fece5b0228addfb4a362fe1 Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Fri, 21 Aug 2026 17:42:19 +0000 Subject: [PATCH 02/11] chore(kernel): update logging callback revision Signed-off-by: Vu Anh Phung --- KERNEL_REV | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/KERNEL_REV b/KERNEL_REV index 04e3f21c..c1ab2919 100644 --- a/KERNEL_REV +++ b/KERNEL_REV @@ -1 +1 @@ -f7aa79c00b35abb3f8236cf723bda289ea78a3b6 +e6e7333d6017266994ab30dcefde7bb2990f9d1e From 0b23fb1121a38e4bd892959554b73cb859d9ddce Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Fri, 21 Aug 2026 18:00:34 +0000 Subject: [PATCH 03/11] chore(kernel): update logging callback revision Signed-off-by: Vu Anh Phung --- KERNEL_REV | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/KERNEL_REV b/KERNEL_REV index c1ab2919..82b86cef 100644 --- a/KERNEL_REV +++ b/KERNEL_REV @@ -1 +1 @@ -e6e7333d6017266994ab30dcefde7bb2990f9d1e +ac34b110e92e21d7d4c09f06c6ca85d4dabb9140 From 1d9b9e5e9da192baf097d477e90df0c7d12d6a7c Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Fri, 21 Aug 2026 18:09:47 +0000 Subject: [PATCH 04/11] chore(kernel): update logging callback revision --- KERNEL_REV | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/KERNEL_REV b/KERNEL_REV index 82b86cef..bd83cc8d 100644 --- a/KERNEL_REV +++ b/KERNEL_REV @@ -1 +1 @@ -ac34b110e92e21d7d4c09f06c6ca85d4dabb9140 +b821d53a5d7a893a2cb9f42ed11ee48a91e19a9f From e056e9329cad578a2998d4cfe418e685984205d2 Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Fri, 21 Aug 2026 18:19:22 +0000 Subject: [PATCH 05/11] chore(kernel): update logging callback revision --- KERNEL_REV | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/KERNEL_REV b/KERNEL_REV index bd83cc8d..87114f3d 100644 --- a/KERNEL_REV +++ b/KERNEL_REV @@ -1 +1 @@ -b821d53a5d7a893a2cb9f42ed11ee48a91e19a9f +2dd4739f1c20e3a560bcaf52abed8aab9bd957cf From 90c58579af231a45305442c9608432ea5e2b16a0 Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Fri, 21 Aug 2026 22:47:05 +0000 Subject: [PATCH 06/11] fix(kernel): harden kernel log forwarding and shared logger output Address PR review of the log-forwarding path: - Drop the cgo.Handle->void* coercion; register with NULL user_data and reach the sink via a package global, so no Go pointer is fabricated into a C pointer the GC can fault on. - Forward off the kernel thread: the C callback copies the strings and does a non-blocking enqueue onto a bounded channel drained by one goroutine, with a drop counter and per-record panic recovery. - Use an exact-signature C adapter instead of a function-pointer cast. - Rework logger.sharedOutput into an atomic, LevelWriter-aware proxy: no lock held across the user Write (no self-deadlock, a stuck writer stays replaceable), nil normalized to io.Discard, and WriteLevel preserved. Adds enqueue/drop, drain-panic, nil-output, and LevelWriter tests. Co-authored-by: Isaac Signed-off-by: Vu Anh Phung --- internal/backend/kernel/log_callback.go | 121 +++++++++++++------ internal/backend/kernel/log_callback_test.go | 91 ++++++++++---- logger/logger.go | 71 ++++++++--- logger/logger_test.go | 69 +++++++++++ 4 files changed, 282 insertions(+), 70 deletions(-) diff --git a/internal/backend/kernel/log_callback.go b/internal/backend/kernel/log_callback.go index 6d3c5e09..7523a519 100644 --- a/internal/backend/kernel/log_callback.go +++ b/internal/backend/kernel/log_callback.go @@ -6,78 +6,129 @@ package kernel #include #include "databricks_kernel.h" -// cgo generates this declaration with mutable char pointers; the kernel lends -// read-only strings, so cast once in the C adapter to match KernelLogCallback. +// The Go export below is generated with mutable char* parameters, but +// KernelLogCallback lends read-only const char*. kernelLogAdapter has the exact +// KernelLogCallback signature and forwards to the Go export, so the function +// pointer handed to the kernel needs no incompatible function-pointer cast. void kernelLogTrampoline(char* level, char* target, char* message, void* user_data); -static KernelLogCallback kernel_log_cb(void) { - return (KernelLogCallback)kernelLogTrampoline; -} -static void* kernel_handle_to_ctx(uintptr_t h) { return (void*)h; } -static void kernel_invoke_log_trampoline_for_test( - char* level, char* target, char* message, void* user_data) { - kernelLogTrampoline(level, target, message, user_data); +static void kernelLogAdapter(const char* level, const char* target, + const char* message, void* user_data) { + kernelLogTrampoline((char*)level, (char*)target, (char*)message, user_data); } +static KernelLogCallback kernel_log_cb(void) { return kernelLogAdapter; } */ import "C" import ( - "runtime/cgo" "sync" + "sync/atomic" "unsafe" "github.com/databricks/databricks-sql-go/logger" ) +// kernelLogRecord is an owned copy of one kernel tracing record. The C strings +// are valid only for the duration of the callback, so the trampoline copies them +// before the record leaves the kernel thread. +type kernelLogRecord struct { + level string + target string + message string +} + +// kernelLogChannelCapacity bounds the hand-off buffer between kernel threads and +// the drain goroutine. Bursts beyond this are dropped rather than blocking a +// kernel thread — logs are advisory and must never back-pressure a kernel path. +const kernelLogChannelCapacity = 4096 + var ( - logCallbackOnce sync.Once - logCallbackHandle cgo.Handle + logCallbackOnce sync.Once + // logQueue publishes the bounded hand-off channel to the trampoline. It is an + // atomic pointer so the read on a kernel thread synchronizes with the write on + // the installing goroutine; nil until (and unless) logging installs. + logQueue atomic.Pointer[chan kernelLogRecord] + // logDropped counts records discarded because the buffer was full — a growing + // value means the sink cannot keep up. Exposed via kernelLogDropped. + logDropped atomic.Uint64 ) +// kernelLogDropped reports how many forwarded kernel records were dropped because +// the bounded hand-off buffer was full. Safe to call at any time. +func kernelLogDropped() uint64 { return logDropped.Load() } + //export kernelLogTrampoline -func kernelLogTrampoline(level, target, message *C.char, userData unsafe.Pointer) { - // A panic must not cross the C ABI. Treat a malformed handle or sink failure - // as a dropped diagnostic; logging must never fail a query. +func kernelLogTrampoline(level, target, message *C.char, _ unsafe.Pointer) { + // A panic must never cross the C ABI. user_data is deliberately unused: the + // kernel is given NULL, and the destination is reached through logQueue, so no + // Go pointer is ever fabricated into a C void* (which the GC could fault on). defer func() { _ = recover() }() - if userData == nil { + // C.GoString copies each borrowed string into owned Go memory before the + // record can outlive the callback; the rest is pure Go (see enqueueKernelLog). + enqueueKernelLog(C.GoString(level), C.GoString(target), C.GoString(message)) +} + +// enqueueKernelLog hands one already-owned record to the drain goroutine without +// blocking. Split out of the cgo trampoline so the enqueue/drop policy is +// testable without cgo (import "C" is not allowed in _test.go files). +func enqueueKernelLog(level, target, message string) { + qp := logQueue.Load() + if qp == nil { return } - sink, ok := cgo.Handle(uintptr(userData)).Value().(*logSink) - if !ok || sink == nil { - return + rec := kernelLogRecord{level: level, target: target, message: message} + // Non-blocking hand-off: never stall a kernel thread on a slow, contended, or + // re-entrant user writer. A full buffer drops the record and counts it. + select { + case *qp <- rec: + default: + logDropped.Add(1) } - sink.forward(C.GoString(level), C.GoString(target), C.GoString(message)) } -func invokeLogTrampolineForTest(h cgo.Handle, level, target, message string) { - clevel := C.CString(level) - defer C.free(unsafe.Pointer(clevel)) - ctarget := C.CString(target) - defer C.free(unsafe.Pointer(ctarget)) - cmessage := C.CString(message) - defer C.free(unsafe.Pointer(cmessage)) - ctx := C.kernel_handle_to_ctx(C.uintptr_t(h)) - C.kernel_invoke_log_trampoline_for_test(clevel, ctarget, cmessage, ctx) +// drainKernelLogs is the single goroutine that moves records off kernel threads +// and into the shared logger. Running the arbitrary user writer here — not in the +// trampoline — keeps user I/O (and any driver re-entry it triggers) off the +// kernel thread, honoring the C ABI's "return promptly / no re-entry" contract. +func drainKernelLogs(ch <-chan kernelLogRecord, sink *logSink) { + for rec := range ch { + // A writer panic must not kill the drain goroutine — an unrecovered + // goroutine panic is fatal to the process. Contain it per record. + func() { + defer func() { _ = recover() }() + sink.forward(rec.level, rec.target, rec.message) + }() + } } func installKernelLogCallback(level string, useNULL bool) { logCallbackOnce.Do(func() { - // OFF intentionally installs no subscriber and retains no callback state. + // OFF intentionally installs no subscriber and starts no drain. if !useNULL && level == "OFF" { return } - logCallbackHandle = cgo.NewHandle(newLogSink()) - ctx := C.kernel_handle_to_ctx(C.uintptr_t(logCallbackHandle)) + ch := make(chan kernelLogRecord, kernelLogChannelCapacity) + go drainKernelLogs(ch, newLogSink()) + // Publish before installing the subscriber so a callback that fires during + // kernel_init_logging_callback already sees the channel. + logQueue.Store(&ch) + var clevel cStr if !useNULL { clevel = newCStr(level) defer clevel.free() } + // NULL user_data: the drain goroutine owns the sink, so nothing + // Go-managed crosses into C as a pointer. if err := call(func() C.KernelStatusCode { - return C.kernel_init_logging_callback(clevel.c, C.kernel_log_cb(), ctx) + return C.kernel_init_logging_callback(clevel.c, C.kernel_log_cb(), nil) }); err != nil { - logCallbackHandle.Delete() - logCallbackHandle = 0 + // The subscriber did not install (commonly: a global subscriber is + // already set), so no callback will ever fire. Retire the drain: + // unpublish the channel first, then close it so the goroutine exits. + // Safe because no producer exists on this path. + logQueue.Store(nil) + close(ch) logger.Logger.Warn().Msgf( "databricks: kernel_init_logging_callback: %v (kernel logs not forwarded; proceeding)", err) } diff --git a/internal/backend/kernel/log_callback_test.go b/internal/backend/kernel/log_callback_test.go index 6e5249a3..c07cb530 100644 --- a/internal/backend/kernel/log_callback_test.go +++ b/internal/backend/kernel/log_callback_test.go @@ -6,7 +6,6 @@ import ( "os" "os/exec" "path/filepath" - "runtime/cgo" "strings" "testing" "time" @@ -51,6 +50,18 @@ func TestKernelCallbackWritesConfiguredFileEndToEnd(t *testing.T) { t.Fatal(err) } + // The kernel record now crosses an async drain goroutine, so wait until it + // lands in the file before retargeting the output or closing it — otherwise + // the drain could write to stderr (post-retarget) or after Close. Bounded so + // a real failure surfaces as a missing probe in the parent, not a hang. + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if b, _ := os.ReadFile(path); strings.Contains(string(b), rustLogFileProbe) { //nolint:gosec // Parent supplies its temp path. + break + } + time.Sleep(10 * time.Millisecond) + } + logger.SetLogOutput(os.Stderr) if err := file.Sync(); err != nil { t.Fatal(err) @@ -80,33 +91,71 @@ func TestKernelCallbackWritesConfiguredFileEndToEnd(t *testing.T) { } } -func TestLogCallbackRoundTrip(t *testing.T) { - type record struct{ level, target, message string } - received := make(chan record, 1) - h := cgo.NewHandle(&logSink{observe: func(level, target, message string) { - received <- record{level, target, message} - }}) - defer h.Delete() +// The forward path enqueues an owned record onto the bounded channel. (The cgo +// trampoline copies the borrowed C strings via C.GoString, then calls this; that +// tiny boundary can't be driven from a _test.go file because import "C" is +// disallowed there, so the testable logic lives in enqueueKernelLog.) +func TestEnqueueKernelLog(t *testing.T) { + ch := make(chan kernelLogRecord, 1) + prev := logQueue.Swap(&ch) + t.Cleanup(func() { logQueue.Store(prev) }) - invokeLogTrampolineForTest(h, "debug", "databricks::sql::kernel", "callback probe") + enqueueKernelLog("debug", "databricks::sql::kernel", "callback probe") select { - case got := <-received: - want := record{"debug", "databricks::sql::kernel", "callback probe"} + case got := <-ch: + want := kernelLogRecord{"debug", "databricks::sql::kernel", "callback probe"} if got != want { - t.Fatalf("callback record = %#v, want %#v", got, want) + t.Fatalf("enqueued record = %#v, want %#v", got, want) } default: - t.Fatal("callback did not reach the Go sink") + t.Fatal("record was not enqueued") + } +} + +// A full buffer drops the record and counts it rather than blocking the kernel +// thread. A nil queue (logging not installed) is a safe no-op. +func TestEnqueueKernelLogDropsWhenFullAndNoopWhenUnset(t *testing.T) { + // Unset queue: must not block or panic. + prev := logQueue.Swap(nil) + t.Cleanup(func() { logQueue.Store(prev) }) + enqueueKernelLog("warn", "t", "before install") + + ch := make(chan kernelLogRecord, 1) + logQueue.Store(&ch) + before := kernelLogDropped() + enqueueKernelLog("warn", "t", "keeps the one slot") // fills the buffer + enqueueKernelLog("warn", "t", "must be dropped") // buffer full → dropped + if got := kernelLogDropped() - before; got != 1 { + t.Fatalf("dropped delta = %d, want 1", got) + } + if len(ch) != 1 { + t.Fatalf("channel len = %d, want 1 (non-blocking drop)", len(ch)) } } -func TestLogCallbackPanicDoesNotCrossABI(t *testing.T) { - h := cgo.NewHandle(&logSink{observe: func(string, string, string) { - panic("sink failure") - }}) - defer h.Delete() +// A panicking writer must not kill the drain goroutine (an unrecovered goroutine +// panic is fatal to the process); the drain contains it and keeps processing. +func TestDrainRecoversFromWriterPanic(t *testing.T) { + done := make(chan string, 1) + sink := &logSink{observe: func(_, _, message string) { + if message == "boom" { + panic("writer failure") + } + done <- message + }} + + ch := make(chan kernelLogRecord, 2) + go drainKernelLogs(ch, sink) + defer close(ch) - // The trampoline's recovery boundary converts a logger panic into a dropped - // diagnostic instead of allowing it to cross cgo and terminate the process. - invokeLogTrampolineForTest(h, "error", "databricks::sql::kernel", "boom") + ch <- kernelLogRecord{"error", "t", "boom"} // triggers the panic + ch <- kernelLogRecord{"info", "t", "after"} // must still be delivered + select { + case got := <-done: + if got != "after" { + t.Fatalf("delivered %q, want %q", got, "after") + } + case <-time.After(2 * time.Second): + t.Fatal("drain goroutine did not survive a writer panic") + } } diff --git a/logger/logger.go b/logger/logger.go index ccbde806..c623ccbc 100644 --- a/logger/logger.go +++ b/logger/logger.go @@ -4,7 +4,7 @@ import ( "io" "os" "runtime" - "sync" + "sync/atomic" "time" "github.com/mattn/go-isatty" @@ -16,24 +16,67 @@ type DBSQLLogger struct { } // sharedOutput is the process-wide destination behind every driver logger, -// including logger values derived before a later SetLogOutput call. Serializing -// writes keeps Go and kernel-callback records intact when they arrive from -// different goroutines or native kernel threads. +// including logger values derived before a later SetLogOutput call. The current +// destination is held behind an atomic pointer and wrapped in a per-destination +// synchronized, level-aware writer. This gives three properties the driver's +// logging (Go, Thrift, and forwarded kernel records) relies on: +// +// - Retargeting (set) is a single atomic store: it never waits on an in-flight +// user Write, is never blocked by a slow/stuck writer, and never holds a lock +// across arbitrary user code — so SetLogOutput can't self-deadlock and a stuck +// writer can always be replaced. +// - Concurrent records to one destination stay intact (zerolog.SyncWriter +// serializes writes per destination). +// - A zerolog.LevelWriter destination keeps severity-aware routing: the proxy +// itself implements LevelWriter, so zerolog calls WriteLevel on it and it +// forwards WriteLevel to the destination. type sharedOutput struct { - mu sync.Mutex - w io.Writer + dst atomic.Pointer[zerolog.LevelWriter] } -func (o *sharedOutput) Write(p []byte) (int, error) { - o.mu.Lock() - defer o.mu.Unlock() - return o.w.Write(p) +func newSharedOutput(w io.Writer) *sharedOutput { + o := &sharedOutput{} + o.set(w) + return o } +// set publishes w as the current destination. A nil writer is normalized to +// io.Discard, matching the historical zerolog.New(nil) behavior (before the proxy +// existed, SetLogOutput(nil) → Logger.Output(nil) → io.Discard); without this the +// next log would panic on a nil-interface Write. func (o *sharedOutput) set(w io.Writer) { - o.mu.Lock() - defer o.mu.Unlock() - o.w = w + if w == nil { + w = io.Discard + } + // SyncWriter serializes concurrent writes to this destination and preserves a + // LevelWriter's WriteLevel (a plain writer is adapted). The result always + // implements LevelWriter; keep a fallback adapter in case that ever changes. + sw := zerolog.SyncWriter(w) + lw, ok := sw.(zerolog.LevelWriter) + if !ok { + lw = plainLevelWriter{sw} + } + o.dst.Store(&lw) +} + +func (o *sharedOutput) current() zerolog.LevelWriter { + return *o.dst.Load() +} + +func (o *sharedOutput) Write(p []byte) (int, error) { + return o.current().Write(p) +} + +func (o *sharedOutput) WriteLevel(l zerolog.Level, p []byte) (int, error) { + return o.current().WriteLevel(l, p) +} + +// plainLevelWriter adapts an io.Writer that is not a zerolog.LevelWriter, routing +// WriteLevel to Write (dropping the level, as zerolog's own adapter does). +type plainLevelWriter struct{ io.Writer } + +func (p plainLevelWriter) WriteLevel(_ zerolog.Level, b []byte) (int, error) { + return p.Write(b) } // Track is a simple utility function to use with logger to log a message with a timestamp. @@ -58,7 +101,7 @@ func (l *DBSQLLogger) Duration(msg string, start time.Time) { l.Debug().Msgf("%v elapsed time: %v", msg, time.Since(start)) } -var output = &sharedOutput{w: os.Stderr} +var output = newSharedOutput(os.Stderr) var Logger = &DBSQLLogger{zerolog.New(output).With().Timestamp().Logger()} diff --git a/logger/logger_test.go b/logger/logger_test.go index ea827b6c..517b4ea0 100644 --- a/logger/logger_test.go +++ b/logger/logger_test.go @@ -3,6 +3,7 @@ package logger import ( "os" "strings" + "sync" "testing" "github.com/rs/zerolog" @@ -64,3 +65,71 @@ func TestSetLogOutputRetargetsExistingLogger(t *testing.T) { t.Fatalf("second log contents = %q", got) } } + +// SetLogOutput(nil) must normalize to io.Discard, not store a nil writer that +// panics on the next enabled log. (Before the shared proxy, zerolog.New(nil) +// handled this; the proxy must preserve it.) +func TestSetLogOutputNilDiscards(t *testing.T) { + prevLevel := Logger.GetLevel() + t.Cleanup(func() { + SetLogOutput(os.Stderr) + Logger.Logger = Logger.Level(prevLevel) + }) + + SetLogOutput(nil) + Logger.Logger = Logger.Level(zerolog.TraceLevel) + // An enabled log actually reaches the writer; a nil writer would panic here. + Logger.Info().Msg("after nil output is discarded, not panicked") +} + +// recordingLevelWriter records whether zerolog reaches it via WriteLevel (level +// preserved) or Write (level lost). +type recordingLevelWriter struct { + mu sync.Mutex + levels []zerolog.Level + plain int +} + +func (w *recordingLevelWriter) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + w.plain++ + return len(p), nil +} + +func (w *recordingLevelWriter) WriteLevel(l zerolog.Level, p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + w.levels = append(w.levels, l) + return len(p), nil +} + +// A LevelWriter destination supplied via SetLogOutput must keep receiving +// WriteLevel (severity-aware routing), not be flattened to Write by the proxy. +func TestSetLogOutputPreservesLevelWriter(t *testing.T) { + prevLevel := Logger.GetLevel() + t.Cleanup(func() { + SetLogOutput(os.Stderr) + Logger.Logger = Logger.Level(prevLevel) + }) + + lw := &recordingLevelWriter{} + SetLogOutput(lw) + Logger.Logger = Logger.Level(zerolog.TraceLevel) + Logger.Warn().Msg("severity aware") + + lw.mu.Lock() + defer lw.mu.Unlock() + if lw.plain != 0 { + t.Fatalf("destination received %d plain Write calls; expected WriteLevel only", lw.plain) + } + foundWarn := false + for _, l := range lw.levels { + if l == zerolog.WarnLevel { + foundWarn = true + } + } + if !foundWarn { + t.Fatalf("WriteLevel not called with WarnLevel; severity routing lost (levels=%v)", lw.levels) + } +} From e1a40eea3720861301c1f3f6c40900ce708759f5 Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Fri, 21 Aug 2026 23:20:49 +0000 Subject: [PATCH 07/11] fix(kernel): add flush, emission timestamps, and drop visibility to log bridge Follow-up to review feedback on the async log-forwarding path: - Stamp each kernel record with the emission time captured in the cgo callback instead of the drain time. The sink is now a hook-free zerolog logger built over the shared output proxy (new logger.Output()), so a backed-up drain no longer skews kernel log timestamps. - Add flushKernelLogs(timeout): a FIFO barrier that drains the async hand-off before the writer is closed, retargeted, or the process exits. The end-to-end test flushes instead of polling. - installKernelLogCallback now starts the drain only after a successful install and drops the channel (no close) on failure, so there is no send-on-closed race to reason about on the error path. - Surface dropped kernel records with a one-shot warning from the drain goroutine (never the kernel thread); the running total stays available via kernelLogDropped(). - Document the SetLogOutput concurrent-retarget constraint and add concurrent-retarget, nil-discard, LevelWriter, and flush regression tests. Co-authored-by: Isaac Signed-off-by: Vu Anh Phung --- internal/backend/kernel/log_callback.go | 93 ++++++++++++++++---- internal/backend/kernel/log_callback_test.go | 67 ++++++++++---- internal/backend/kernel/logforward.go | 37 +++++--- internal/backend/kernel/logforward_test.go | 15 +++- logger/logger.go | 17 ++++ logger/logger_test.go | 79 +++++++++++++++++ 6 files changed, 255 insertions(+), 53 deletions(-) diff --git a/internal/backend/kernel/log_callback.go b/internal/backend/kernel/log_callback.go index 7523a519..b04ea479 100644 --- a/internal/backend/kernel/log_callback.go +++ b/internal/backend/kernel/log_callback.go @@ -22,6 +22,7 @@ import "C" import ( "sync" "sync/atomic" + "time" "unsafe" "github.com/databricks/databricks-sql-go/logger" @@ -29,11 +30,15 @@ import ( // kernelLogRecord is an owned copy of one kernel tracing record. The C strings // are valid only for the duration of the callback, so the trampoline copies them -// before the record leaves the kernel thread. +// before the record leaves the kernel thread. A record with a non-nil done +// channel is a flush barrier: the drain closes done (in FIFO order, after every +// earlier record is written) and forwards nothing — see flushKernelLogs. type kernelLogRecord struct { - level string - target string - message string + emittedAt time.Time + level string + target string + message string + done chan struct{} } // kernelLogChannelCapacity bounds the hand-off buffer between kernel threads and @@ -62,20 +67,22 @@ func kernelLogTrampoline(level, target, message *C.char, _ unsafe.Pointer) { // kernel is given NULL, and the destination is reached through logQueue, so no // Go pointer is ever fabricated into a C void* (which the GC could fault on). defer func() { _ = recover() }() - // C.GoString copies each borrowed string into owned Go memory before the - // record can outlive the callback; the rest is pure Go (see enqueueKernelLog). - enqueueKernelLog(C.GoString(level), C.GoString(target), C.GoString(message)) + // time.Now() here is the emission time — the callback fires synchronously on + // the kernel thread as the event is logged. C.GoString copies each borrowed + // string into owned Go memory before the record can outlive the callback; the + // rest is pure Go (see enqueueKernelLog). + enqueueKernelLog(time.Now(), C.GoString(level), C.GoString(target), C.GoString(message)) } // enqueueKernelLog hands one already-owned record to the drain goroutine without // blocking. Split out of the cgo trampoline so the enqueue/drop policy is // testable without cgo (import "C" is not allowed in _test.go files). -func enqueueKernelLog(level, target, message string) { +func enqueueKernelLog(emittedAt time.Time, level, target, message string) { qp := logQueue.Load() if qp == nil { return } - rec := kernelLogRecord{level: level, target: target, message: message} + rec := kernelLogRecord{emittedAt: emittedAt, level: level, target: target, message: message} // Non-blocking hand-off: never stall a kernel thread on a slow, contended, or // re-entrant user writer. A full buffer drops the record and counts it. select { @@ -90,13 +97,61 @@ func enqueueKernelLog(level, target, message string) { // trampoline — keeps user I/O (and any driver re-entry it triggers) off the // kernel thread, honoring the C ABI's "return promptly / no re-entry" contract. func drainKernelLogs(ch <-chan kernelLogRecord, sink *logSink) { + baselineDrops := logDropped.Load() + warnedDrop := false for rec := range ch { + // A flush barrier carries no record: closing done signals that every + // earlier record has been written (channel + drain are FIFO). + if rec.done != nil { + close(rec.done) + continue + } // A writer panic must not kill the drain goroutine — an unrecovered // goroutine panic is fatal to the process. Contain it per record. func() { defer func() { _ = recover() }() - sink.forward(rec.level, rec.target, rec.message) + sink.forward(rec.emittedAt, rec.level, rec.target, rec.message) }() + // Surface log loss the first time the sink falls behind — from this + // goroutine, never the kernel thread. One-shot so a burst can't turn into + // log spam; the running total stays available via kernelLogDropped(). + if !warnedDrop && logDropped.Load() > baselineDrops { + warnedDrop = true + logger.Logger.Warn().Uint64("dropped", logDropped.Load()-baselineDrops).Msg( + "[kernel] kernel log records dropped; the log sink is not keeping up " + + "(raise capacity or lower kernel verbosity)") + } + } +} + +// flushKernelLogs blocks until every kernel record already queued has been +// written, or until timeout elapses; it returns whether the flush completed. It +// is a no-op returning true when kernel logging was never installed. Callers use +// it to drain the asynchronous hand-off before closing the log writer, retargeting +// output, or exiting — records still buffered at process exit are otherwise lost. +// Best-effort: records dropped for a full buffer are gone, and records enqueued +// after this call are not waited on. +func flushKernelLogs(timeout time.Duration) bool { + qp := logQueue.Load() + if qp == nil { + return true + } + timer := time.NewTimer(timeout) + defer timer.Stop() + + done := make(chan struct{}) + // Enqueue the barrier behind everything already queued. A full buffer means + // the drain is behind; wait (bounded) for room rather than dropping the barrier. + select { + case *qp <- kernelLogRecord{done: done}: + case <-timer.C: + return false + } + select { + case <-done: + return true + case <-timer.C: + return false } } @@ -108,9 +163,9 @@ func installKernelLogCallback(level string, useNULL bool) { } ch := make(chan kernelLogRecord, kernelLogChannelCapacity) - go drainKernelLogs(ch, newLogSink()) - // Publish before installing the subscriber so a callback that fires during - // kernel_init_logging_callback already sees the channel. + // Publish before installing so a callback that fires during + // kernel_init_logging_callback already has a channel to enqueue onto; + // records buffer until the drain starts just below. logQueue.Store(&ch) var clevel cStr @@ -123,14 +178,16 @@ func installKernelLogCallback(level string, useNULL bool) { if err := call(func() C.KernelStatusCode { return C.kernel_init_logging_callback(clevel.c, C.kernel_log_cb(), nil) }); err != nil { - // The subscriber did not install (commonly: a global subscriber is - // already set), so no callback will ever fire. Retire the drain: - // unpublish the channel first, then close it so the goroutine exits. - // Safe because no producer exists on this path. + // Install failed, so the callback layer was not installed. Unpublish the + // channel; no drain was started and nothing references it, so it is simply + // collected — no close (and thus no send-on-closed race to reason about). logQueue.Store(nil) - close(ch) logger.Logger.Warn().Msgf( "databricks: kernel_init_logging_callback: %v (kernel logs not forwarded; proceeding)", err) + return } + // Installed: start the single drain goroutine. Any records enqueued during + // the call above are buffered and delivered once it runs. + go drainKernelLogs(ch, newLogSink()) }) } diff --git a/internal/backend/kernel/log_callback_test.go b/internal/backend/kernel/log_callback_test.go index c07cb530..a6801bef 100644 --- a/internal/backend/kernel/log_callback_test.go +++ b/internal/backend/kernel/log_callback_test.go @@ -50,16 +50,11 @@ func TestKernelCallbackWritesConfiguredFileEndToEnd(t *testing.T) { t.Fatal(err) } - // The kernel record now crosses an async drain goroutine, so wait until it - // lands in the file before retargeting the output or closing it — otherwise - // the drain could write to stderr (post-retarget) or after Close. Bounded so - // a real failure surfaces as a missing probe in the parent, not a hang. - deadline := time.Now().Add(5 * time.Second) - for time.Now().Before(deadline) { - if b, _ := os.ReadFile(path); strings.Contains(string(b), rustLogFileProbe) { //nolint:gosec // Parent supplies its temp path. - break - } - time.Sleep(10 * time.Millisecond) + // The kernel record crosses an async drain goroutine, so flush it into the + // file before retargeting the output or closing it — otherwise the drain + // could write to stderr (post-retarget) or after Close. + if !flushKernelLogs(5 * time.Second) { + t.Fatal("kernel log flush timed out") } logger.SetLogOutput(os.Stderr) @@ -100,12 +95,15 @@ func TestEnqueueKernelLog(t *testing.T) { prev := logQueue.Swap(&ch) t.Cleanup(func() { logQueue.Store(prev) }) - enqueueKernelLog("debug", "databricks::sql::kernel", "callback probe") + emittedAt := time.Now() + enqueueKernelLog(emittedAt, "debug", "databricks::sql::kernel", "callback probe") select { case got := <-ch: - want := kernelLogRecord{"debug", "databricks::sql::kernel", "callback probe"} - if got != want { - t.Fatalf("enqueued record = %#v, want %#v", got, want) + if got.level != "debug" || got.target != "databricks::sql::kernel" || got.message != "callback probe" { + t.Fatalf("enqueued record = %#v", got) + } + if !got.emittedAt.Equal(emittedAt) { + t.Fatalf("emittedAt = %v, want %v", got.emittedAt, emittedAt) } default: t.Fatal("record was not enqueued") @@ -118,13 +116,13 @@ func TestEnqueueKernelLogDropsWhenFullAndNoopWhenUnset(t *testing.T) { // Unset queue: must not block or panic. prev := logQueue.Swap(nil) t.Cleanup(func() { logQueue.Store(prev) }) - enqueueKernelLog("warn", "t", "before install") + enqueueKernelLog(time.Now(), "warn", "t", "before install") ch := make(chan kernelLogRecord, 1) logQueue.Store(&ch) before := kernelLogDropped() - enqueueKernelLog("warn", "t", "keeps the one slot") // fills the buffer - enqueueKernelLog("warn", "t", "must be dropped") // buffer full → dropped + enqueueKernelLog(time.Now(), "warn", "t", "keeps the one slot") // fills the buffer + enqueueKernelLog(time.Now(), "warn", "t", "must be dropped") // buffer full → dropped if got := kernelLogDropped() - before; got != 1 { t.Fatalf("dropped delta = %d, want 1", got) } @@ -148,8 +146,8 @@ func TestDrainRecoversFromWriterPanic(t *testing.T) { go drainKernelLogs(ch, sink) defer close(ch) - ch <- kernelLogRecord{"error", "t", "boom"} // triggers the panic - ch <- kernelLogRecord{"info", "t", "after"} // must still be delivered + ch <- kernelLogRecord{level: "error", target: "t", message: "boom"} // triggers the panic + ch <- kernelLogRecord{level: "info", target: "t", message: "after"} // must still be delivered select { case got := <-done: if got != "after" { @@ -159,3 +157,34 @@ func TestDrainRecoversFromWriterPanic(t *testing.T) { t.Fatal("drain goroutine did not survive a writer panic") } } + +// flushKernelLogs returns only after every record queued before it is written. +func TestFlushKernelLogsWaitsForQueued(t *testing.T) { + seen := make(chan string, 8) + sink := &logSink{observe: func(_, _, message string) { seen <- message }} + + ch := make(chan kernelLogRecord, 8) + prev := logQueue.Swap(&ch) + t.Cleanup(func() { logQueue.Store(prev) }) + go drainKernelLogs(ch, sink) + defer close(ch) + + enqueueKernelLog(time.Now(), "info", "t", "one") + enqueueKernelLog(time.Now(), "info", "t", "two") + if !flushKernelLogs(2 * time.Second) { + t.Fatal("flush timed out") + } + // The barrier is FIFO-ordered behind both records, so both are delivered. + if got := len(seen); got != 2 { + t.Fatalf("after flush, delivered %d records, want 2", got) + } +} + +// flushKernelLogs is a no-op returning true when logging was never installed. +func TestFlushKernelLogsNoopWhenUnset(t *testing.T) { + prev := logQueue.Swap(nil) + t.Cleanup(func() { logQueue.Store(prev) }) + if !flushKernelLogs(time.Second) { + t.Fatal("flush with no queue should be a no-op returning true") + } +} diff --git a/internal/backend/kernel/logforward.go b/internal/backend/kernel/logforward.go index 01547f89..3f73d9f0 100644 --- a/internal/backend/kernel/logforward.go +++ b/internal/backend/kernel/logforward.go @@ -2,43 +2,56 @@ package kernel import ( "strings" + "time" "github.com/databricks/databricks-sql-go/logger" "github.com/rs/zerolog" ) -// logSink is the Go destination for kernel tracing records. The logger value is -// immutable and ungated because the kernel already applied its configured level. -// Its writer is the logger package's stable shared proxy, so SetLogOutput can -// safely retarget this snapshot after the first kernel connection. +// logSink is the Go destination for kernel tracing records. The logger is built +// directly over the driver's shared output proxy (logger.Output) rather than +// derived from logger.Logger, so it (a) still follows SetLogOutput, (b) is ungated +// at TraceLevel because the kernel already applied its configured level, and (c) +// carries no auto-timestamp hook — forward stamps each record with the emission +// time captured on the kernel thread, not the drain time. type logSink struct { log zerolog.Logger observe func(level, target, message string) } func newLogSink() *logSink { - return &logSink{log: logger.Logger.Level(zerolog.TraceLevel)} + return &logSink{log: zerolog.New(logger.Output()).Level(zerolog.TraceLevel)} } -func (s *logSink) forward(level, target, message string) { +// forward writes one kernel record. emittedAt is the time the kernel emitted the +// event (captured in the cgo callback), stamped as the record's timestamp so a +// backed-up drain does not skew kernel log times toward drain time. +func (s *logSink) forward(emittedAt time.Time, level, target, message string) { if s == nil { return } if s.observe != nil { s.observe(level, target, message) } + ev := s.event(level) + ev.Time(zerolog.TimestampFieldName, emittedAt).Str("target", target).Msg(message) +} + +// event picks the zerolog event for a kernel level string. An unknown level maps +// to Debug and preserves the raw kernel level as a field. +func (s *logSink) event(level string) *zerolog.Event { switch strings.ToLower(level) { case "error": - s.log.Error().Str("target", target).Msg(message) + return s.log.Error() case "warn": - s.log.Warn().Str("target", target).Msg(message) + return s.log.Warn() case "info": - s.log.Info().Str("target", target).Msg(message) + return s.log.Info() case "debug": - s.log.Debug().Str("target", target).Msg(message) + return s.log.Debug() case "trace": - s.log.Trace().Str("target", target).Msg(message) + return s.log.Trace() default: - s.log.Debug().Str("target", target).Str("kernelLevel", level).Msg(message) + return s.log.Debug().Str("kernelLevel", level) } } diff --git a/internal/backend/kernel/logforward_test.go b/internal/backend/kernel/logforward_test.go index 56f72cc0..5238991a 100644 --- a/internal/backend/kernel/logforward_test.go +++ b/internal/backend/kernel/logforward_test.go @@ -6,6 +6,7 @@ import ( "os" "strings" "testing" + "time" "github.com/databricks/databricks-sql-go/logger" "github.com/rs/zerolog" @@ -23,10 +24,13 @@ func TestLogSinkForwardMapsLevels(t *testing.T) { {"trace", "trace"}, {"future", "debug"}, } + emittedAt := time.Now() for _, tc := range cases { var buf bytes.Buffer - sink := &logSink{log: logger.Logger.Output(&buf).Level(zerolog.TraceLevel)} - sink.forward(tc.level, "databricks::sql::kernel", "hello") + // Build the sink the way production does: hook-free over an explicit writer, + // so the only "time" field is the emission time forward stamps. + sink := &logSink{log: zerolog.New(&buf).Level(zerolog.TraceLevel)} + sink.forward(emittedAt, tc.level, "databricks::sql::kernel", "hello") var record map[string]any if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &record); err != nil { t.Fatalf("level %q: %v", tc.level, err) @@ -34,6 +38,9 @@ func TestLogSinkForwardMapsLevels(t *testing.T) { if record["level"] != tc.want || record["target"] != "databricks::sql::kernel" || record["message"] != "hello" { t.Errorf("level %q: record = %#v", tc.level, record) } + if _, ok := record[zerolog.TimestampFieldName]; !ok { + t.Errorf("level %q: record missing %q field: %#v", tc.level, zerolog.TimestampFieldName, record) + } } } @@ -65,9 +72,9 @@ func TestLogSinkFollowsLocalFileRetarget(t *testing.T) { logger.SetLogOutput(first) sink := newLogSink() - sink.forward("debug", "databricks::sql::kernel", "kernel first destination") + sink.forward(time.Now(), "debug", "databricks::sql::kernel", "kernel first destination") logger.SetLogOutput(second) - sink.forward("warn", "databricks::sql::kernel", "kernel second destination") + sink.forward(time.Now(), "warn", "databricks::sql::kernel", "kernel second destination") if err := first.Sync(); err != nil { t.Fatal(err) diff --git a/logger/logger.go b/logger/logger.go index c623ccbc..65d28c88 100644 --- a/logger/logger.go +++ b/logger/logger.go @@ -136,10 +136,27 @@ func SetLogLevel(l string) error { } // Sets logging output. Default is os.Stderr. If in terminal, pretty logs are enabled. +// A nil writer is treated as io.Discard. Existing logger values (and the kernel +// log bridge) follow later calls to this function. +// +// Writes are serialized per destination. If you hot-swap the output under +// concurrent logging while reusing the same underlying writer across swaps, pass a +// writer that is safe for concurrent use (os.File is; a bare bytes.Buffer is not): +// records in flight across a swap are serialized by that writer, not by the driver. func SetLogOutput(w io.Writer) { output.set(w) } +// Output returns the process-wide destination proxy that Logger writes through. +// Writes to it follow SetLogOutput exactly as Logger does, but it carries no +// level/timestamp context of its own. Callers that forward already-rendered +// records — such as the kernel log bridge — build their own zerolog.Logger over it +// so they can supply their own fields (including an accurate emission timestamp) +// instead of inheriting Logger's log-time one. +func Output() io.Writer { + return output +} + // Sets log to trace. -1 // You must call Msg on the returned event in order to send the event. func Trace() *zerolog.Event { diff --git a/logger/logger_test.go b/logger/logger_test.go index 517b4ea0..501ac38c 100644 --- a/logger/logger_test.go +++ b/logger/logger_test.go @@ -1,6 +1,8 @@ package logger import ( + "bytes" + "encoding/json" "os" "strings" "sync" @@ -133,3 +135,80 @@ func TestSetLogOutputPreservesLevelWriter(t *testing.T) { t.Fatalf("WriteLevel not called with WarnLevel; severity routing lost (levels=%v)", lw.levels) } } + +// countingWriter is a self-synchronized destination: it counts complete records +// and flags any that isn't a single well-formed JSON line (which is what byte +// interleaving from unsynchronized concurrent writes would produce). +type countingWriter struct { + mu sync.Mutex + lines int + bad int +} + +func (w *countingWriter) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + if json.Valid(bytes.TrimSpace(p)) { + w.lines++ + } else { + w.bad++ + } + return len(p), nil +} + +// Hot-swapping SetLogOutput under concurrent logging must not corrupt output when +// the destination is safe for concurrent use — even when the same writer is +// reapplied (each set builds a fresh SyncWriter, so the writer's own lock, not the +// driver's, is what serializes writes straddling a swap). Run under -race. +func TestConcurrentRetargetToSyncWriter(t *testing.T) { + prevLevel := Logger.GetLevel() + t.Cleanup(func() { + SetLogOutput(os.Stderr) + Logger.Logger = Logger.Level(prevLevel) + }) + + w := &countingWriter{} + SetLogOutput(w) + snapshot := Logger.Level(zerolog.TraceLevel) + + const writers, perWriter = 8, 200 + + // Retargeter: reapply the same writer until stopped, each call replacing the + // SyncWriter wrapper while writes are in flight. + stop := make(chan struct{}) + retargeterDone := make(chan struct{}) + go func() { + defer close(retargeterDone) + for { + select { + case <-stop: + return + default: + SetLogOutput(w) + } + } + }() + + var wg sync.WaitGroup + wg.Add(writers) + for i := 0; i < writers; i++ { + go func() { + defer wg.Done() + for j := 0; j < perWriter; j++ { + snapshot.Info().Int("j", j).Msg("concurrent") + } + }() + } + wg.Wait() + close(stop) + <-retargeterDone + + w.mu.Lock() + defer w.mu.Unlock() + if w.bad != 0 { + t.Fatalf("observed %d corrupted (non-JSON) writes", w.bad) + } + if w.lines != writers*perWriter { + t.Fatalf("wrote %d complete records, want %d", w.lines, writers*perWriter) + } +} From a9bf939518cb771f9fe11dfaf8b96aa71674c1f4 Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Fri, 21 Aug 2026 23:37:05 +0000 Subject: [PATCH 08/11] fix(kernel): contain drop-warning panics and remove SetLogOutput self-wrap footgun MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up review fixes: - HIGH: the one-shot drop warning wrote through the shared logger outside the drain's per-record recover, so a panicking writer plus a queue overflow could crash the process from the drain goroutine. Both the forwarded record and the drop warning now run inside a shared contain() recover. Adds a regression test that combines a panicking writer with dropped records. - MEDIUM: replace logger.Output() (which returned the live proxy, so SetLogOutput(Output()) — e.g. a save/restore of the current output — wrapped the proxy around itself and deadlocked the next write) with logger.NewForwardingLogger(), which returns a hook-free zerolog.Logger over the shared output and cannot be round-tripped back into SetLogOutput. - Clarify the flushKernelLogs doc: it drains the async hand-off for clean shutdown/retarget, but has no public entry point yet — only the end-to-end test calls it; a supported flush API is a separate change. Co-authored-by: Isaac Signed-off-by: Vu Anh Phung --- internal/backend/kernel/log_callback.go | 41 +++++++++++++------- internal/backend/kernel/log_callback_test.go | 40 +++++++++++++++++++ internal/backend/kernel/logforward.go | 14 +++---- logger/logger.go | 20 ++++++---- 4 files changed, 86 insertions(+), 29 deletions(-) diff --git a/internal/backend/kernel/log_callback.go b/internal/backend/kernel/log_callback.go index b04ea479..0a22377d 100644 --- a/internal/backend/kernel/log_callback.go +++ b/internal/backend/kernel/log_callback.go @@ -106,31 +106,44 @@ func drainKernelLogs(ch <-chan kernelLogRecord, sink *logSink) { close(rec.done) continue } - // A writer panic must not kill the drain goroutine — an unrecovered - // goroutine panic is fatal to the process. Contain it per record. - func() { - defer func() { _ = recover() }() + // Every write below goes to the user's writer, which may panic — an + // unrecovered goroutine panic is fatal to the process, so contain each one. + contain(func() { sink.forward(rec.emittedAt, rec.level, rec.target, rec.message) - }() + }) // Surface log loss the first time the sink falls behind — from this // goroutine, never the kernel thread. One-shot so a burst can't turn into - // log spam; the running total stays available via kernelLogDropped(). + // log spam; the running total stays available via kernelLogDropped(). This + // writes to the same destination as forward, so it is contained too. if !warnedDrop && logDropped.Load() > baselineDrops { warnedDrop = true - logger.Logger.Warn().Uint64("dropped", logDropped.Load()-baselineDrops).Msg( - "[kernel] kernel log records dropped; the log sink is not keeping up " + - "(raise capacity or lower kernel verbosity)") + dropped := logDropped.Load() - baselineDrops + contain(func() { + logger.Logger.Warn().Uint64("dropped", dropped).Msg( + "[kernel] kernel log records dropped; the log sink is not keeping up " + + "(raise capacity or lower kernel verbosity)") + }) } } } +// contain runs fn, swallowing any panic. A misbehaving user writer (reached via +// the shared logger) must never take down the drain goroutine. +func contain(fn func()) { + defer func() { _ = recover() }() + fn() +} + // flushKernelLogs blocks until every kernel record already queued has been // written, or until timeout elapses; it returns whether the flush completed. It -// is a no-op returning true when kernel logging was never installed. Callers use -// it to drain the asynchronous hand-off before closing the log writer, retargeting -// output, or exiting — records still buffered at process exit are otherwise lost. -// Best-effort: records dropped for a full buffer are gone, and records enqueued -// after this call are not waited on. +// is a no-op returning true when kernel logging was never installed. +// +// It drains the asynchronous hand-off so records are not lost or misrouted when +// the log writer is closed, output is retargeted, or the process exits. There is +// currently no public entry point that calls it — only the end-to-end test does; +// exposing a supported flush API (or wiring it into a shutdown path) is a separate +// change. Best-effort: records already dropped for a full buffer are gone, and +// records enqueued after this call are not waited on. func flushKernelLogs(timeout time.Duration) bool { qp := logQueue.Load() if qp == nil { diff --git a/internal/backend/kernel/log_callback_test.go b/internal/backend/kernel/log_callback_test.go index a6801bef..dbc5575c 100644 --- a/internal/backend/kernel/log_callback_test.go +++ b/internal/backend/kernel/log_callback_test.go @@ -188,3 +188,43 @@ func TestFlushKernelLogsNoopWhenUnset(t *testing.T) { t.Fatal("flush with no queue should be a no-op returning true") } } + +type panicWriter struct{} + +func (panicWriter) Write([]byte) (int, error) { panic("writer always panics") } + +// A panicking writer combined with dropped records must not crash the drain: the +// forwarded record AND the one-shot drop warning both write to that writer, and +// both must be contained. observe advances the drop counter past the drain's +// baseline (so the warning fires) and panics (so forward is exercised too); the +// shared output is the panicking writer (so the warning write panics). +func TestDrainContainsPanicFromForwardAndDropWarning(t *testing.T) { + prevLevel := logger.Logger.GetLevel() + t.Cleanup(func() { + logger.SetLogOutput(os.Stderr) + logger.Logger.Logger = logger.Logger.Level(prevLevel) + }) + if err := logger.SetLogLevel("warn"); err != nil { + t.Fatal(err) + } + logger.SetLogOutput(panicWriter{}) + + sink := newLogSink() + sink.observe = func(_, _, _ string) { + logDropped.Add(1) // advance past the drain's baseline, deterministically + panic("forward failure") + } + + ch := make(chan kernelLogRecord, 4) + prev := logQueue.Swap(&ch) + t.Cleanup(func() { logQueue.Store(prev) }) + go drainKernelLogs(ch, sink) + defer close(ch) + + enqueueKernelLog(time.Now(), "info", "databricks::sql::kernel", "boom") + + // The drain reaches the flush barrier only if it survived both panics. + if !flushKernelLogs(2 * time.Second) { + t.Fatal("drain did not survive a panicking writer combined with dropped records") + } +} diff --git a/internal/backend/kernel/logforward.go b/internal/backend/kernel/logforward.go index 3f73d9f0..51c8e7e6 100644 --- a/internal/backend/kernel/logforward.go +++ b/internal/backend/kernel/logforward.go @@ -8,19 +8,19 @@ import ( "github.com/rs/zerolog" ) -// logSink is the Go destination for kernel tracing records. The logger is built -// directly over the driver's shared output proxy (logger.Output) rather than -// derived from logger.Logger, so it (a) still follows SetLogOutput, (b) is ungated -// at TraceLevel because the kernel already applied its configured level, and (c) -// carries no auto-timestamp hook — forward stamps each record with the emission -// time captured on the kernel thread, not the drain time. +// logSink is the Go destination for kernel tracing records. Its logger comes from +// logger.NewForwardingLogger rather than being derived from logger.Logger, so it +// (a) still follows SetLogOutput, (b) is ungated at TraceLevel because the kernel +// already applied its configured level, and (c) carries no auto-timestamp hook — +// forward stamps each record with the emission time captured on the kernel thread, +// not the drain time. type logSink struct { log zerolog.Logger observe func(level, target, message string) } func newLogSink() *logSink { - return &logSink{log: zerolog.New(logger.Output()).Level(zerolog.TraceLevel)} + return &logSink{log: logger.NewForwardingLogger().Level(zerolog.TraceLevel)} } // forward writes one kernel record. emittedAt is the time the kernel emitted the diff --git a/logger/logger.go b/logger/logger.go index 65d28c88..d118f3e4 100644 --- a/logger/logger.go +++ b/logger/logger.go @@ -147,14 +147,18 @@ func SetLogOutput(w io.Writer) { output.set(w) } -// Output returns the process-wide destination proxy that Logger writes through. -// Writes to it follow SetLogOutput exactly as Logger does, but it carries no -// level/timestamp context of its own. Callers that forward already-rendered -// records — such as the kernel log bridge — build their own zerolog.Logger over it -// so they can supply their own fields (including an accurate emission timestamp) -// instead of inheriting Logger's log-time one. -func Output() io.Writer { - return output +// NewForwardingLogger returns a zerolog.Logger that writes through the same +// process-wide destination as Logger — so it follows SetLogOutput — but carries +// none of Logger's context: no level gate and no timestamp hook. It is for callers +// that forward already-rendered records and supply their own fields (including an +// accurate timestamp), such as the kernel log bridge. +// +// It deliberately returns a Logger, not the underlying writer: exposing the writer +// invites SetLogOutput(Output()) (e.g. a save/restore of the "current" output), +// which would wrap the shared proxy around itself and deadlock the next write on +// the re-entered SyncWriter mutex. +func NewForwardingLogger() zerolog.Logger { + return zerolog.New(output) } // Sets log to trace. -1 From 759f185db0cf9f191118732125fd487a172ef413 Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Sat, 22 Aug 2026 01:05:52 +0000 Subject: [PATCH 09/11] fix(logger): forward kernel logs through a non-io.Writer sink The previous change returned a zerolog.Logger from NewForwardingLogger, but zerolog.Logger implements io.Writer (value-receiver Write), so SetLogOutput(NewForwardingLogger()) still compiled and deadlocked: the forwarding logger's Write routes back through the shared output into the same SyncWriter mutex. Re-entrant custom writers hit the same trap. Replace it with logger.ForwardingSink, a method-only type (Event(level) *zerolog.Event) that is deliberately not an io.Writer, so it cannot be passed to SetLogOutput to form a self-referential, deadlocking sink. The kernel log sink now holds a *ForwardingSink and keeps its level mapping. Adds a regression test pinning that ForwardingSink is not an io.Writer. Co-authored-by: Isaac Signed-off-by: Vu Anh Phung --- internal/backend/kernel/log_callback_test.go | 13 +++++-- internal/backend/kernel/logforward.go | 32 ++++++++--------- internal/backend/kernel/logforward_test.go | 9 +++-- logger/logger.go | 36 ++++++++++++++------ logger/logger_test.go | 12 +++++++ 5 files changed, 69 insertions(+), 33 deletions(-) diff --git a/internal/backend/kernel/log_callback_test.go b/internal/backend/kernel/log_callback_test.go index dbc5575c..d98a07de 100644 --- a/internal/backend/kernel/log_callback_test.go +++ b/internal/backend/kernel/log_callback_test.go @@ -3,6 +3,7 @@ package kernel import ( + "io" "os" "os/exec" "path/filepath" @@ -134,13 +135,16 @@ func TestEnqueueKernelLogDropsWhenFullAndNoopWhenUnset(t *testing.T) { // A panicking writer must not kill the drain goroutine (an unrecovered goroutine // panic is fatal to the process); the drain contains it and keeps processing. func TestDrainRecoversFromWriterPanic(t *testing.T) { + t.Cleanup(func() { logger.SetLogOutput(os.Stderr) }) + logger.SetLogOutput(io.Discard) // discard the sink's forwarded records done := make(chan string, 1) - sink := &logSink{observe: func(_, _, message string) { + sink := newLogSink() + sink.observe = func(_, _, message string) { if message == "boom" { panic("writer failure") } done <- message - }} + } ch := make(chan kernelLogRecord, 2) go drainKernelLogs(ch, sink) @@ -160,8 +164,11 @@ func TestDrainRecoversFromWriterPanic(t *testing.T) { // flushKernelLogs returns only after every record queued before it is written. func TestFlushKernelLogsWaitsForQueued(t *testing.T) { + t.Cleanup(func() { logger.SetLogOutput(os.Stderr) }) + logger.SetLogOutput(io.Discard) // discard the sink's forwarded records seen := make(chan string, 8) - sink := &logSink{observe: func(_, _, message string) { seen <- message }} + sink := newLogSink() + sink.observe = func(_, _, message string) { seen <- message } ch := make(chan kernelLogRecord, 8) prev := logQueue.Swap(&ch) diff --git a/internal/backend/kernel/logforward.go b/internal/backend/kernel/logforward.go index 51c8e7e6..7ae810e2 100644 --- a/internal/backend/kernel/logforward.go +++ b/internal/backend/kernel/logforward.go @@ -8,19 +8,20 @@ import ( "github.com/rs/zerolog" ) -// logSink is the Go destination for kernel tracing records. Its logger comes from -// logger.NewForwardingLogger rather than being derived from logger.Logger, so it -// (a) still follows SetLogOutput, (b) is ungated at TraceLevel because the kernel -// already applied its configured level, and (c) carries no auto-timestamp hook — -// forward stamps each record with the emission time captured on the kernel thread, -// not the drain time. +// logSink is the Go destination for kernel tracing records. It forwards through +// logger.ForwardingSink (not a logger derived from logger.Logger), so it (a) still +// follows SetLogOutput, (b) is ungated at TraceLevel because the kernel already +// applied its configured level, (c) carries no auto-timestamp hook — forward stamps +// each record with the emission time captured on the kernel thread, not the drain +// time — and (d) cannot be round-tripped into SetLogOutput (ForwardingSink is not +// an io.Writer), which would deadlock. type logSink struct { - log zerolog.Logger + sink *logger.ForwardingSink observe func(level, target, message string) } func newLogSink() *logSink { - return &logSink{log: logger.NewForwardingLogger().Level(zerolog.TraceLevel)} + return &logSink{sink: logger.NewForwardingSink()} } // forward writes one kernel record. emittedAt is the time the kernel emitted the @@ -33,8 +34,7 @@ func (s *logSink) forward(emittedAt time.Time, level, target, message string) { if s.observe != nil { s.observe(level, target, message) } - ev := s.event(level) - ev.Time(zerolog.TimestampFieldName, emittedAt).Str("target", target).Msg(message) + s.event(level).Time(zerolog.TimestampFieldName, emittedAt).Str("target", target).Msg(message) } // event picks the zerolog event for a kernel level string. An unknown level maps @@ -42,16 +42,16 @@ func (s *logSink) forward(emittedAt time.Time, level, target, message string) { func (s *logSink) event(level string) *zerolog.Event { switch strings.ToLower(level) { case "error": - return s.log.Error() + return s.sink.Event(zerolog.ErrorLevel) case "warn": - return s.log.Warn() + return s.sink.Event(zerolog.WarnLevel) case "info": - return s.log.Info() + return s.sink.Event(zerolog.InfoLevel) case "debug": - return s.log.Debug() + return s.sink.Event(zerolog.DebugLevel) case "trace": - return s.log.Trace() + return s.sink.Event(zerolog.TraceLevel) default: - return s.log.Debug().Str("kernelLevel", level) + return s.sink.Event(zerolog.DebugLevel).Str("kernelLevel", level) } } diff --git a/internal/backend/kernel/logforward_test.go b/internal/backend/kernel/logforward_test.go index 5238991a..b0c80683 100644 --- a/internal/backend/kernel/logforward_test.go +++ b/internal/backend/kernel/logforward_test.go @@ -24,12 +24,15 @@ func TestLogSinkForwardMapsLevels(t *testing.T) { {"trace", "trace"}, {"future", "debug"}, } + t.Cleanup(func() { logger.SetLogOutput(os.Stderr) }) emittedAt := time.Now() for _, tc := range cases { var buf bytes.Buffer - // Build the sink the way production does: hook-free over an explicit writer, - // so the only "time" field is the emission time forward stamps. - sink := &logSink{log: zerolog.New(&buf).Level(zerolog.TraceLevel)} + // The sink forwards through the shared output; point it at a buffer to + // capture the single record and confirm it is one hook-free JSON line whose + // only timestamp is the emission time forward stamps. + logger.SetLogOutput(&buf) + sink := newLogSink() sink.forward(emittedAt, tc.level, "databricks::sql::kernel", "hello") var record map[string]any if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &record); err != nil { diff --git a/logger/logger.go b/logger/logger.go index d118f3e4..650c274b 100644 --- a/logger/logger.go +++ b/logger/logger.go @@ -147,18 +147,32 @@ func SetLogOutput(w io.Writer) { output.set(w) } -// NewForwardingLogger returns a zerolog.Logger that writes through the same -// process-wide destination as Logger — so it follows SetLogOutput — but carries -// none of Logger's context: no level gate and no timestamp hook. It is for callers -// that forward already-rendered records and supply their own fields (including an -// accurate timestamp), such as the kernel log bridge. +// ForwardingSink emits already-rendered log records from an external source (such +// as the Rust kernel) to the driver's shared destination. It follows SetLogOutput +// and bypasses Logger's level gate and timestamp hook, so the source can supply its +// own level and its own (emission-time) timestamp. // -// It deliberately returns a Logger, not the underlying writer: exposing the writer -// invites SetLogOutput(Output()) (e.g. a save/restore of the "current" output), -// which would wrap the shared proxy around itself and deadlock the next write on -// the re-entered SyncWriter mutex. -func NewForwardingLogger() zerolog.Logger { - return zerolog.New(output) +// It is deliberately NOT an io.Writer — and neither is the *zerolog.Event it hands +// out. A forwarding sink that is an io.Writer can be passed to SetLogOutput, which +// wraps it in a SyncWriter and stores it as the destination; the sink's own writes +// then route back through that same SyncWriter (output → SyncWriter → sink → +// output), and because SyncWriter holds its mutex across the write, the next record +// deadlocks. Handing out a Logger fails the same way (zerolog.Logger implements +// io.Writer). A method-only sink cannot form that cycle. +type ForwardingSink struct { + log zerolog.Logger +} + +// NewForwardingSink returns a sink over the shared output, ungated at TraceLevel +// because the external source has already applied its own level filter. +func NewForwardingSink() *ForwardingSink { + return &ForwardingSink{log: zerolog.New(output).Level(zerolog.TraceLevel)} +} + +// Event begins a record at level on the shared destination; the caller adds fields +// and terminates with Msg. Like Logger, it follows SetLogOutput. +func (s *ForwardingSink) Event(level zerolog.Level) *zerolog.Event { + return s.log.WithLevel(level) } // Sets log to trace. -1 diff --git a/logger/logger_test.go b/logger/logger_test.go index 501ac38c..b5448f03 100644 --- a/logger/logger_test.go +++ b/logger/logger_test.go @@ -3,6 +3,7 @@ package logger import ( "bytes" "encoding/json" + "io" "os" "strings" "sync" @@ -136,6 +137,17 @@ func TestSetLogOutputPreservesLevelWriter(t *testing.T) { } } +// A ForwardingSink must not be an io.Writer. If it were, SetLogOutput(sink) would +// store it as the destination and its own writes would route back through the +// SyncWriter that wraps it (output -> SyncWriter -> sink -> output), deadlocking +// the next record on the re-entered mutex — the same trap that made returning a +// zerolog.Logger (which implements io.Writer) unsafe. +func TestForwardingSinkIsNotAnIOWriter(t *testing.T) { + if _, ok := any(NewForwardingSink()).(io.Writer); ok { + t.Fatal("ForwardingSink must not implement io.Writer (would enable a self-referential SetLogOutput deadlock)") + } +} + // countingWriter is a self-synchronized destination: it counts complete records // and flags any that isn't a single well-formed JSON line (which is what byte // interleaving from unsynchronized concurrent writes would produce). From e09d2cd7977fe9fa080545b074bf7258e6b9dc3c Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Sat, 22 Aug 2026 03:39:34 +0000 Subject: [PATCH 10/11] docs(logger): warn that SetLogOutput writers must not be re-entrant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-destination serialization holds a non-reentrant lock across the user Write/WriteLevel, so a destination that logs through the driver from within its own write (a self-referential sink) deadlocks — and would recurse without bound even if it did not. This is inherent to any logging library; document the constraint on SetLogOutput rather than dropping serialization (which would only re-open the concurrent-corruption case and turn the deadlock into a stack overflow). Co-authored-by: Isaac Signed-off-by: Vu Anh Phung --- logger/logger.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/logger/logger.go b/logger/logger.go index 650c274b..e858dcf3 100644 --- a/logger/logger.go +++ b/logger/logger.go @@ -143,6 +143,12 @@ func SetLogLevel(l string) error { // concurrent logging while reusing the same underlying writer across swaps, pass a // writer that is safe for concurrent use (os.File is; a bare bytes.Buffer is not): // records in flight across a swap are serialized by that writer, not by the driver. +// +// The writer must not log through this driver from within its own Write/WriteLevel, +// directly or indirectly: per-destination serialization holds a non-reentrant lock +// across the write, so a re-entrant writer deadlocks (and a truly self-referential +// one would recurse without bound regardless). As with any logging library, do not +// log from your log destination. func SetLogOutput(w io.Writer) { output.set(w) } From 29bfd3381c95d519d4197530141f14a0e94a9678 Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Sat, 22 Aug 2026 04:35:30 +0000 Subject: [PATCH 11/11] refactor(kernel): untag pure-Go log pipeline and route drop warning off logger.Logger Two review fixes: - Move the pure-Go async forwarding pipeline (record type, bounded queue, drain, flush barrier, drop accounting, panic containment) out of the cgo-tagged log_callback.go into the untagged logforward_async.go, and its tests into logforward_async_test.go. This matches the repo convention documented in logging_level.go: the FIFO-flush, drop-policy, and panic-containment tests now run in the default CGO_ENABLED=0 lane instead of only the kernel-linked lane. log_callback.go keeps only the cgo trampoline and the one-time install. - Route the drain's one-shot drop warning through the sink's own immutable logger (new logSink.warnDropped) instead of logger.Logger, whose embedded value SetLogLevel reassigns. The long-lived drain goroutine no longer reads a field that races SetLogLevel; drainKernelLogs is now pure Go. Co-authored-by: Isaac Signed-off-by: Vu Anh Phung --- internal/backend/kernel/log_callback.go | 144 ++------------- internal/backend/kernel/log_callback_test.go | 150 ---------------- internal/backend/kernel/logforward.go | 13 ++ internal/backend/kernel/logforward_async.go | 136 +++++++++++++++ .../backend/kernel/logforward_async_test.go | 165 ++++++++++++++++++ 5 files changed, 326 insertions(+), 282 deletions(-) create mode 100644 internal/backend/kernel/logforward_async.go create mode 100644 internal/backend/kernel/logforward_async_test.go diff --git a/internal/backend/kernel/log_callback.go b/internal/backend/kernel/log_callback.go index 0a22377d..bdb7279f 100644 --- a/internal/backend/kernel/log_callback.go +++ b/internal/backend/kernel/log_callback.go @@ -21,45 +21,19 @@ import "C" import ( "sync" - "sync/atomic" "time" "unsafe" "github.com/databricks/databricks-sql-go/logger" ) -// kernelLogRecord is an owned copy of one kernel tracing record. The C strings -// are valid only for the duration of the callback, so the trampoline copies them -// before the record leaves the kernel thread. A record with a non-nil done -// channel is a flush barrier: the drain closes done (in FIFO order, after every -// earlier record is written) and forwards nothing — see flushKernelLogs. -type kernelLogRecord struct { - emittedAt time.Time - level string - target string - message string - done chan struct{} -} - -// kernelLogChannelCapacity bounds the hand-off buffer between kernel threads and -// the drain goroutine. Bursts beyond this are dropped rather than blocking a -// kernel thread — logs are advisory and must never back-pressure a kernel path. -const kernelLogChannelCapacity = 4096 - -var ( - logCallbackOnce sync.Once - // logQueue publishes the bounded hand-off channel to the trampoline. It is an - // atomic pointer so the read on a kernel thread synchronizes with the write on - // the installing goroutine; nil until (and unless) logging installs. - logQueue atomic.Pointer[chan kernelLogRecord] - // logDropped counts records discarded because the buffer was full — a growing - // value means the sink cannot keep up. Exposed via kernelLogDropped. - logDropped atomic.Uint64 -) +// This file is the thin cgo layer of the kernel log bridge: the exported callback +// trampoline and the one-time kernel_init_logging_callback install. The pure-Go +// pipeline it drives (queue, drain, flush, drop accounting) lives untagged in +// logforward_async.go so its tests run in the default CGO_ENABLED=0 build. -// kernelLogDropped reports how many forwarded kernel records were dropped because -// the bounded hand-off buffer was full. Safe to call at any time. -func kernelLogDropped() uint64 { return logDropped.Load() } +// logCallbackOnce guards the process-wide, first-call-wins install. +var logCallbackOnce sync.Once //export kernelLogTrampoline func kernelLogTrampoline(level, target, message *C.char, _ unsafe.Pointer) { @@ -67,107 +41,13 @@ func kernelLogTrampoline(level, target, message *C.char, _ unsafe.Pointer) { // kernel is given NULL, and the destination is reached through logQueue, so no // Go pointer is ever fabricated into a C void* (which the GC could fault on). defer func() { _ = recover() }() - // time.Now() here is the emission time — the callback fires synchronously on - // the kernel thread as the event is logged. C.GoString copies each borrowed - // string into owned Go memory before the record can outlive the callback; the - // rest is pure Go (see enqueueKernelLog). + // time.Now() here is the emission time — the callback fires synchronously on the + // kernel thread as the event is logged. C.GoString copies each borrowed string + // into owned Go memory before the record can outlive the callback; the rest is + // pure Go (see enqueueKernelLog). enqueueKernelLog(time.Now(), C.GoString(level), C.GoString(target), C.GoString(message)) } -// enqueueKernelLog hands one already-owned record to the drain goroutine without -// blocking. Split out of the cgo trampoline so the enqueue/drop policy is -// testable without cgo (import "C" is not allowed in _test.go files). -func enqueueKernelLog(emittedAt time.Time, level, target, message string) { - qp := logQueue.Load() - if qp == nil { - return - } - rec := kernelLogRecord{emittedAt: emittedAt, level: level, target: target, message: message} - // Non-blocking hand-off: never stall a kernel thread on a slow, contended, or - // re-entrant user writer. A full buffer drops the record and counts it. - select { - case *qp <- rec: - default: - logDropped.Add(1) - } -} - -// drainKernelLogs is the single goroutine that moves records off kernel threads -// and into the shared logger. Running the arbitrary user writer here — not in the -// trampoline — keeps user I/O (and any driver re-entry it triggers) off the -// kernel thread, honoring the C ABI's "return promptly / no re-entry" contract. -func drainKernelLogs(ch <-chan kernelLogRecord, sink *logSink) { - baselineDrops := logDropped.Load() - warnedDrop := false - for rec := range ch { - // A flush barrier carries no record: closing done signals that every - // earlier record has been written (channel + drain are FIFO). - if rec.done != nil { - close(rec.done) - continue - } - // Every write below goes to the user's writer, which may panic — an - // unrecovered goroutine panic is fatal to the process, so contain each one. - contain(func() { - sink.forward(rec.emittedAt, rec.level, rec.target, rec.message) - }) - // Surface log loss the first time the sink falls behind — from this - // goroutine, never the kernel thread. One-shot so a burst can't turn into - // log spam; the running total stays available via kernelLogDropped(). This - // writes to the same destination as forward, so it is contained too. - if !warnedDrop && logDropped.Load() > baselineDrops { - warnedDrop = true - dropped := logDropped.Load() - baselineDrops - contain(func() { - logger.Logger.Warn().Uint64("dropped", dropped).Msg( - "[kernel] kernel log records dropped; the log sink is not keeping up " + - "(raise capacity or lower kernel verbosity)") - }) - } - } -} - -// contain runs fn, swallowing any panic. A misbehaving user writer (reached via -// the shared logger) must never take down the drain goroutine. -func contain(fn func()) { - defer func() { _ = recover() }() - fn() -} - -// flushKernelLogs blocks until every kernel record already queued has been -// written, or until timeout elapses; it returns whether the flush completed. It -// is a no-op returning true when kernel logging was never installed. -// -// It drains the asynchronous hand-off so records are not lost or misrouted when -// the log writer is closed, output is retargeted, or the process exits. There is -// currently no public entry point that calls it — only the end-to-end test does; -// exposing a supported flush API (or wiring it into a shutdown path) is a separate -// change. Best-effort: records already dropped for a full buffer are gone, and -// records enqueued after this call are not waited on. -func flushKernelLogs(timeout time.Duration) bool { - qp := logQueue.Load() - if qp == nil { - return true - } - timer := time.NewTimer(timeout) - defer timer.Stop() - - done := make(chan struct{}) - // Enqueue the barrier behind everything already queued. A full buffer means - // the drain is behind; wait (bounded) for room rather than dropping the barrier. - select { - case *qp <- kernelLogRecord{done: done}: - case <-timer.C: - return false - } - select { - case <-done: - return true - case <-timer.C: - return false - } -} - func installKernelLogCallback(level string, useNULL bool) { logCallbackOnce.Do(func() { // OFF intentionally installs no subscriber and starts no drain. @@ -186,8 +66,8 @@ func installKernelLogCallback(level string, useNULL bool) { clevel = newCStr(level) defer clevel.free() } - // NULL user_data: the drain goroutine owns the sink, so nothing - // Go-managed crosses into C as a pointer. + // NULL user_data: the drain goroutine owns the sink, so nothing Go-managed + // crosses into C as a pointer. if err := call(func() C.KernelStatusCode { return C.kernel_init_logging_callback(clevel.c, C.kernel_log_cb(), nil) }); err != nil { diff --git a/internal/backend/kernel/log_callback_test.go b/internal/backend/kernel/log_callback_test.go index d98a07de..e3247a92 100644 --- a/internal/backend/kernel/log_callback_test.go +++ b/internal/backend/kernel/log_callback_test.go @@ -3,7 +3,6 @@ package kernel import ( - "io" "os" "os/exec" "path/filepath" @@ -86,152 +85,3 @@ func TestKernelCallbackWritesConfiguredFileEndToEnd(t *testing.T) { t.Errorf("local log file is missing Rust kernel record: %q", got) } } - -// The forward path enqueues an owned record onto the bounded channel. (The cgo -// trampoline copies the borrowed C strings via C.GoString, then calls this; that -// tiny boundary can't be driven from a _test.go file because import "C" is -// disallowed there, so the testable logic lives in enqueueKernelLog.) -func TestEnqueueKernelLog(t *testing.T) { - ch := make(chan kernelLogRecord, 1) - prev := logQueue.Swap(&ch) - t.Cleanup(func() { logQueue.Store(prev) }) - - emittedAt := time.Now() - enqueueKernelLog(emittedAt, "debug", "databricks::sql::kernel", "callback probe") - select { - case got := <-ch: - if got.level != "debug" || got.target != "databricks::sql::kernel" || got.message != "callback probe" { - t.Fatalf("enqueued record = %#v", got) - } - if !got.emittedAt.Equal(emittedAt) { - t.Fatalf("emittedAt = %v, want %v", got.emittedAt, emittedAt) - } - default: - t.Fatal("record was not enqueued") - } -} - -// A full buffer drops the record and counts it rather than blocking the kernel -// thread. A nil queue (logging not installed) is a safe no-op. -func TestEnqueueKernelLogDropsWhenFullAndNoopWhenUnset(t *testing.T) { - // Unset queue: must not block or panic. - prev := logQueue.Swap(nil) - t.Cleanup(func() { logQueue.Store(prev) }) - enqueueKernelLog(time.Now(), "warn", "t", "before install") - - ch := make(chan kernelLogRecord, 1) - logQueue.Store(&ch) - before := kernelLogDropped() - enqueueKernelLog(time.Now(), "warn", "t", "keeps the one slot") // fills the buffer - enqueueKernelLog(time.Now(), "warn", "t", "must be dropped") // buffer full → dropped - if got := kernelLogDropped() - before; got != 1 { - t.Fatalf("dropped delta = %d, want 1", got) - } - if len(ch) != 1 { - t.Fatalf("channel len = %d, want 1 (non-blocking drop)", len(ch)) - } -} - -// A panicking writer must not kill the drain goroutine (an unrecovered goroutine -// panic is fatal to the process); the drain contains it and keeps processing. -func TestDrainRecoversFromWriterPanic(t *testing.T) { - t.Cleanup(func() { logger.SetLogOutput(os.Stderr) }) - logger.SetLogOutput(io.Discard) // discard the sink's forwarded records - done := make(chan string, 1) - sink := newLogSink() - sink.observe = func(_, _, message string) { - if message == "boom" { - panic("writer failure") - } - done <- message - } - - ch := make(chan kernelLogRecord, 2) - go drainKernelLogs(ch, sink) - defer close(ch) - - ch <- kernelLogRecord{level: "error", target: "t", message: "boom"} // triggers the panic - ch <- kernelLogRecord{level: "info", target: "t", message: "after"} // must still be delivered - select { - case got := <-done: - if got != "after" { - t.Fatalf("delivered %q, want %q", got, "after") - } - case <-time.After(2 * time.Second): - t.Fatal("drain goroutine did not survive a writer panic") - } -} - -// flushKernelLogs returns only after every record queued before it is written. -func TestFlushKernelLogsWaitsForQueued(t *testing.T) { - t.Cleanup(func() { logger.SetLogOutput(os.Stderr) }) - logger.SetLogOutput(io.Discard) // discard the sink's forwarded records - seen := make(chan string, 8) - sink := newLogSink() - sink.observe = func(_, _, message string) { seen <- message } - - ch := make(chan kernelLogRecord, 8) - prev := logQueue.Swap(&ch) - t.Cleanup(func() { logQueue.Store(prev) }) - go drainKernelLogs(ch, sink) - defer close(ch) - - enqueueKernelLog(time.Now(), "info", "t", "one") - enqueueKernelLog(time.Now(), "info", "t", "two") - if !flushKernelLogs(2 * time.Second) { - t.Fatal("flush timed out") - } - // The barrier is FIFO-ordered behind both records, so both are delivered. - if got := len(seen); got != 2 { - t.Fatalf("after flush, delivered %d records, want 2", got) - } -} - -// flushKernelLogs is a no-op returning true when logging was never installed. -func TestFlushKernelLogsNoopWhenUnset(t *testing.T) { - prev := logQueue.Swap(nil) - t.Cleanup(func() { logQueue.Store(prev) }) - if !flushKernelLogs(time.Second) { - t.Fatal("flush with no queue should be a no-op returning true") - } -} - -type panicWriter struct{} - -func (panicWriter) Write([]byte) (int, error) { panic("writer always panics") } - -// A panicking writer combined with dropped records must not crash the drain: the -// forwarded record AND the one-shot drop warning both write to that writer, and -// both must be contained. observe advances the drop counter past the drain's -// baseline (so the warning fires) and panics (so forward is exercised too); the -// shared output is the panicking writer (so the warning write panics). -func TestDrainContainsPanicFromForwardAndDropWarning(t *testing.T) { - prevLevel := logger.Logger.GetLevel() - t.Cleanup(func() { - logger.SetLogOutput(os.Stderr) - logger.Logger.Logger = logger.Logger.Level(prevLevel) - }) - if err := logger.SetLogLevel("warn"); err != nil { - t.Fatal(err) - } - logger.SetLogOutput(panicWriter{}) - - sink := newLogSink() - sink.observe = func(_, _, _ string) { - logDropped.Add(1) // advance past the drain's baseline, deterministically - panic("forward failure") - } - - ch := make(chan kernelLogRecord, 4) - prev := logQueue.Swap(&ch) - t.Cleanup(func() { logQueue.Store(prev) }) - go drainKernelLogs(ch, sink) - defer close(ch) - - enqueueKernelLog(time.Now(), "info", "databricks::sql::kernel", "boom") - - // The drain reaches the flush barrier only if it survived both panics. - if !flushKernelLogs(2 * time.Second) { - t.Fatal("drain did not survive a panicking writer combined with dropped records") - } -} diff --git a/internal/backend/kernel/logforward.go b/internal/backend/kernel/logforward.go index 7ae810e2..673de63a 100644 --- a/internal/backend/kernel/logforward.go +++ b/internal/backend/kernel/logforward.go @@ -37,6 +37,19 @@ func (s *logSink) forward(emittedAt time.Time, level, target, message string) { s.event(level).Time(zerolog.TimestampFieldName, emittedAt).Str("target", target).Msg(message) } +// warnDropped emits a one-shot advisory that forwarded records were dropped. It +// goes through the sink's own immutable logger — not logger.Logger, whose embedded +// value SetLogLevel reassigns — so the long-lived drain goroutine never races +// SetLogLevel. Like forwarded records it is ungated, which is what we want: log loss +// should surface regardless of the driver level. +func (s *logSink) warnDropped(dropped uint64) { + s.sink.Event(zerolog.WarnLevel). + Uint64("dropped", dropped). + Time(zerolog.TimestampFieldName, time.Now()). + Msg("[kernel] kernel log records dropped; the log sink is not keeping up " + + "(raise capacity or lower kernel verbosity)") +} + // event picks the zerolog event for a kernel level string. An unknown level maps // to Debug and preserves the raw kernel level as a field. func (s *logSink) event(level string) *zerolog.Event { diff --git a/internal/backend/kernel/logforward_async.go b/internal/backend/kernel/logforward_async.go new file mode 100644 index 00000000..e1b080ba --- /dev/null +++ b/internal/backend/kernel/logforward_async.go @@ -0,0 +1,136 @@ +package kernel + +// This file is intentionally NOT behind the `cgo && databricks_kernel` build tag +// (matching logging_level.go and logforward.go). It holds the pure-Go async +// forwarding pipeline — the bounded hand-off queue, its drain, the flush barrier, +// drop accounting, and panic containment. Only the cgo trampoline and the +// kernel_init_logging_callback call actually need cgo; keeping the rest untagged +// lets its tests (FIFO flush, drop policy, panic containment) run in the default +// CGO_ENABLED=0 build rather than only in the kernel-linked lane. + +import ( + "sync/atomic" + "time" +) + +// kernelLogRecord is an owned copy of one kernel tracing record. The C strings are +// valid only for the duration of the callback, so the trampoline copies them before +// the record leaves the kernel thread. A record with a non-nil done channel is a +// flush barrier: the drain closes done (in FIFO order, after every earlier record is +// written) and forwards nothing — see flushKernelLogs. +type kernelLogRecord struct { + emittedAt time.Time + level string + target string + message string + done chan struct{} +} + +// kernelLogChannelCapacity bounds the hand-off buffer between kernel threads and the +// drain goroutine. Bursts beyond this are dropped rather than blocking a kernel +// thread — logs are advisory and must never back-pressure a kernel path. +const kernelLogChannelCapacity = 4096 + +var ( + // logQueue publishes the bounded hand-off channel to the trampoline. It is an + // atomic pointer so the read on a kernel thread synchronizes with the write on + // the installing goroutine; nil until (and unless) logging installs. + logQueue atomic.Pointer[chan kernelLogRecord] + // logDropped counts records discarded because the buffer was full — a growing + // value means the sink cannot keep up. Exposed via kernelLogDropped. + logDropped atomic.Uint64 +) + +// kernelLogDropped reports how many forwarded kernel records were dropped because +// the bounded hand-off buffer was full. Safe to call at any time. +func kernelLogDropped() uint64 { return logDropped.Load() } + +// enqueueKernelLog hands one already-owned record to the drain goroutine without +// blocking. Split out of the cgo trampoline so the enqueue/drop policy is testable +// without cgo (import "C" is not allowed in _test.go files). +func enqueueKernelLog(emittedAt time.Time, level, target, message string) { + qp := logQueue.Load() + if qp == nil { + return + } + rec := kernelLogRecord{emittedAt: emittedAt, level: level, target: target, message: message} + // Non-blocking hand-off: never stall a kernel thread on a slow, contended, or + // re-entrant user writer. A full buffer drops the record and counts it. + select { + case *qp <- rec: + default: + logDropped.Add(1) + } +} + +// drainKernelLogs is the single goroutine that moves records off kernel threads and +// into the shared logger. Running the arbitrary user writer here — not in the +// trampoline — keeps user I/O (and any driver re-entry it triggers) off the kernel +// thread, honoring the C ABI's "return promptly / no re-entry" contract. +func drainKernelLogs(ch <-chan kernelLogRecord, sink *logSink) { + baselineDrops := logDropped.Load() + warnedDrop := false + for rec := range ch { + // A flush barrier carries no record: closing done signals that every earlier + // record has been written (channel + drain are FIFO). + if rec.done != nil { + close(rec.done) + continue + } + // Every write below goes to the user's writer, which may panic — an + // unrecovered goroutine panic is fatal to the process, so contain each one. + contain(func() { + sink.forward(rec.emittedAt, rec.level, rec.target, rec.message) + }) + // Surface log loss the first time the sink falls behind. Routed through the + // sink's own (immutable) logger, not logger.Logger, so this long-lived + // goroutine never races SetLogLevel's reassignment of Logger.Logger. One-shot + // so a burst can't turn into log spam; the total stays in kernelLogDropped(). + if !warnedDrop && logDropped.Load() > baselineDrops { + warnedDrop = true + dropped := logDropped.Load() - baselineDrops + contain(func() { sink.warnDropped(dropped) }) + } + } +} + +// contain runs fn, swallowing any panic. A misbehaving user writer (reached via the +// sink) must never take down the drain goroutine. +func contain(fn func()) { + defer func() { _ = recover() }() + fn() +} + +// flushKernelLogs blocks until every kernel record already queued has been written, +// or until timeout elapses; it returns whether the flush completed. It is a no-op +// returning true when kernel logging was never installed. +// +// It drains the asynchronous hand-off so records are not lost or misrouted when the +// log writer is closed, output is retargeted, or the process exits. There is +// currently no public entry point that calls it — only the end-to-end test does; +// exposing a supported flush API (or wiring it into a shutdown path) is a separate +// change. Best-effort: records already dropped for a full buffer are gone, and +// records enqueued after this call are not waited on. +func flushKernelLogs(timeout time.Duration) bool { + qp := logQueue.Load() + if qp == nil { + return true + } + timer := time.NewTimer(timeout) + defer timer.Stop() + + done := make(chan struct{}) + // Enqueue the barrier behind everything already queued. A full buffer means the + // drain is behind; wait (bounded) for room rather than dropping the barrier. + select { + case *qp <- kernelLogRecord{done: done}: + case <-timer.C: + return false + } + select { + case <-done: + return true + case <-timer.C: + return false + } +} diff --git a/internal/backend/kernel/logforward_async_test.go b/internal/backend/kernel/logforward_async_test.go new file mode 100644 index 00000000..e18b3c42 --- /dev/null +++ b/internal/backend/kernel/logforward_async_test.go @@ -0,0 +1,165 @@ +package kernel + +// These exercise the pure-Go async forwarding pipeline (logforward_async.go) and +// are intentionally untagged, so the FIFO-flush, drop-policy, and panic-containment +// guarantees run in the default CGO_ENABLED=0 build — not only the kernel-linked +// lane. The cgo trampoline that feeds this pipeline is covered by the end-to-end +// test in log_callback_test.go. + +import ( + "io" + "os" + "testing" + "time" + + "github.com/databricks/databricks-sql-go/logger" +) + +// The forward path enqueues an owned record onto the bounded channel. (The cgo +// trampoline copies the borrowed C strings via C.GoString, then calls this; that +// tiny boundary can't be driven from a _test.go file because import "C" is +// disallowed there, so the testable logic lives in enqueueKernelLog.) +func TestEnqueueKernelLog(t *testing.T) { + ch := make(chan kernelLogRecord, 1) + prev := logQueue.Swap(&ch) + t.Cleanup(func() { logQueue.Store(prev) }) + + emittedAt := time.Now() + enqueueKernelLog(emittedAt, "debug", "databricks::sql::kernel", "callback probe") + select { + case got := <-ch: + if got.level != "debug" || got.target != "databricks::sql::kernel" || got.message != "callback probe" { + t.Fatalf("enqueued record = %#v", got) + } + if !got.emittedAt.Equal(emittedAt) { + t.Fatalf("emittedAt = %v, want %v", got.emittedAt, emittedAt) + } + default: + t.Fatal("record was not enqueued") + } +} + +// A full buffer drops the record and counts it rather than blocking the kernel +// thread. A nil queue (logging not installed) is a safe no-op. +func TestEnqueueKernelLogDropsWhenFullAndNoopWhenUnset(t *testing.T) { + // Unset queue: must not block or panic. + prev := logQueue.Swap(nil) + t.Cleanup(func() { logQueue.Store(prev) }) + enqueueKernelLog(time.Now(), "warn", "t", "before install") + + ch := make(chan kernelLogRecord, 1) + logQueue.Store(&ch) + before := kernelLogDropped() + enqueueKernelLog(time.Now(), "warn", "t", "keeps the one slot") // fills the buffer + enqueueKernelLog(time.Now(), "warn", "t", "must be dropped") // buffer full → dropped + if got := kernelLogDropped() - before; got != 1 { + t.Fatalf("dropped delta = %d, want 1", got) + } + if len(ch) != 1 { + t.Fatalf("channel len = %d, want 1 (non-blocking drop)", len(ch)) + } +} + +// A panicking writer must not kill the drain goroutine (an unrecovered goroutine +// panic is fatal to the process); the drain contains it and keeps processing. +func TestDrainRecoversFromWriterPanic(t *testing.T) { + t.Cleanup(func() { logger.SetLogOutput(os.Stderr) }) + logger.SetLogOutput(io.Discard) // discard the sink's forwarded records + done := make(chan string, 1) + sink := newLogSink() + sink.observe = func(_, _, message string) { + if message == "boom" { + panic("writer failure") + } + done <- message + } + + ch := make(chan kernelLogRecord, 2) + go drainKernelLogs(ch, sink) + defer close(ch) + + ch <- kernelLogRecord{level: "error", target: "t", message: "boom"} // triggers the panic + ch <- kernelLogRecord{level: "info", target: "t", message: "after"} // must still be delivered + select { + case got := <-done: + if got != "after" { + t.Fatalf("delivered %q, want %q", got, "after") + } + case <-time.After(2 * time.Second): + t.Fatal("drain goroutine did not survive a writer panic") + } +} + +// flushKernelLogs returns only after every record queued before it is written. +func TestFlushKernelLogsWaitsForQueued(t *testing.T) { + t.Cleanup(func() { logger.SetLogOutput(os.Stderr) }) + logger.SetLogOutput(io.Discard) // discard the sink's forwarded records + seen := make(chan string, 8) + sink := newLogSink() + sink.observe = func(_, _, message string) { seen <- message } + + ch := make(chan kernelLogRecord, 8) + prev := logQueue.Swap(&ch) + t.Cleanup(func() { logQueue.Store(prev) }) + go drainKernelLogs(ch, sink) + defer close(ch) + + enqueueKernelLog(time.Now(), "info", "t", "one") + enqueueKernelLog(time.Now(), "info", "t", "two") + if !flushKernelLogs(2 * time.Second) { + t.Fatal("flush timed out") + } + // The barrier is FIFO-ordered behind both records, so both are delivered. + if got := len(seen); got != 2 { + t.Fatalf("after flush, delivered %d records, want 2", got) + } +} + +// flushKernelLogs is a no-op returning true when logging was never installed. +func TestFlushKernelLogsNoopWhenUnset(t *testing.T) { + prev := logQueue.Swap(nil) + t.Cleanup(func() { logQueue.Store(prev) }) + if !flushKernelLogs(time.Second) { + t.Fatal("flush with no queue should be a no-op returning true") + } +} + +type panicWriter struct{} + +func (panicWriter) Write([]byte) (int, error) { panic("writer always panics") } + +// A panicking writer combined with dropped records must not crash the drain: the +// forwarded record AND the one-shot drop warning both write to that writer, and +// both must be contained. observe advances the drop counter past the drain's +// baseline (so the warning fires) and panics (so forward is exercised too); the +// shared output is the panicking writer (so the warning write panics). +func TestDrainContainsPanicFromForwardAndDropWarning(t *testing.T) { + prevLevel := logger.Logger.GetLevel() + t.Cleanup(func() { + logger.SetLogOutput(os.Stderr) + logger.Logger.Logger = logger.Logger.Level(prevLevel) + }) + if err := logger.SetLogLevel("warn"); err != nil { + t.Fatal(err) + } + logger.SetLogOutput(panicWriter{}) + + sink := newLogSink() + sink.observe = func(_, _, _ string) { + logDropped.Add(1) // advance past the drain's baseline, deterministically + panic("forward failure") + } + + ch := make(chan kernelLogRecord, 4) + prev := logQueue.Swap(&ch) + t.Cleanup(func() { logQueue.Store(prev) }) + go drainKernelLogs(ch, sink) + defer close(ch) + + enqueueKernelLog(time.Now(), "info", "databricks::sql::kernel", "boom") + + // The drain reaches the flush barrier only if it survived both panics. + if !flushKernelLogs(2 * time.Second) { + t.Fatal("drain did not survive a panicking writer combined with dropped records") + } +}