feat(kernel): forward kernel logs into the driver logger via callback

doc.go

@@ -195,19 +195,31 @@ 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, which is
+routed into the driver's logger via a reverse-call callback (kernel_set_log_callback):
+kernel-internal tracing events are forwarded into the SAME unified sink as the Go
+binding lines — including a custom logger.SetLogOutput — rather than going to stderr.
+One caveat specific to the Rust lines: BOTH their verbosity AND their output
+destination are captured when the first kernel session in the process is opened (the
+forwarding sink snapshots the logger then). A later dbsql.SetLogLevel or
+logger.SetLogOutput re-targets the Go binding lines but NOT the already-forwarded
+kernel lines, so set the level and any custom output before that first connect to
+govern them.
 
 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 its level 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
+	# kernel logs at the kernel's own verbosity, forwarded into the driver logger:
+	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
+
+The kernel exposes one process-global tracing subscriber reachable two mutually
+exclusive ways over the C ABI (kernel_init_logging → stderr, kernel_set_log_callback
+→ the driver logger); the driver installs the callback. A host that has already
+installed its own global tracing subscriber will see the registration report a
+benign, logged failure and kernel events simply won't forward.
 
 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

@@ -101,56 +101,50 @@ 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) logs into the driver's
+// logger 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 kernel exposes exactly one process-global tracing subscriber, reachable
+// over the C ABI two mutually-exclusive ways: kernel_init_logging (RUST_LOG →
+// stderr) and kernel_set_log_callback (reverse-call → a host sink). We install
+// the CALLBACK (see installKernelLogCallback) so kernel-internal events flow into
+// logger.Logger — the SAME unified stream as the Go binding lines (klog/klogCtx)
+// and honouring logger.SetLogOutput — instead of only reaching stderr. This is
+// the completion of the one-knob logging story (PECOBLR-3650 shipped the level
+// unification against the stderr sink; K4 replaces that sink with the callback).
+//
+// resolveKernelLogArg maps the driver's zerolog level to the kernel's
+// OFF/ERROR/WARN/INFO/DEBUG/TRACE string; DBSQL_KERNEL_DEBUG instead defers to
+// RUST_LOG (useNULL → an empty level string → NULL to the kernel, the advanced
+// override for tuning kernel-only verbosity). The callback filters at that level
+// with the same precedence kernel_init_logging used, so the swap is level-transparent.
 //
-// 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: a non-Success install (e.g. the host already installed a global
+// subscriber) is a documented, benign outcome — logged, never fatal to connect
+// (handled inside installKernelLogCallback). A driver left at the default Warn
+// level (benchmarks included, DBSQL_KERNEL_DEBUG unset) installs at WARN, so the
+// kernel emits nothing below Warn and there is no hot-path cost.
 //
-// Scope caveat: the kernel subscriber is PROCESS-WIDE, first-call-wins, and never
+// Scope caveat: the 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 subsequent kernel sessions. 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.
 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)
-		}
-	})
+	// resolveKernelLogArg decides the level string (or the DBSQL_KERNEL_DEBUG
+	// override → useNULL → "" → NULL, so the kernel honors RUST_LOG). The pure
+	// decision lives in logging_level.go so it's unit-tested without cgo.
+	level, useNULL := resolveKernelLogArg()
+	if useNULL {
+		level = ""
+	}
+	// installKernelLogCallback self-guards with logCallbackOnce (process-wide,
+	// first-call-wins), so no extra sync.Once here.
+	installKernelLogCallback(level)
 }
 
 // abiCheckOnce ensures the ABI-version handshake runs at most once per process

internal/backend/kernel/kernel_test.go

@@ -9,6 +9,7 @@ import (
 	"encoding/json"
 	"errors"
 	"os"
+	"runtime/cgo"
 	"testing"
 
 	"github.com/databricks/databricks-sql-go/driverctx"
@@ -100,6 +101,110 @@ func TestABIVersionMatches(t *testing.T) {
 	}
 }
 
+// TestLogCallbackRoundTrip exercises the full reverse-call path: wrap an
+// observable *logSink in a cgo.Handle, drive the C→Go trampoline exactly as the
+// kernel's drain thread would (via the test seam), and assert the event arrived
+// with the handle correctly unwrapped and the fields intact. This proves the
+// cgo.Handle round-trip + recover firewall + sink dispatch that a real kernel log
+// event flows through (kernel-side delivery itself is covered by the Rust
+// integration test v0_c_abi_log_callback.rs and the C smoke).
+func TestLogCallbackRoundTrip(t *testing.T) {
+	type got struct {
+		level           int
+		target, message string
+	}
+	ch := make(chan got, 1)
+	sink := &logSink{observe: func(level int, target, message string) {
+		ch <- got{level, target, message}
+	}}
+	h := cgo.NewHandle(sink)
+	defer h.Delete()
+
+	invokeLogTrampolineForTest(h, kernelLevelWarn, "databricks::sql::kernel", "retrying request")
+
+	select {
+	case g := <-ch:
+		if g.level != kernelLevelWarn || g.target != "databricks::sql::kernel" || g.message != "retrying request" {
+			t.Errorf("delivered event = %+v, want level=WARN target=databricks::sql::kernel msg='retrying request'", g)
+		}
+	default:
+		t.Fatal("the trampoline did not deliver the event to the sink")
+	}
+}
+
+// captureDriverLog points the global driver logger at a buffer (TraceLevel) for
+// the duration of a test, returning the buffer and restoring the logger after.
+// Used to assert a trampoline guard branch delivered NOTHING (the branches bail
+// before reaching a sink, so there's no observe hook to check — the driver
+// logger output is the observable).
+func captureDriverLog(t *testing.T) *bytes.Buffer {
+	t.Helper()
+	var buf bytes.Buffer
+	prev := logger.Logger.GetLevel()
+	logger.SetLogOutput(&buf)
+	_ = logger.SetLogLevel("trace")
+	t.Cleanup(func() {
+		logger.SetLogOutput(os.Stderr)
+		logger.Logger.Logger = logger.Logger.Level(prev)
+	})
+	return &buf
+}
+
+// A zero ctx must be a no-op (no crash, no delivery). cgo.Handle(0) casts to a
+// NULL void* ctx, so the trampoline returns at its `ctx == nil` guard before ever
+// calling Value() (calling Value() on the zero handle would itself panic — the
+// recover firewall would catch it, but the nil guard means we never reach it).
+func TestLogCallbackTrampolineNilCtxSafe(t *testing.T) {
+	buf := captureDriverLog(t)
+	invokeLogTrampolineForTest(cgo.Handle(0), kernelLevelError, "databricks::sql::kernel", "should-not-appear")
+	if buf.Len() != 0 {
+		t.Errorf("nil-ctx trampoline delivered a line, want none: %q", buf.String())
+	}
+}
+
+// TestLogCallbackTrampolineWrongTypeSafe covers the "handle holds a non-*logSink
+// value" branch: a live, non-nil handle whose Value() type-asserts to false. The
+// trampoline must return without panicking AND without delivering (asserted via
+// the captured driver logger staying empty).
+func TestLogCallbackTrampolineWrongTypeSafe(t *testing.T) {
+	buf := captureDriverLog(t)
+	h := cgo.NewHandle("not a logSink")
+	defer h.Delete()
+	invokeLogTrampolineForTest(h, kernelLevelError, "databricks::sql::kernel", "should-not-appear")
+	if buf.Len() != 0 {
+		t.Errorf("wrong-type trampoline delivered a line, want none: %q", buf.String())
+	}
+}
+
+// TestLogCallbackPanicFirewall is the load-bearing safety test: a sink whose
+// routing panics must NOT crash the process — the trampoline's defer recover()
+// converts it into a dropped line (a panic unwinding across the cgo boundary
+// would abort). We drive a panicking observe hook, assert the call returns, and
+// assert a subsequent well-formed sink still delivers (the firewall didn't wedge
+// anything process-global).
+func TestLogCallbackPanicFirewall(t *testing.T) {
+	panicSink := &logSink{observe: func(int, string, string) { panic("boom in the sink") }}
+	ph := cgo.NewHandle(panicSink)
+	defer ph.Delete()
+	// Must not crash the test binary.
+	invokeLogTrampolineForTest(ph, kernelLevelError, "databricks::sql::kernel", "will panic")
+
+	// A subsequent well-formed delivery still works.
+	ch := make(chan string, 1)
+	okSink := &logSink{observe: func(_ int, _, message string) { ch <- message }}
+	oh := cgo.NewHandle(okSink)
+	defer oh.Delete()
+	invokeLogTrampolineForTest(oh, kernelLevelWarn, "databricks::sql::kernel", "recovered")
+	select {
+	case got := <-ch:
+		if got != "recovered" {
+			t.Errorf("post-panic delivery = %q, want 'recovered'", got)
+		}
+	default:
+		t.Fatal("delivery after a panicking callback did not work — firewall wedged")
+	}
+}
+
 // TestKernelLogLevel and TestResolveKernelLogArg — the pure level-resolution tests —
 // live in the untagged logging_level_test.go so they run under CGO_ENABLED=0. The
 // tests below exercise klog/klogCtx and so need the cgo build.

internal/backend/kernel/log_callback.go

@@ -0,0 +1,136 @@
+//go:build cgo && databricks_kernel
+
+package kernel
+
+/*
+#include <stdlib.h>
+#include "databricks_kernel.h"
+
+// Adapter matching the kernel_log_callback typedef (const char*) that forwards
+// to the //export'd Go trampoline. cgo generates the trampoline's prototype in
+// _cgo_export.h with non-const char* params, which don't match the typedef, so
+// this shim bridges the two and gives kernel_set_log_callback one clean address
+// to take. Forward-declared with the SAME (char*) signature cgo generates to
+// avoid a conflicting-declaration error; the kernel only ever reads the strings.
+void kernelLogTrampoline(void* ctx, int level, char* target, char* message);
+// Hand kernel_set_log_callback the trampoline's address as a kernel_log_callback.
+// The //export'd trampoline's generated signature uses non-const char* while the
+// typedef uses const char*, which are function-pointer compatible (a const-only
+// difference); do the cast once here, on the C side, so the Go call site stays
+// clean and there is no separately-linked adapter symbol.
+static kernel_log_callback kernel_log_cb(void) {
+    return (kernel_log_callback)kernelLogTrampoline;
+}
+// Cast a cgo.Handle (an integer token) to the opaque void* ctx on the C side,
+// so Go never does unsafe.Pointer(uintptr) directly (which `go vet` flags).
+static void* kernel_handle_to_ctx(uintptr_t h) { return (void*)h; }
+// Test-only: invoke the trampoline exactly as the kernel drain thread would.
+// (A C func-pointer value can't be called directly from Go, so a helper does it.)
+static void kernel_invoke_trampoline_for_test(void* ctx, int level, char* target, char* message) {
+    kernelLogTrampoline(ctx, level, target, message);
+}
+*/
+import "C"
+
+import (
+	"runtime/cgo"
+	"sync"
+	"unsafe"
+
+	"github.com/databricks/databricks-sql-go/logger"
+)
+
+// This file holds the reverse-call machinery for the kernel log callback (K4):
+// a cgo-exported Go function the kernel invokes from its log-drain thread, made
+// safe with a recover() panic firewall (a panic across the cgo boundary would
+// abort the process) and a runtime/cgo.Handle so a Go pointer (the *logSink) can
+// round-trip through the C void* ctx under cgo's pointer rules. This is the same
+// machinery a kernel→host token-provider callback (OAuth U2M external creds)
+// would need.
+
+// logCallbackOnce guards the one-time callback registration; logCallbackHandle
+// keeps the cgo.Handle alive for the process (the kernel holds its numeric value
+// as the opaque ctx and there is no detach in the v0 C ABI, so we never Delete
+// it — a deliberate process-lifetime pin, not a leak).
+var (
+	logCallbackOnce   sync.Once
+	logCallbackHandle cgo.Handle
+)
+
+// invokeLogTrampolineForTest drives the C→Go trampoline exactly as the kernel's
+// drain thread would — allocating C strings, casting a cgo.Handle to void*, and
+// calling through kernel_log_cb() — so a test can assert the full reverse-call
+// round-trip (handle unwrap + recover firewall + sink dispatch) without a live
+// kernel event. Not used in production.
+func invokeLogTrampolineForTest(h cgo.Handle, level int, target, message string) {
+	ct := C.CString(target)
+	defer C.free(unsafe.Pointer(ct))
+	cm := C.CString(message)
+	defer C.free(unsafe.Pointer(cm))
+	ctx := C.kernel_handle_to_ctx(C.uintptr_t(h))
+	C.kernel_invoke_trampoline_for_test(ctx, C.int(level), ct, cm)
+}
+
+//export kernelLogTrampoline
+func kernelLogTrampoline(ctx unsafe.Pointer, level C.int, target, message *C.char) {
+	// Panic firewall: a Go panic unwinding across the cgo boundary aborts the
+	// process. Convert any panic in the sink routing into a dropped log line.
+	defer func() { _ = recover() }()
+	if ctx == nil {
+		return
+	}
+	sink, ok := cgo.Handle(uintptr(ctx)).Value().(*logSink)
+	if !ok || sink == nil {
+		return
+	}
+	// Copy the borrowed C strings out immediately — they are valid only for the
+	// duration of this call.
+	sink.forward(int(level), C.GoString(target), C.GoString(message))
+}
+
+// installKernelLogCallback registers the trampoline as the kernel's log sink,
+// once per process, wrapping a *logSink in a cgo.Handle passed as the opaque ctx.
+// The kernel filters forwarded events at `level` (an OFF/ERROR/WARN/INFO/DEBUG/
+// TRACE string, or "" → NULL to defer to RUST_LOG), which the caller maps from
+// the driver's own log level so kernel Rust lines follow the one
+// DATABRICKS_LOG_LEVEL knob — the same string kernel_init_logging would have
+// received (see initKernelLogging).
+//
+// Best-effort: a non-Success status (e.g. Internal because a global tracing
+// subscriber was already installed) is logged and ignored — kernel logs simply
+// won't flow through the callback, which is a documented, benign outcome. Returns
+// nothing; the caller does not gate connect on it.
+func installKernelLogCallback(level string) {
+	logCallbackOnce.Do(func() {
+		// Snapshot the driver logger at install (TraceLevel, immutable) so the
+		// kernel drain thread never re-gates already-approved events and never
+		// reads the mutable global logger.Logger (see logSink.log).
+		logCallbackHandle = cgo.NewHandle(newLogSink())
+		// Pass the handle (an integer token) as the opaque ctx via a C-side cast,
+		// so Go doesn't do unsafe.Pointer(uintptr) directly.
+		ctx := C.kernel_handle_to_ctx(C.uintptr_t(logCallbackHandle))
+		// An empty level maps to a NULL char* (kernel defers to RUST_LOG); a
+		// non-empty level is passed as a C string the kernel parses with the same
+		// precedence as kernel_init_logging (explicit wins > RUST_LOG > warn).
+		clevel := newCStrOrNull(level)
+		defer clevel.free()
+		if err := call(func() C.KernelStatusCode {
+			return C.kernel_set_log_callback(C.kernel_log_cb(), ctx, clevel.c)
+		}); err != nil {
+			// On a non-Success return the kernel installed no subscriber and did
+			// NOT retain ctx (it stores the sink only on the success path), so the
+			// process-lifetime pin at logCallbackHandle would be a true leak here —
+			// free it. (On success we deliberately never Delete: the kernel holds
+			// the handle value as its opaque ctx for the process lifetime.)
+			logCallbackHandle.Delete()
+			logCallbackHandle = 0
+			// Surface at Warn (visible at the default level), NOT klog: klog is
+			// Debug-gated and no-ops at the default, and this install runs once so
+			// a later level change can't re-surface it. The message names the
+			// benign common cause (a global subscriber already installed) so
+			// on-call doesn't mistake a logging no-op for a connect failure.
+			logger.Logger.Warn().Msgf(
+				"databricks: kernel_set_log_callback: %v (kernel logs not forwarded; proceeding)", err)
+		}
+	})
+}