Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions lib/DBSQLClient.ts
Original file line number Diff line number Diff line change
Expand Up @@ -729,7 +729,8 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
// doesn't ship in the public `.d.ts`. Mirrors Python's `kwargs.get("use_kernel")`
// pattern (see databricks-sql-python/src/databricks/sql/session.py).
const internalOptions = options as ConnectionOptions & InternalConnectionOptions;
const backend = internalOptions.useKernel
const useKernel = internalOptions.useKernel === true;
const backend = useKernel
? new KernelBackend({ context: this })
: new ThriftBackend({
context: this,
Expand Down Expand Up @@ -777,7 +778,7 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
`Telemetry remains controlled by the runtime config and feature flag.`,
);
}
if (this.config.telemetryEnabled && !envDisabled) {
if (!useKernel && this.config.telemetryEnabled && !envDisabled) {
await this.initializeTelemetry();
}

Expand Down
117 changes: 117 additions & 0 deletions lib/kernel/KernelAuth.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,16 @@
// See the License for the specific language governing permissions and
// limitations under the License.

import os from 'os';
import { ConnectionOptions } from '../contracts/IDBSQLClient';
import { ClientConfig } from '../contracts/IClientContext';
import { InternalConnectionOptions } from '../contracts/InternalConnectionOptions';
import AuthenticationError from '../errors/AuthenticationError';
import HiveDriverError from '../errors/HiveDriverError';
import { buildUserAgentString, normalizePemBytes } from '../utils';
import driverVersion from '../version';
import { DRIVER_NAME } from '../telemetry/types';
import { sanitizeProcessName } from '../telemetry/telemetryUtils';

/**
* Default local listener port for the U2M authorization-code callback.
Expand Down Expand Up @@ -131,6 +136,32 @@ export interface KernelSessionDefaults {
retryOverallTimeoutSecs?: number;
}

export interface KernelTelemetryOptions {
/** Driver/runtime identity forwarded to kernel-owned telemetry. */
driverName?: string;
driverVersion?: string;
runtimeName?: string;
runtimeVersion?: string;
runtimeVendor?: string;
osName?: string;
osVersion?: string;
osArch?: string;
clientAppName?: string;
localeName?: string;
charSetEncoding?: string;
processName?: string;
/** Kernel-owned telemetry switch and batching. */
telemetryEnabled?: boolean;
telemetryBatchSize?: number;
telemetryFlushIntervalMs?: number;
telemetryMaxRetries?: number;
telemetryRetryDelayMs?: number;
telemetryCloseFlushTimeoutMs?: number;
telemetryCircuitBreakerEnabled?: boolean;
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
telemetryCircuitBreakerThreshold?: number;
telemetryCircuitBreakerTimeoutMs?: number;
}

/**
* TLS options shared across all auth-mode variants. Mirror the napi
* binding's `ConnectionOptions.checkServerCertificate` / `.customCaCert`
Expand Down Expand Up @@ -227,6 +258,7 @@ export interface KernelFederationOptions {
export type KernelNativeConnectionOptions = KernelSessionDefaults &
KernelTlsOptions &
KernelHttpOptions &
KernelTelemetryOptions &
KernelProxyOptions &
KernelFederationOptions &
(
Expand Down Expand Up @@ -588,6 +620,91 @@ export function buildKernelRetryOptions(config: {
return out;
}

function getLocaleName(env: NodeJS.ProcessEnv = process.env): string {
try {
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
const lang = env.LANG || env.LC_ALL || env.LC_MESSAGES || '';
const match = lang.match(/^([a-z]{2}_[A-Z]{2})/);
return match?.[1] ?? 'en_US';
} catch {
return 'en_US';
}
}

function getProcessName(): string {
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
try {
if (process.title && process.title !== 'node') {
return sanitizeProcessName(process.title) || 'node';
}
const scriptPath = process.argv?.[1];
if (scriptPath) {
return sanitizeProcessName(scriptPath).replace(/\.[^.]*$/, '') || 'node';
}
return 'node';
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
} catch {
return 'node';
}
}

export function isTelemetryDisabledByEnv(env: NodeJS.ProcessEnv = process.env): boolean {
const raw = env.DATABRICKS_TELEMETRY_DISABLED;
const trimmed = typeof raw === 'string' ? raw.trim() : '';
return trimmed.length > 0 && /^(1|true|yes|on)$/i.test(trimmed);
}

export function buildKernelTelemetryOptions(
config: Pick<
ClientConfig,
| 'telemetryEnabled'
| 'telemetryBatchSize'
| 'telemetryFlushIntervalMs'
| 'telemetryMaxRetries'
| 'telemetryBackoffBaseMs'
| 'telemetryCloseTimeoutMs'
| 'telemetryCircuitBreakerThreshold'
| 'telemetryCircuitBreakerTimeout'
>,
) {
const telemetry: KernelTelemetryOptions = {
driverName: DRIVER_NAME,
driverVersion,
runtimeName: 'Node.js',
runtimeVersion: process.version,
runtimeVendor: 'Node.js Foundation',
osName: process.platform,
osVersion: os.release(),
osArch: os.arch(),
clientAppName: undefined,
localeName: getLocaleName(),
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
charSetEncoding: 'UTF-8',
processName: getProcessName(),
telemetryEnabled: (config.telemetryEnabled ?? true) && !isTelemetryDisabledByEnv(),
};

if (Number.isFinite(config.telemetryBatchSize)) {
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
Outdated
telemetry.telemetryBatchSize = config.telemetryBatchSize;
}
if (Number.isFinite(config.telemetryFlushIntervalMs)) {
telemetry.telemetryFlushIntervalMs = config.telemetryFlushIntervalMs;
}
if (Number.isFinite(config.telemetryMaxRetries)) {
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
Outdated
telemetry.telemetryMaxRetries = config.telemetryMaxRetries;

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Medium — The Number.isFinite(...) guards here are meant to omit a knob so the kernel keeps its own default — native/kernel/index.d.ts documents each as "Omitted ⇒ kernel default." But when this runs through the normal DBSQLClient flow, getDefaultConfig() (lib/DBSQLClient.ts:170-185) always populates every one of these telemetry fields with a finite value from DEFAULT_TELEMETRY_CONFIG. So every guard always passes and the kernel always receives the Node connector's telemetry defaults (batchSize 100, flushIntervalMs 5000, maxRetries 3, backoffBaseMs 100, closeTimeoutMs 2000, circuitBreakerThreshold 5, circuitBreakerTimeout 60000) — its own tuned defaults are never used.

Those defaults were chosen for the JS HTTP exporter's batching/backoff, not the kernel's Rust telemetry pipeline. If that override is intended, the "Omitted ⇒ kernel default" wording and the isFinite guards are misleading (the omit path is only reachable from a hand-built config, e.g. the unit tests). If it isn't intended, the connector is silently overriding the kernel's telemetry tuning. Worth confirming which behavior you want and aligning the guards/docs accordingly.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

NEEDS HUMAN DECISION — the bots can't resolve this thread; a maintainer's input is required.

Confirmed the reviewer is correct: DBSQLClient.getDefaultConfig() (lib/DBSQLClient.ts:170-185) always populates every telemetry field from DEFAULT_TELEMETRY_CONFIG with finite values, so the Number.isFinite guards in buildKernelTelemetryOptions always pass in the normal client flow and the kernel never uses its own tuned defaults (the omit path is only reachable from hand-built configs like the unit tests). Resolving this is a design decision with two opposite outcomes — either (a) the override is intended and the "Omitted ⇒ kernel default" docs + isFinite guards should be reworded/removed, or (b) it's unintended and getDefaultConfig should stop populating these so the kernel keeps its Rust-pipeline tuning (a behavioral change to a widely-consumed connector). Which telemetry defaults should win (JS connector vs Rust kernel) is a product/design judgment I can't make from the code and can't verify here; needs a human to decide intent before either the docs/guards or getDefaultConfig are changed.

}
if (Number.isFinite(config.telemetryBackoffBaseMs)) {
telemetry.telemetryRetryDelayMs = config.telemetryBackoffBaseMs;
}
if (Number.isFinite(config.telemetryCloseTimeoutMs)) {
telemetry.telemetryCloseFlushTimeoutMs = config.telemetryCloseTimeoutMs;
}
if (Number.isFinite(config.telemetryCircuitBreakerThreshold)) {
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
Outdated
telemetry.telemetryCircuitBreakerThreshold = config.telemetryCircuitBreakerThreshold;
}
if (Number.isFinite(config.telemetryCircuitBreakerTimeout)) {
telemetry.telemetryCircuitBreakerTimeoutMs = config.telemetryCircuitBreakerTimeout;
}

return telemetry;
}

/**
* Map the public `ConnectionOptions.proxy` (`{protocol, host, port, auth}` —
* the same shape the Thrift backend accepts) onto the kernel's structured napi
Expand Down
8 changes: 7 additions & 1 deletion lib/kernel/KernelBackend.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,12 @@ import HiveDriverError from '../errors/HiveDriverError';
import { serializeQueryTags } from '../utils';
import { getKernelNative, KernelNativeBinding, KernelConnection } from './KernelNativeLoader';
import { decodeNapiKernelError } from './KernelErrorMapping';
import { buildKernelConnectionOptions, buildKernelRetryOptions, KernelNativeConnectionOptions } from './KernelAuth';
import {
buildKernelConnectionOptions,
buildKernelRetryOptions,
buildKernelTelemetryOptions,
KernelNativeConnectionOptions,
} from './KernelAuth';
import { installKernelLogBridge } from './KernelLogging';
import KernelSessionBackend from './KernelSessionBackend';

Expand Down Expand Up @@ -93,6 +98,7 @@ export default class KernelBackend implements IBackend {
this.nativeOptions = {
...buildKernelConnectionOptions(options),
...buildKernelRetryOptions(this.context.getConfig()),
...buildKernelTelemetryOptions(this.context.getConfig()),
};

// Bridge the Rust kernel's `tracing` logs into the SAME `DBSQLLogger` the
Expand Down
14 changes: 14 additions & 0 deletions native/kernel/index.d.ts

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

17 changes: 17 additions & 0 deletions tests/unit/DBSQLClient.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import fs from 'fs';
import DBSQLClient, { ThriftLibrary } from '../../lib/DBSQLClient';
import DBSQLSession from '../../lib/DBSQLSession';
import ThriftBackend from '../../lib/thrift-backend/ThriftBackend';
import KernelBackend from '../../lib/kernel/KernelBackend';

import PlainHttpAuthentication from '../../lib/connection/auth/PlainHttpAuthentication';
import DatabricksOAuth from '../../lib/connection/auth/DatabricksOAuth';
Expand Down Expand Up @@ -957,6 +958,22 @@ describe('DBSQLClient telemetry paths', () => {
.filter((c) => c.args[0] === LogLevel.warn && /DATABRICKS_TELEMETRY_DISABLED/.test(c.args[1] as string));
expect(warnCalls.length).to.equal(0);
});

it('does not initialize Node telemetry on the kernel path', async () => {
delete process.env.DATABRICKS_TELEMETRY_DISABLED;
const client = new DBSQLClient();
const initStub = sinon.stub(client as any, 'initializeTelemetry').resolves();
sinon.stub(KernelBackend.prototype, 'connect').resolves();
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
sinon.stub(KernelBackend.prototype, 'close').resolves();

try {
await client.connect({ ...connectOptions, telemetryEnabled: true, useKernel: true } as any);

expect(initStub.callCount).to.equal(0);
} finally {
await client.close();
}
});
});

describe('extractWorkspaceId', () => {
Expand Down
25 changes: 25 additions & 0 deletions tests/unit/kernel/_helpers/nativeOptions.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,31 @@ export default function expectNativeConnectionOptions(actual: unknown, expectedR
const { customHeaders, ...rest } = actual as Record<string, unknown> & {
customHeaders?: Array<{ name: string; value: string }>;
};
for (const key of [
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
'driverName',
'driverVersion',
'runtimeName',
'runtimeVersion',
'runtimeVendor',
'osName',
'osVersion',
'osArch',
'clientAppName',
'localeName',
'charSetEncoding',
'processName',
'telemetryEnabled',
'telemetryBatchSize',
'telemetryFlushIntervalMs',
'telemetryMaxRetries',
'telemetryRetryDelayMs',
'telemetryCloseFlushTimeoutMs',
'telemetryCircuitBreakerEnabled',
'telemetryCircuitBreakerThreshold',
'telemetryCircuitBreakerTimeoutMs',
]) {
delete rest[key];
}
expect(rest).to.deep.equal(expectedRest);
expect(customHeaders, 'customHeaders').to.be.an('array').with.lengthOf(1);
expect(customHeaders?.[0].name).to.equal('User-Agent');
Expand Down
93 changes: 91 additions & 2 deletions tests/unit/kernel/execution.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -418,13 +418,13 @@ function makeBinding(connection: KernelConnection): KernelNativeBinding & {
return Object.assign(binding, { openSessionStub });
}

function makeContext(logger?: IDBSQLLogger): IClientContext {
function makeContext(logger?: IDBSQLLogger, configOverrides: Partial<ClientConfig> = {}): IClientContext {
const log: IDBSQLLogger = logger ?? {
log(_level: LogLevel, _message: string): void {
// no-op
},
};
const config = {} as ClientConfig;
const config = configOverrides as ClientConfig;
return {
getConfig: () => config,
getLogger: () => log,
Expand Down Expand Up @@ -551,6 +551,95 @@ describe('KernelBackend', () => {
});
});

it('openSession() forwards kernel-owned telemetry config and runtime identity to napi binding', async () => {
const savedEnv = process.env.DATABRICKS_TELEMETRY_DISABLED;
delete process.env.DATABRICKS_TELEMETRY_DISABLED;

const connection = new FakeNativeConnection();
const binding = makeBinding(connection);
const backend = new KernelBackend({
context: makeContext(undefined, {
telemetryEnabled: false,
telemetryBatchSize: 17,
telemetryFlushIntervalMs: 1_000,
telemetryMaxRetries: 2,
telemetryBackoffBaseMs: 50,
telemetryCloseTimeoutMs: 2_500,
telemetryCircuitBreakerThreshold: 3,
telemetryCircuitBreakerTimeout: 60_000,
}),
nativeBinding: binding,
});

try {
await backend.connect({
host: 'workspace.example',
path: '/sql/1.0/warehouses/xyz',
token: 'dapi-token',
} as ConnectionOptions);

await backend.openSession({});

const args = binding.openSessionStub.firstCall.args[0] as Record<string, unknown>;
expect(args.driverName).to.equal('nodejs-sql-driver');
expect(args.driverVersion).to.be.a('string').and.not.equal('');
expect(args.runtimeName).to.equal('Node.js');
expect(args.runtimeVersion).to.equal(process.version);
expect(args.runtimeVendor).to.equal('Node.js Foundation');
expect(args.osName).to.equal(process.platform);
expect(args.osVersion).to.be.a('string').and.not.equal('');
expect(args.osArch).to.be.a('string').and.not.equal('');
expect(args.localeName).to.be.a('string').and.not.equal('');
expect(args.charSetEncoding).to.equal('UTF-8');
expect(args.processName).to.be.a('string').and.not.equal('');
expect(args.telemetryEnabled).to.equal(false);
expect(args.telemetryBatchSize).to.equal(17);
expect(args.telemetryFlushIntervalMs).to.equal(1_000);
expect(args.telemetryMaxRetries).to.equal(2);
expect(args.telemetryRetryDelayMs).to.equal(50);
expect(args.telemetryCloseFlushTimeoutMs).to.equal(2_500);
expect(args.telemetryCircuitBreakerEnabled).to.equal(undefined);
expect(args.telemetryCircuitBreakerThreshold).to.equal(3);
expect(args.telemetryCircuitBreakerTimeoutMs).to.equal(60_000);
} finally {
if (savedEnv === undefined) {
delete process.env.DATABRICKS_TELEMETRY_DISABLED;
} else {
process.env.DATABRICKS_TELEMETRY_DISABLED = savedEnv;
}
}
});

it('openSession() forwards env-disabled kernel telemetry even when config enables telemetry', async () => {
const savedEnv = process.env.DATABRICKS_TELEMETRY_DISABLED;
process.env.DATABRICKS_TELEMETRY_DISABLED = 'true';

const connection = new FakeNativeConnection();
const binding = makeBinding(connection);
const backend = new KernelBackend({
context: makeContext(undefined, { telemetryEnabled: true }),
nativeBinding: binding,
});

try {
await backend.connect({
host: 'workspace.example',
path: '/sql/1.0/warehouses/xyz',
token: 'dapi-token',
} as ConnectionOptions);
await backend.openSession({});

const args = binding.openSessionStub.firstCall.args[0] as { telemetryEnabled?: boolean };
expect(args.telemetryEnabled).to.equal(false);
} finally {
if (savedEnv === undefined) {
delete process.env.DATABRICKS_TELEMETRY_DISABLED;
} else {
process.env.DATABRICKS_TELEMETRY_DISABLED = savedEnv;
}
}
});

it('openSession() serializes session-level queryTags into sessionConf.QUERY_TAGS', async () => {
const connection = new FakeNativeConnection();
const binding = makeBinding(connection);
Expand Down
Loading