fix(telemetry): preserve unspecified kernel enablement

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 - **Behavior change:** `m2m.NewAuthenticatorWithScopes` now requests exactly the given scopes instead of always appending `all-apis`, so service principals with scoped OAuth secrets (e.g. `sql` only) can authenticate (databricks/databricks-sql-go#476). Empty scopes and `m2m.NewAuthenticator` still request `all-apis`; callers that relied on it being appended should pass no scopes, or add `all-apis` to their list if they need both. The SEA/kernel backend now forwards custom M2M scopes instead of rejecting them.
+- Let the kernel telemetry feature flag decide enablement when `enableTelemetry` is unset; explicit `true` and `false` continue to override it.
 
 ## v1.16.0 (2026-09-24)
 - Upgrade the kernel bindings to v1.1.0; the kernel dependency is now stable.

CONNECTION_PARAMETERS.md

@@ -214,7 +214,7 @@ kernel config.
 
 | DSN parameter | Thrift | Kernel | Default | Notes |
 |---|:---:|:---:|---|---|
-| `enableTelemetry` | ✅ | ✅ | unset (server flag decides wrapper telemetry; kernel telemetry defaults off) | Force Go wrapper telemetry on/off on the Thrift path, overriding the server feature flag. On the kernel path, forwarded to kernel-owned telemetry; unset forwards disabled. |
+| `enableTelemetry` | ✅ | ✅ | unset (server flag decides) | Force telemetry on/off, overriding the server feature flag. On the kernel path, an unset value remains unspecified so the kernel-owned feature flag decides. |
 | `telemetry_batch_size` | ✅ | ✅ | `200` wrapper default; kernel default when unset | Events per batch. Forwarded to the kernel only when explicitly set. |
 | `telemetry_flush_interval` | ✅ | ✅ | `30s` wrapper default; kernel default when unset | Flush interval. Forwarded to the kernel only when explicitly set. |
 | `telemetry_retry_count` | ⚠️ | ⚠️ | — | **Deprecated and ignored** (retries are owned by the HTTP client + circuit breaker); logs a one-time warning. |
@@ -223,7 +223,7 @@ kernel config.
 These telemetry knobs are **DSN-only** — there are no `WithX` connector options for them.
 An app assembled with `NewConnector(...)` options rather than a DSN cannot tune telemetry:
 wrapper telemetry falls back to the server feature flag (since `enableTelemetry` is
-unset), kernel telemetry defaults off, and `telemetry_batch_size` /
+unset), kernel telemetry follows its server feature flag, and `telemetry_batch_size` /
 `telemetry_flush_interval` use their backend defaults.
 
 The Go wrapper telemetry interceptor is skipped on the kernel path so it does not

KERNEL_REV

@@ -1 +1 @@
-80f2aee7d884994d7b0af9a9ea6078872859a9cd
+e5008635d35af1f83a88057c4cf4d97718e31ab4

internal/backend/kernel/backend.go

@@ -29,12 +29,19 @@ static inline KernelStatusCode go_kernel_set_max_connections(
 }
 
 static inline KernelStatusCode go_kernel_set_telemetry_config(
-    KernelSessionConfig* config, bool enabled, size_t batch_size,
+    KernelSessionConfig* config, bool has_enabled, bool enabled, size_t batch_size,
     uint64_t flush_interval_ms, uint32_t max_retries, uint64_t retry_delay_ms,
     uint64_t close_flush_timeout_ms) {
-  return kernel_session_config_set_telemetry_config(
-      config, enabled, batch_size, flush_interval_ms, max_retries, retry_delay_ms,
-      close_flush_timeout_ms);
+  KernelTelemetryConfig telemetry_config = {
+      .has_enabled = has_enabled,
+      .enabled = enabled,
+      .batch_size = batch_size,
+      .flush_interval_ms = flush_interval_ms,
+      .max_retries = max_retries,
+      .retry_delay_ms = retry_delay_ms,
+      .close_flush_timeout_ms = close_flush_timeout_ms,
+  };
+  return kernel_session_config_set_telemetry_config_v2(config, &telemetry_config);
 }
 
 static inline KernelStatusCode go_kernel_set_driver_system_configuration(
@@ -485,9 +492,14 @@ func (k *KernelBackend) applyTelemetry(cfg *C.KernelSessionConfig) error {
 	if closeFlushTimeout <= 0 {
 		closeFlushTimeout = defaultTelemetryCloseFlushTimeout
 	}
+	hasEnabled := t.Enabled != nil
+	enabled := false
+	if hasEnabled {
+		enabled = *t.Enabled
+	}
 	if err := call(func() C.KernelStatusCode {
 		return C.go_kernel_set_telemetry_config(cfg,
-			C.bool(t.Enabled), C.size_t(batchSize), C.uint64_t(flushInterval.Milliseconds()),
+			C.bool(hasEnabled), C.bool(enabled), C.size_t(batchSize), C.uint64_t(flushInterval.Milliseconds()),
 			C.uint32_t(maxRetries), C.uint64_t(retryDelay.Milliseconds()),
 			C.uint64_t(closeFlushTimeout.Milliseconds()))
 	}); err != nil {

internal/backend/kernel/config.go

@@ -191,11 +191,12 @@ func requestTimeoutMilliseconds(timeout time.Duration) int64 {
 }
 
 // TelemetryConfig is the kernel telemetry subset exposed by the Go driver. It
-// mirrors the kernel telemetry C ABI: Enabled follows the user-supplied
-// enableTelemetry value, defaulting to false when unset. Zero-valued tuning fields
-// keep the kernel defaults until applyTelemetry fills them for the setter.
+// mirrors the kernel telemetry C ABI: Enabled is nil when enableTelemetry is
+// unset so the kernel feature flag decides, and non-nil for an explicit client
+// override. Zero-valued tuning fields keep the kernel defaults until
+// applyTelemetry fills them for the setter.
 type TelemetryConfig struct {
-	Enabled           bool
+	Enabled           *bool
 	BatchSize         int
 	FlushInterval     time.Duration
 	MaxRetries        uint32

internal/backend/kernel/include/databricks_kernel.h

@@ -487,6 +487,25 @@ KernelStatusCode kernel_session_config_set_telemetry_config(KernelSessionConfig*
                                                             uint64_t retry_delay_ms,
                                                             uint64_t close_flush_timeout_ms);
 
+/* Optional telemetry configuration for hosts that must distinguish an omitted
+ * enablement setting from an explicit false. Passing `telemetry_config == NULL`
+ * leaves the kernel telemetry configuration unchanged. Otherwise the tuning
+ * fields are applied and `has_enabled` controls whether `enabled` overrides the
+ * workspace feature flag. The original positional setter above remains ABI
+ * compatible and is equivalent to this function with `has_enabled = true`. */
+typedef struct KernelTelemetryConfig {
+  bool has_enabled;
+  bool enabled;
+  size_t batch_size;
+  uint64_t flush_interval_ms;
+  uint32_t max_retries;
+  uint64_t retry_delay_ms;
+  uint64_t close_flush_timeout_ms;
+} KernelTelemetryConfig;
+
+KernelStatusCode kernel_session_config_set_telemetry_config_v2(
+    KernelSessionConfig* config, const KernelTelemetryConfig* telemetry_config);
+
 /* Configure telemetry circuit-breaker behavior. `threshold` and `timeout_ms`
  * must be > 0 when circuit breaking is enabled. If this setter is not called,
  * the kernel keeps its default circuit-breaker policy. */

internal/backend/kernel/kernel_test.go

@@ -290,25 +290,28 @@ func TestSetMaxConnections(t *testing.T) {
 }
 
 func TestSetTelemetry(t *testing.T) {
+	enabled := true
+	disabled := false
 	for _, tc := range []struct {
 		name    string
 		cfg     Config
 		wantErr bool
 	}{
 		{"nil telemetry is no-op", Config{}, false},
-		{"defaults filled for zero tuning fields", Config{Telemetry: &TelemetryConfig{Enabled: true}}, false},
+		{"unspecified enablement preserves feature flag", Config{Telemetry: &TelemetryConfig{}}, false},
+		{"defaults filled for zero tuning fields", Config{Telemetry: &TelemetryConfig{Enabled: &enabled}}, false},
 		{"explicit tuning", Config{Telemetry: &TelemetryConfig{
-			Enabled:           false,
+			Enabled:           &disabled,
 			BatchSize:         17,
 			FlushInterval:     2 * time.Second,
 			MaxRetries:        2,
 			RetryDelay:        25 * time.Millisecond,
 			CloseFlushTimeout: 3 * time.Second,
 		}}, false},
-		{"negative batch size is defaulted", Config{Telemetry: &TelemetryConfig{Enabled: true, BatchSize: -1}}, false},
-		{"negative flush interval is defaulted", Config{Telemetry: &TelemetryConfig{Enabled: true, FlushInterval: -time.Second}}, false},
-		{"negative retry delay is defaulted", Config{Telemetry: &TelemetryConfig{Enabled: true, RetryDelay: -time.Second}}, false},
-		{"negative close timeout is defaulted", Config{Telemetry: &TelemetryConfig{Enabled: true, CloseFlushTimeout: -time.Second}}, false},
+		{"negative batch size is defaulted", Config{Telemetry: &TelemetryConfig{Enabled: &enabled, BatchSize: -1}}, false},
+		{"negative flush interval is defaulted", Config{Telemetry: &TelemetryConfig{Enabled: &enabled, FlushInterval: -time.Second}}, false},
+		{"negative retry delay is defaulted", Config{Telemetry: &TelemetryConfig{Enabled: &enabled, RetryDelay: -time.Second}}, false},
+		{"negative close timeout is defaulted", Config{Telemetry: &TelemetryConfig{Enabled: &enabled, CloseFlushTimeout: -time.Second}}, false},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			err := trySetTelemetry(tc.cfg)

kernel_config.go

@@ -219,9 +219,9 @@ func buildKernelConfig(cfg *config.Config, kauth kernel.Auth) kernel.Config {
 }
 
 func kernelTelemetryConfig(cfg *config.Config) *kernel.TelemetryConfig {
-	enabled := false
+	var enabled *bool
 	if val, isSet := cfg.EnableTelemetry.Get(); isSet {
-		enabled = val
+		enabled = &val
 	}
 	return &kernel.TelemetryConfig{
 		Enabled:       enabled,

kernel_config_test.go

@@ -647,8 +647,8 @@ func TestBuildKernelConfig(t *testing.T) {
 		if kc.Telemetry == nil {
 			t.Fatal("Telemetry not forwarded")
 		}
-		if kc.Telemetry.Enabled {
-			t.Error("Telemetry.Enabled = true, want false when enableTelemetry is unset")
+		if kc.Telemetry.Enabled != nil {
+			t.Errorf("Telemetry.Enabled = %v, want nil when enableTelemetry is unset", *kc.Telemetry.Enabled)
 		}
 		if kc.Telemetry.BatchSize != 0 {
 			t.Errorf("Telemetry.BatchSize = %d, want 0 when telemetry_batch_size is unset", kc.Telemetry.BatchSize)
@@ -664,8 +664,8 @@ func TestBuildKernelConfig(t *testing.T) {
 		if kc.Telemetry == nil {
 			t.Fatal("Telemetry not forwarded")
 		}
-		if kc.Telemetry.Enabled {
-			t.Error("Telemetry.Enabled = true, want false from explicit enableTelemetry=false")
+		if kc.Telemetry.Enabled == nil || *kc.Telemetry.Enabled {
+			t.Errorf("Telemetry.Enabled = %v, want explicit false", kc.Telemetry.Enabled)
 		}
 		if kc.Telemetry.BatchSize != 17 {
 			t.Errorf("Telemetry.BatchSize = %d, want 17", kc.Telemetry.BatchSize)
@@ -676,8 +676,8 @@ func TestBuildKernelConfig(t *testing.T) {
 
 		c.EnableTelemetry = config.NewConfigValue(true)
 		kc = buildKernelConfig(c, kernel.Auth{Mode: kernel.AuthPAT, Token: "dapi-x"})
-		if !kc.Telemetry.Enabled {
-			t.Error("Telemetry.Enabled = false, want true from explicit enableTelemetry=true")
+		if kc.Telemetry.Enabled == nil || !*kc.Telemetry.Enabled {
+			t.Errorf("Telemetry.Enabled = %v, want explicit true", kc.Telemetry.Enabled)
 		}
 	})