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

connector.go

@@ -620,6 +620,13 @@ func WithKernelClientCertificate(cert, key []byte) ConnOption {
 		// caller mutating the buffers between NewConnector and Connect can't change
 		// the identity out from under us.
 		k := kernelExperimental(c)
+		// Record that mTLS was explicitly requested, independent of whether the
+		// caller passed non-empty bytes. Without this marker an empty cert+key
+		// pair (e.g. from a failed PEM load) is indistinguishable from the option
+		// never being called, and validateKernelConfig would accept it while
+		// applyKernelTLS silently skipped the setter — connecting with no client
+		// identity. The marker lets validation reject the incomplete request loudly.
+		k.TLSClientCertConfigured = true
 		if len(cert) > 0 {
 			k.TLSClientCertPEM = append([]byte(nil), cert...)
 		} else {

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/abi.go

@@ -0,0 +1,22 @@
+package kernel
+
+import "fmt"
+
+// compareABI reports whether the C-ABI version compiled into the linked kernel
+// library (got) matches the version the driver's header declares (want): it
+// returns a descriptive, operator-facing error on mismatch and nil when they
+// agree.
+//
+// It is split out from checkABIVersion (in the cgo-tagged cgo.go) as pure,
+// cgo-free logic so the mismatch verdict and its message are unit-testable under
+// the default CGO_ENABLED=0 build. The negative path can't otherwise be reached
+// from a test: a real build always links a .a and header produced together, and
+// abiVersions() reads fixed cgo constants with no injection seam. checkABIVersion
+// runs abiVersions() through this on every connect.
+func compareABI(got, want uint32) error {
+	if got != want {
+		return fmt.Errorf("databricks: kernel ABI version mismatch: linked library reports %d, "+
+			"driver header expects %d; rebuild the kernel static lib and header together (make kernel-lib)", got, want)
+	}
+	return nil
+}

internal/backend/kernel/abi_test.go

@@ -0,0 +1,37 @@
+package kernel
+
+import (
+	"strings"
+	"testing"
+)
+
+// TestCompareABI covers both verdicts of the ABI-version handshake without the
+// cgo symbols, so it runs under the default CGO_ENABLED=0 build. The matching
+// (happy) path is additionally proven against the real linked library by
+// TestABIVersionMatches in the cgo build; the mismatch path — the runtime hazard
+// the check exists for (a driver header linked against a differently-built
+// prebuilt .a) — is otherwise unreachable from a test, since a real build always
+// links a matching .a + header.
+func TestCompareABI(t *testing.T) {
+	// Matching versions must produce no error: checkABIVersion opens the session.
+	for _, v := range []uint32{0, 1, 42} {
+		if err := compareABI(v, v); err != nil {
+			t.Errorf("compareABI(%d, %d) = %v, want nil for matching versions", v, v, err)
+		}
+	}
+
+	// Mismatched versions must produce an error so checkABIVersion refuses to open
+	// (rather than silently misread status codes / error-struct fields). The
+	// message must name both versions and the remediation so an operator hitting a
+	// stale prebuilt .a knows what to rebuild.
+	err := compareABI(2, 1)
+	if err == nil {
+		t.Fatal("compareABI(2, 1) = nil, want an error for mismatched versions")
+	}
+	msg := err.Error()
+	for _, want := range []string{"ABI version mismatch", "reports 2", "expects 1", "make kernel-lib"} {
+		if !strings.Contains(msg, want) {
+			t.Errorf("mismatch error %q does not contain %q", msg, want)
+		}
+	}
+}

internal/backend/kernel/cgo.go

@@ -101,56 +101,45 @@ 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.
 //
-// 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.
+// 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.
 //
-// Scope caveat: the kernel subscriber is PROCESS-WIDE, first-call-wins, and never
+// 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 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 "" for the
+	// DBSQL_KERNEL_DEBUG override, so the kernel honors RUST_LOG). The pure
+	// decision lives in logging_level.go so it's unit-tested without cgo.
+	level, _ := resolveKernelLogArg()
+	// 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
@@ -169,15 +158,14 @@ var (
 // download-a-prebuilt-lib distribution path). A mismatch means the
 // KernelStatusCode enum / KernelError struct layout the driver reads may not
 // match what the library writes, so we refuse to open rather than silently
-// misread status codes / error fields. Runs once; the result is cached.
+// misread status codes / error fields. Runs once; the result is cached. The
+// pure got-vs-want verdict lives in compareABI (untagged) so its mismatch path
+// is testable without the cgo symbols.
 func checkABIVersion() error {
 	abiCheckOnce.Do(func() {
 		got, want := abiVersions()
 		klog("checkABIVersion got=%d want=%d", got, want)
-		if got != want {
-			abiCheckErr = fmt.Errorf("databricks: kernel ABI version mismatch: linked library reports %d, "+
-				"driver header expects %d; rebuild the kernel static lib and header together (make kernel-lib)", got, want)
-		}
+		abiCheckErr = compareABI(got, want)
 	})
 	return abiCheckErr
 }

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 kernel's own
+// integration tests).
+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.