feat(kernel): forward logs through shared logger

KERNEL_REV

@@ -1 +1 @@
-eff8950428f4e6cc9975c663ec919f334962f7d0
+2dd4739f1c20e3a560bcaf52abed8aab9bd957cf

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

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

internal/backend/kernel/log_callback.go

@@ -0,0 +1,86 @@
+//go:build cgo && databricks_kernel
+
+package kernel
+
+/*
+#include <stdlib.h>
+#include "databricks_kernel.h"
+
+// 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 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 (
+	"sync"
+	"time"
+	"unsafe"
+
+	"github.com/databricks/databricks-sql-go/logger"
+)
+
+// 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.
+
+// logCallbackOnce guards the process-wide, first-call-wins install.
+var logCallbackOnce sync.Once
+
+//export kernelLogTrampoline
+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() }()
+	// 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))
+}
+
+func installKernelLogCallback(level string, useNULL bool) {
+	logCallbackOnce.Do(func() {
+		// OFF intentionally installs no subscriber and starts no drain.
+		if !useNULL && level == "OFF" {
+			return
+		}
+
+		ch := make(chan kernelLogRecord, kernelLogChannelCapacity)
+		// 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
+		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(), nil)
+		}); err != nil {
+			// 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)
+			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())
+	})
+}

internal/backend/kernel/log_callback_test.go

@@ -0,0 +1,87 @@
+//go:build cgo && databricks_kernel
+
+package kernel
+
+import (
+	"os"
+	"os/exec"
+	"path/filepath"
+	"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)
+		}
+
+		// 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)
+		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)
+	}
+}

internal/backend/kernel/logforward.go

@@ -0,0 +1,70 @@
+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. 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 {
+	sink    *logger.ForwardingSink
+	observe func(level, target, message string)
+}
+
+func newLogSink() *logSink {
+	return &logSink{sink: logger.NewForwardingSink()}
+}
+
+// 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)
+	}
+	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 {
+	switch strings.ToLower(level) {
+	case "error":
+		return s.sink.Event(zerolog.ErrorLevel)
+	case "warn":
+		return s.sink.Event(zerolog.WarnLevel)
+	case "info":
+		return s.sink.Event(zerolog.InfoLevel)
+	case "debug":
+		return s.sink.Event(zerolog.DebugLevel)
+	case "trace":
+		return s.sink.Event(zerolog.TraceLevel)
+	default:
+		return s.sink.Event(zerolog.DebugLevel).Str("kernelLevel", level)
+	}
+}