diff --git a/.eslintrc b/.eslintrc index c027bbd0..ba0e8a85 100644 --- a/.eslintrc +++ b/.eslintrc @@ -29,7 +29,8 @@ "files": ["*.test.js", "*.test.ts"], "rules": { "no-unused-expressions": "off", - "@typescript-eslint/no-unused-expressions": "off" + "@typescript-eslint/no-unused-expressions": "off", + "func-names": "off" } } ] diff --git a/CHANGELOG.md b/CHANGELOG.md index 849a12a1..78b97286 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,13 @@ - Add metric view metadata support (databricks/databricks-sql-nodejs#312 by @shivam2680) - Fix: Avoid calling require('lz4') if it's really not required (databricks/databricks-sql-nodejs#316 by @ikkala) - Add telemetry foundation (off by default) (databricks/databricks-sql-nodejs#324 by @samikshya-db) +- Telemetry event emission and per-host aggregation (databricks/databricks-sql-nodejs#327 by @samikshya-db). + **Default change:** `telemetryEnabled` now defaults to `true` (gated by a remote feature flag). + To opt out programmatically, pass `telemetryEnabled: false` to `connect()`. + To disable globally without code changes, set the environment variable + `DATABRICKS_TELEMETRY_DISABLED` to one of `1`, `true`, `yes`, or `on` + (case-insensitive). Other values (empty, `0`, `false`, etc.) are ignored + — the runtime config takes precedence. ## 1.12.0 diff --git a/lib/DBSQLClient.ts b/lib/DBSQLClient.ts index ee53e790..5155f96d 100644 --- a/lib/DBSQLClient.ts +++ b/lib/DBSQLClient.ts @@ -1,5 +1,6 @@ import thrift from 'thrift'; import Int64 from 'node-int64'; +import os from 'os'; import { EventEmitter } from 'events'; import TCLIService from '../thrift/TCLIService'; @@ -31,6 +32,12 @@ import IDBSQLLogger, { LogLevel } from './contracts/IDBSQLLogger'; import DBSQLLogger from './DBSQLLogger'; import CloseableCollection from './utils/CloseableCollection'; import IConnectionProvider from './connection/contracts/IConnectionProvider'; +import TelemetryClient from './telemetry/TelemetryClient'; +import TelemetryClientProvider from './telemetry/TelemetryClientProvider'; +import TelemetryEventEmitter from './telemetry/TelemetryEventEmitter'; +import { DriverConfiguration, DRIVER_NAME, TelemetryEventType } from './telemetry/types'; +import { safeEmit } from './telemetry/telemetryUtils'; +import driverVersion from './version'; function prependSlash(str: string): string { if (str.length > 0 && str.charAt(0) !== '/') { @@ -75,6 +82,21 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I private readonly sessions = new CloseableCollection(); + // Telemetry components — `telemetryClient` is the shared per-host owner + // (process-wide via TelemetryClientProvider). The exporter, aggregator, + // circuit-breaker registry and feature-flag cache live on it. Each + // DBSQLClient still owns its own `telemetryEmitter` so it respects its + // own `telemetryEnabled` flag. + private host?: string; + + private httpPath?: string; + + private authType?: string; + + private telemetryClient?: TelemetryClient; + + private telemetryEmitter?: TelemetryEventEmitter; + private static getDefaultLogger(): IDBSQLLogger { if (!this.defaultLogger) { this.defaultLogger = new DBSQLLogger(); @@ -101,6 +123,15 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I cloudFetchSpeedThresholdMBps: 0.1, useLZ4Compression: true, + + // Telemetry defaults + telemetryEnabled: true, // Enabled by default, gated by feature flag + telemetryBatchSize: 100, + telemetryFlushIntervalMs: 5000, + telemetryMaxRetries: 3, + telemetryAuthenticatedExport: true, + telemetryCircuitBreakerThreshold: 5, + telemetryCircuitBreakerTimeout: 60000, // 1 minute }; } @@ -212,6 +243,172 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I return new HttpConnection(this.getConnectionOptions(options), this); } + /** + * Extract workspace ID from hostname. + * @param host - The host string (e.g., "workspace-id.cloud.databricks.com") + * @returns Workspace ID or host if extraction fails + */ + private extractWorkspaceId(host: string): string { + // Extract workspace ID from hostname (first segment before first dot) + const parts = host.split('.'); + return parts.length > 0 ? parts[0] : host; + } + + /** + * Build driver configuration for telemetry reporting. + * @returns DriverConfiguration object with current driver settings + */ + private buildDriverConfiguration(): DriverConfiguration { + return { + driverVersion, + driverName: DRIVER_NAME, + nodeVersion: process.version, + platform: process.platform, + osVersion: os.release(), + osArch: os.arch(), + runtimeVendor: 'Node.js Foundation', + localeName: this.getLocaleName(), + charSetEncoding: 'UTF-8', + processName: this.getProcessName(), + authType: this.authType || 'pat', + + // Feature flags + cloudFetchEnabled: this.config.useCloudFetch ?? false, + lz4Enabled: this.config.useLZ4Compression ?? false, + arrowEnabled: this.config.arrowEnabled ?? false, + directResultsEnabled: true, // Direct results always enabled + + // Configuration values + socketTimeout: this.config.socketTimeout ?? 0, + retryMaxAttempts: this.config.retryMaxAttempts ?? 0, + cloudFetchConcurrentDownloads: this.config.cloudFetchConcurrentDownloads ?? 0, + + // Connection parameters + httpPath: this.httpPath, + enableMetricViewMetadata: this.config.enableMetricViewMetadata, + }; + } + + /** + * Map Node.js auth type to telemetry auth enum string. + * Distinguishes between U2M and M2M OAuth flows. + */ + private mapAuthType(options: ConnectionOptions): string { + switch (options.authType) { + case 'databricks-oauth': + return options.oauthClientSecret === undefined ? 'external-browser' : 'oauth-m2m'; + case 'custom': + return 'custom'; + case 'token-provider': + return 'token-provider'; + case 'external-token': + return 'external-token'; + case 'static-token': + return 'static-token'; + case 'access-token': + case undefined: + return 'pat'; + default: + return 'unknown'; + } + } + + /** + * Get locale name in format language_country (e.g., en_US). + * Matches JDBC format: user.language + '_' + user.country + */ + private getLocaleName(): string { + try { + // Try to get from environment variables + const lang = process.env.LANG || process.env.LC_ALL || process.env.LC_MESSAGES || ''; + if (lang) { + // LANG format is typically "en_US.UTF-8", extract "en_US" + const match = lang.match(/^([a-z]{2}_[A-Z]{2})/); + if (match) { + return match[1]; + } + } + // Fallback to en_US + return 'en_US'; + } catch { + return 'en_US'; + } + } + + /** + * Get process name, similar to JDBC's ProcessNameUtil. + * Returns the script name or process title. + */ + private getProcessName(): string { + try { + // Try process.title first (can be set by application) + if (process.title && process.title !== 'node') { + return process.title; + } + // Try to get the main script name from argv[1] + if (process.argv && process.argv.length > 1) { + const scriptPath = process.argv[1]; + // Extract filename without path + const filename = scriptPath.split('/').pop()?.split('\\').pop() || ''; + // Remove extension + const nameWithoutExt = filename.replace(/\.[^.]*$/, ''); + if (nameWithoutExt) { + return nameWithoutExt; + } + } + return 'node'; + } catch { + return 'node'; + } + } + + /** + * Initialize telemetry components if enabled. + * CRITICAL: All errors swallowed and logged at LogLevel.debug ONLY. + * Driver NEVER throws exceptions due to telemetry. + */ + private async initializeTelemetry(): Promise { + if (!this.host) { + return; + } + + try { + // Acquire (or create) the per-host TelemetryClient from the + // process-wide provider. The shared client owns the circuit-breaker + // registry, feature-flag cache, exporter, and aggregator. Multiple + // DBSQLClient instances on the same host share these resources so + // breaker counters and HTTP batches don't fragment per-instance. + this.telemetryClient = TelemetryClientProvider.getInstance().getOrCreateClient(this, this.host); + + // Use the shared feature-flag cache (registered in the previous step). + const enabled = await this.telemetryClient.getFeatureFlagCache().isTelemetryEnabled(this.host); + + if (!enabled) { + // Release our refcount immediately; we won't be emitting. + await TelemetryClientProvider.getInstance().releaseClient(this, this.host); + this.telemetryClient = undefined; + this.logger.log(LogLevel.debug, 'Telemetry: disabled'); + return; + } + + // Each DBSQLClient still owns its own emitter so it respects its own + // `telemetryEnabled` flag and feature-flag result. All emitters bridge + // into the SHARED aggregator on the TelemetryClient. + this.telemetryEmitter = new TelemetryEventEmitter(this); + const sharedAggregator = this.telemetryClient.getAggregator(); + for (const eventType of Object.values(TelemetryEventType)) { + this.telemetryEmitter.on(eventType, (event) => { + sharedAggregator.processEvent(event); + }); + } + + this.logger.log(LogLevel.debug, 'Telemetry: enabled'); + } catch (error: any) { + // Swallow all telemetry initialization errors + this.logger.log(LogLevel.debug, `Telemetry initialization error: ${error.message}`); + } + } + /** * Connects DBSQLClient to endpoint * @public @@ -233,11 +430,48 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I } } + // If connect() is being called a second time (reconnect, host switch), + // release the prior telemetry refcount and emitter so we don't leak a + // refcount in the process-wide TelemetryClientProvider for the old host. + if (this.host && this.telemetryClient) { + try { + await TelemetryClientProvider.getInstance().releaseClient(this, this.host); + } catch (error: any) { + this.logger.log(LogLevel.debug, `Telemetry release-on-reconnect error: ${error.message}`); + } + this.telemetryClient = undefined; + this.telemetryEmitter = undefined; + } + + // Store connection params for telemetry + this.host = options.host; + this.httpPath = options.path; + this.authType = this.mapAuthType(options); + // Store enableMetricViewMetadata configuration if (options.enableMetricViewMetadata !== undefined) { this.config.enableMetricViewMetadata = options.enableMetricViewMetadata; } + // Override telemetry config if provided in options + const telemetryOverrides = [ + 'telemetryEnabled', + 'telemetryBatchSize', + 'telemetryFlushIntervalMs', + 'telemetryMaxRetries', + 'telemetryAuthenticatedExport', + 'telemetryCircuitBreakerThreshold', + 'telemetryCircuitBreakerTimeout', + 'telemetryCloseTimeoutMs', + 'telemetryMaxStatementMetrics', + ] as const; + for (const k of telemetryOverrides) { + if (options[k] !== undefined) { + // The narrow union forces a cast; values are validated at point of use. + (this.config as any)[k] = options[k]; + } + } + // Persist userAgentEntry so telemetry and feature-flag call sites reuse // the same value as the primary Thrift connection's User-Agent. if (options.userAgentEntry !== undefined) { @@ -277,6 +511,18 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I this.emit('timeout'); }); + // Initialize telemetry if enabled. The env var DATABRICKS_TELEMETRY_DISABLED + // is a hard kill switch for ops/IT teams who can't redeploy app code. + // Recognized truthy values: 1, true, yes, on (case-insensitive). Anything + // else (empty, "0", "false", "no", "off") leaves the runtime config in + // charge — avoiding the footgun where a sysadmin sets the var to "false" + // expecting to enable telemetry. + const envKill = process.env.DATABRICKS_TELEMETRY_DISABLED; + const envDisabled = typeof envKill === 'string' && /^(1|true|yes|on)$/i.test(envKill.trim()); + if (this.config.telemetryEnabled && !envDisabled) { + await this.initializeTelemetry(); + } + return this; } @@ -290,6 +536,9 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I * const session = await client.openSession(); */ public async openSession(request: OpenSessionRequest = {}): Promise { + // Track connection open latency + const startTime = Date.now(); + // Prepare session configuration const configuration = request.configuration ? { ...request.configuration } : {}; @@ -312,12 +561,52 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I serverProtocolVersion: response.serverProtocolVersion, }); this.sessions.add(session); + + // Emit connection.open telemetry event + safeEmit(this, (emitter) => { + if (!this.host) return; + const latencyMs = Date.now() - startTime; + const workspaceId = this.extractWorkspaceId(this.host); + const driverConfig = this.buildDriverConfiguration(); + emitter.emitConnectionOpen({ + sessionId: session.id, + workspaceId, + driverConfig, + latencyMs, + }); + }); + return session; } + /** + * Closes the client, releasing sessions and telemetry resources. + * + * The internal telemetry flush timer uses `setInterval(...).unref()` so it + * cannot keep the Node.js process alive on its own. As a consequence, any + * telemetry buffered between flush ticks is lost if the process exits + * without calling `close()`. Long-lived applications should `await` this + * method on shutdown so the aggregator drains its remaining metrics. + */ public async close(): Promise { await this.sessions.closeAll(); + // Cleanup telemetry. Releasing our refcount on the shared TelemetryClient + // is awaited because the underlying close() drains the final HTTP POST — + // a caller doing `await client.close(); process.exit(0)` would otherwise + // truncate the in-flight request when this is the last refcount holder. + if (this.host && this.telemetryClient) { + try { + await TelemetryClientProvider.getInstance().releaseClient(this, this.host); + } catch (error: any) { + this.logger.log(LogLevel.debug, `Telemetry cleanup error: ${error.message}`); + } + this.telemetryClient = undefined; + } + // Drop the emitter ref so post-close calls (e.g. session.close racing + // with client.close) cannot smuggle events into the closed aggregator. + this.telemetryEmitter = undefined; + this.client = undefined; this.connectionProvider = undefined; this.authProvider = undefined; @@ -369,4 +658,14 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I public getAuthProvider(): IAuthentication | undefined { return this.authProvider; } + + /** @internal */ + public getTelemetryEmitter(): TelemetryEventEmitter | undefined { + return this.telemetryEmitter; + } + + /** @internal */ + public getTelemetryAggregator() { + return this.telemetryClient?.getAggregator(); + } } diff --git a/lib/DBSQLOperation.ts b/lib/DBSQLOperation.ts index fe22995d..af27195a 100644 --- a/lib/DBSQLOperation.ts +++ b/lib/DBSQLOperation.ts @@ -34,11 +34,15 @@ import { definedOrError } from './utils'; import { OperationChunksIterator, OperationRowsIterator } from './utils/OperationIterator'; import HiveDriverError from './errors/HiveDriverError'; import IClientContext from './contracts/IClientContext'; +import { mapOperationTypeToTelemetryType, mapResultFormatToTelemetryType } from './telemetry/telemetryTypeMappers'; +import ExceptionClassifier from './telemetry/ExceptionClassifier'; +import { safeEmit } from './telemetry/telemetryUtils'; interface DBSQLOperationConstructorOptions { handle: TOperationHandle; directResults?: TSparkDirectResults; context: IClientContext; + sessionId?: string; } async function delay(ms?: number): Promise { @@ -76,9 +80,17 @@ export default class DBSQLOperation implements IOperation { private resultHandler?: ResultSlicer; - constructor({ handle, directResults, context }: DBSQLOperationConstructorOptions) { + // Telemetry tracking fields + private startTime: number = Date.now(); + + private pollCount: number = 0; + + private sessionId?: string; + + constructor({ handle, directResults, context, sessionId }: DBSQLOperationConstructorOptions) { this.operationHandle = handle; this.context = context; + this.sessionId = sessionId; const useOnlyPrefetchedResults = Boolean(directResults?.closeOperation); @@ -92,9 +104,13 @@ export default class DBSQLOperation implements IOperation { this.operationHandle, [directResults?.resultSet], useOnlyPrefetchedResults, + this.id, ); this.closeOperation = directResults?.closeOperation; this.context.getLogger().log(LogLevel.debug, `Operation created with id: ${this.id}`); + + // Emit statement.start telemetry event + this.emitStatementStart(); } public iterateChunks(options?: IteratorOptions): IOperationChunksIterator { @@ -166,6 +182,10 @@ export default class DBSQLOperation implements IOperation { * const result = await queryOperation.fetchChunk({maxRows: 1000}); */ public async fetchChunk(options?: FetchOptions): Promise> { + return this.withErrorTelemetry(() => this.fetchChunkInternal(options)); + } + + private async fetchChunkInternal(options?: FetchOptions): Promise> { await this.failIfClosed(); if (!this.operationHandle.hasResultSet) { @@ -225,6 +245,9 @@ export default class DBSQLOperation implements IOperation { return this.operationStatus; } + // Track poll count for telemetry + this.pollCount += 1; + const driver = await this.context.getDriver(); const response = await driver.getOperationStatus({ operationHandle: this.operationHandle, @@ -239,6 +262,10 @@ export default class DBSQLOperation implements IOperation { * @throws {StatusError} */ public async cancel(): Promise { + return this.withErrorTelemetry(() => this.cancelInternal()); + } + + private async cancelInternal(): Promise { if (this.closed || this.cancelled) { return Status.success(); } @@ -263,6 +290,10 @@ export default class DBSQLOperation implements IOperation { * @throws {StatusError} */ public async close(): Promise { + return this.withErrorTelemetry(() => this.closeInternal()); + } + + private async closeInternal(): Promise { if (this.closed || this.cancelled) { return Status.success(); } @@ -279,50 +310,75 @@ export default class DBSQLOperation implements IOperation { this.closed = true; const result = new Status(response.status); + // Emit statement.complete telemetry event + await this.emitStatementComplete(); + this.onClose?.(); return result; } public async finished(options?: FinishedOptions): Promise { - await this.failIfClosed(); - await this.waitUntilReady(options); + return this.withErrorTelemetry(async () => { + await this.failIfClosed(); + await this.waitUntilReady(options); + }); } public async hasMoreRows(): Promise { - // If operation is closed or cancelled - we should not try to get data from it - if (this.closed || this.cancelled) { - return false; - } + return this.withErrorTelemetry(async () => { + // If operation is closed or cancelled - we should not try to get data from it + if (this.closed || this.cancelled) { + return false; + } - // Wait for operation to finish before checking for more rows - // This ensures metadata can be fetched successfully - if (this.operationHandle.hasResultSet) { - await this.waitUntilReady(); - } + // Wait for operation to finish before checking for more rows + // This ensures metadata can be fetched successfully + if (this.operationHandle.hasResultSet) { + await this.waitUntilReady(); + } - // If we fetched all the data from server - check if there's anything buffered in result handler - const resultHandler = await this.getResultHandler(); - return resultHandler.hasMore(); + // If we fetched all the data from server - check if there's anything buffered in result handler + const resultHandler = await this.getResultHandler(); + return resultHandler.hasMore(); + }); } public async getSchema(options?: GetSchemaOptions): Promise { - await this.failIfClosed(); + return this.withErrorTelemetry(async () => { + await this.failIfClosed(); - if (!this.operationHandle.hasResultSet) { - return null; - } + if (!this.operationHandle.hasResultSet) { + return null; + } - await this.waitUntilReady(options); + await this.waitUntilReady(options); - this.context.getLogger().log(LogLevel.debug, `Fetching schema for operation with id: ${this.id}`); - const metadata = await this.fetchMetadata(); - return metadata.schema ?? null; + this.context.getLogger().log(LogLevel.debug, `Fetching schema for operation with id: ${this.id}`); + const metadata = await this.fetchMetadata(); + return metadata.schema ?? null; + }); } public async getMetadata(): Promise { - await this.failIfClosed(); - await this.waitUntilReady(); - return this.fetchMetadata(); + return this.withErrorTelemetry(async () => { + await this.failIfClosed(); + await this.waitUntilReady(); + return this.fetchMetadata(); + }); + } + + /** + * Wrap a public IOperation method so any thrown error is captured as an + * error telemetry event before being rethrown to the caller. Telemetry + * never alters the throw semantics. + */ + private async withErrorTelemetry(fn: () => Promise): Promise { + try { + return await fn(); + } catch (err: any) { + this.emitErrorEvent(err); + throw err; + } } private async failIfClosed(): Promise { @@ -441,7 +497,7 @@ export default class DBSQLOperation implements IOperation { case TSparkRowSetType.URL_BASED_SET: resultSource = new ArrowResultConverter( this.context, - new CloudFetchResultHandler(this.context, this._data, metadata), + new CloudFetchResultHandler(this.context, this._data, metadata, this.id), metadata, ); break; @@ -481,4 +537,68 @@ export default class DBSQLOperation implements IOperation { return response; } + + /** + * Emit statement.start telemetry event. + * CRITICAL: All exceptions swallowed and logged at LogLevel.debug ONLY. + */ + private emitStatementStart(): void { + safeEmit(this.context, (emitter) => { + emitter.emitStatementStart({ + statementId: this.id, + sessionId: this.sessionId || '', + operationType: mapOperationTypeToTelemetryType(this.operationHandle.operationType), + }); + }); + } + + /** + * Emit statement.complete telemetry event and complete aggregation. + * CRITICAL: All exceptions swallowed and logged at LogLevel.debug ONLY. + */ + private async emitStatementComplete(): Promise { + safeEmit(this.context, (emitter) => { + const aggregator = this.context.getTelemetryAggregator?.(); + if (!aggregator) return; + + // Use whatever metadata was already fetched by the result-handling + // path. Do NOT trigger a `getMetadata()` here — that issues a Thrift + // RPC on every close (doubles close latency for short DDL/DML) AND + // throws if the operation is already in an error/closed state, which + // would then fire spurious error telemetry from `getMetadata`'s error + // wrapper. + const resultFormat = mapResultFormatToTelemetryType(this.metadata?.resultFormat); + const latencyMs = Date.now() - this.startTime; + + emitter.emitStatementComplete({ + statementId: this.id, + sessionId: this.sessionId || '', + latencyMs, + resultFormat, + pollCount: this.pollCount, + }); + + aggregator.completeStatement(this.id); + }); + } + + /** + * Emit a telemetry error event for an exception thrown by an operation. + * Terminal errors (per `ExceptionClassifier`) trigger an immediate flush + * in the aggregator; retryable errors are buffered until the statement + * completes. All exceptions from this method itself are swallowed at + * debug level — telemetry must never break the driver. + */ + private emitErrorEvent(error: Error): void { + safeEmit(this.context, (emitter) => { + emitter.emitError({ + statementId: this.id, + sessionId: this.sessionId, + errorName: error.name || 'Error', + errorMessage: error.message || 'Unknown error', + errorStack: error.stack, + isTerminal: ExceptionClassifier.isTerminal(error), + }); + }); + } } diff --git a/lib/DBSQLSession.ts b/lib/DBSQLSession.ts index 9b4245c3..d15116a1 100644 --- a/lib/DBSQLSession.ts +++ b/lib/DBSQLSession.ts @@ -39,6 +39,7 @@ import StagingError from './errors/StagingError'; import { DBSQLParameter, DBSQLParameterValue } from './DBSQLParameter'; import ParameterError from './errors/ParameterError'; import IClientContext, { ClientConfig } from './contracts/IClientContext'; +import { safeEmit } from './telemetry/telemetryUtils'; // Explicitly promisify a callback-style `pipeline` because `node:stream/promises` is not available in Node 14 const pipeline = util.promisify(stream.pipeline); @@ -151,6 +152,8 @@ export default class DBSQLSession implements IDBSQLSession { private isOpen = true; + private openTime: number; + private serverProtocolVersion?: TProtocolVersion; public onClose?: () => void; @@ -169,6 +172,7 @@ export default class DBSQLSession implements IDBSQLSession { constructor({ handle, context, serverProtocolVersion }: DBSQLSessionConstructorOptions) { this.sessionHandle = handle; this.context = context; + this.openTime = Date.now(); // Get the server protocol version from the provided parameter (from TOpenSessionResp) this.serverProtocolVersion = serverProtocolVersion; this.context.getLogger().log(LogLevel.debug, `Session created with id: ${this.id}`); @@ -583,19 +587,36 @@ export default class DBSQLSession implements IDBSQLSession { // Close owned operations one by one, removing successfully closed ones from the list await this.operations.closeAll(); - const driver = await this.context.getDriver(); - const response = await driver.closeSession({ - sessionHandle: this.sessionHandle, - }); - // check status for being successful - Status.assert(response.status); - - // notify owner connection - this.onClose?.(); - this.isOpen = false; + const emitClose = () => { + // Emit connection.close regardless of whether closeSession succeeded — + // a failed close is the most diagnostic event for connection-failure rate. + safeEmit(this.context, (emitter) => { + emitter.emitConnectionClose({ + sessionId: this.id, + latencyMs: Date.now() - this.openTime, + }); + }); + }; - this.context.getLogger().log(LogLevel.debug, `Session closed with id: ${this.id}`); - return new Status(response.status); + try { + const driver = await this.context.getDriver(); + const response = await driver.closeSession({ + sessionHandle: this.sessionHandle, + }); + Status.assert(response.status); + + this.onClose?.(); + this.isOpen = false; + emitClose(); + + this.context.getLogger().log(LogLevel.debug, `Session closed with id: ${this.id}`); + return new Status(response.status); + } catch (err) { + this.onClose?.(); + this.isOpen = false; + emitClose(); + throw err; + } } private createOperation(response: OperationResponseShape): DBSQLOperation { @@ -605,6 +626,7 @@ export default class DBSQLSession implements IDBSQLSession { handle, directResults: response.directResults, context: this.context, + sessionId: this.id, }); this.operations.add(operation); diff --git a/lib/contracts/IClientContext.ts b/lib/contracts/IClientContext.ts index c7274a1b..b62fbf9d 100644 --- a/lib/contracts/IClientContext.ts +++ b/lib/contracts/IClientContext.ts @@ -2,6 +2,9 @@ import IDBSQLLogger from './IDBSQLLogger'; import IDriver from './IDriver'; import IConnectionProvider from '../connection/contracts/IConnectionProvider'; import IThriftClient from './IThriftClient'; +import IAuthentication from '../connection/contracts/IAuthentication'; +import type TelemetryEventEmitter from '../telemetry/TelemetryEventEmitter'; +import type MetricsAggregator from '../telemetry/MetricsAggregator'; export interface ClientConfig { directResultsDefaultMaxRows: number; @@ -37,6 +40,8 @@ export interface ClientConfig { telemetryMaxPendingMetrics?: number; telemetryMaxErrorsPerStatement?: number; telemetryStatementTtlMs?: number; + telemetryCloseTimeoutMs?: number; + telemetryMaxStatementMetrics?: number; userAgentEntry?: string; } @@ -50,4 +55,12 @@ export default interface IClientContext { getClient(): Promise; getDriver(): Promise; + + getAuthProvider?(): IAuthentication | undefined; + + /** @internal Telemetry event emitter, undefined when telemetry is disabled. */ + getTelemetryEmitter?(): TelemetryEventEmitter | undefined; + + /** @internal Telemetry aggregator, undefined when telemetry is disabled. */ + getTelemetryAggregator?(): MetricsAggregator | undefined; } diff --git a/lib/contracts/IDBSQLClient.ts b/lib/contracts/IDBSQLClient.ts index 4b2f39a4..11fd6713 100644 --- a/lib/contracts/IDBSQLClient.ts +++ b/lib/contracts/IDBSQLClient.ts @@ -54,6 +54,81 @@ export type ConnectionOptions = { socketTimeout?: number; proxy?: ProxyOptions; enableMetricViewMetadata?: boolean; + + /** + * Whether the driver emits telemetry events (connection / statement / + * cloud-fetch / error). Defaults to `true`. + * + * Activation is gated by **two** conditions: + * 1. This flag is `true` **and** + * 2. The remote feature flag for the workspace allows telemetry. + * + * Setting this to `false` is a hard, unconditional opt-out. Setting to + * `true` only requests telemetry; the workspace must also allow it. + * + * The environment variable `DATABRICKS_TELEMETRY_DISABLED` set to one of + * `1`, `true`, `yes`, or `on` (case-insensitive) overrides this flag and + * disables telemetry entirely. + */ + telemetryEnabled?: boolean; + + /** + * Maximum number of metrics to batch before flushing to the telemetry + * endpoint. Default 100. + */ + telemetryBatchSize?: number; + + /** + * How often to flush buffered telemetry metrics, in milliseconds. + * The flush timer is `unref()`'d so it cannot keep the Node.js process + * alive on its own. Default 5000ms. + */ + telemetryFlushIntervalMs?: number; + + /** + * Maximum retry attempts for a telemetry export *after* the initial call. + * Default 3. + */ + telemetryMaxRetries?: number; + + /** + * When `true`, telemetry is sent to the authenticated `/telemetry-ext` + * endpoint with workspace + session + statement IDs and a system + * configuration block. When `false`, only error names are emitted via the + * unauthenticated endpoint. Default `true`. + * + * Privacy-relevant: setting `false` minimizes the data surface at the + * cost of losing most observability. + */ + telemetryAuthenticatedExport?: boolean; + + /** + * Number of consecutive telemetry export failures before the per-host + * circuit breaker trips and pauses exports. Default 5. + */ + telemetryCircuitBreakerThreshold?: number; + + /** + * How long the circuit breaker stays open before re-probing the + * telemetry endpoint, in milliseconds. Default 60000ms (1 minute). + */ + telemetryCircuitBreakerTimeout?: number; + + /** + * Maximum wall-clock time `client.close()` will wait for the final + * telemetry flush HTTP POST. Bounds shutdown latency so callers + * doing `await client.close(); process.exit(0)` are not held up by a + * misbehaving telemetry endpoint. Default 2000ms. + */ + telemetryCloseTimeoutMs?: number; + + /** + * Hard cap on the per-statement aggregation map size. When the cap is + * reached, the oldest entry is evicted (its buffered errors are emitted + * as standalone metrics first so the first-failure signal survives). + * Default 5000. + */ + telemetryMaxStatementMetrics?: number; } & AuthOptions; export interface OpenSessionRequest { diff --git a/lib/index.ts b/lib/index.ts index 81e3aaae..bda86866 100644 --- a/lib/index.ts +++ b/lib/index.ts @@ -23,8 +23,6 @@ import { LogLevel } from './contracts/IDBSQLLogger'; // Re-export types for TypeScript users export type { default as ITokenProvider } from './connection/auth/tokenProvider/ITokenProvider'; -// Re-export telemetry error classes so consumers can instanceof-check rather -// than string-matching error messages. export { CircuitBreakerOpenError, CIRCUIT_BREAKER_OPEN_CODE } from './telemetry/CircuitBreaker'; export { TelemetryTerminalError } from './telemetry/DatabricksTelemetryExporter'; diff --git a/lib/result/CloudFetchResultHandler.ts b/lib/result/CloudFetchResultHandler.ts index 91878813..a9e977e0 100644 --- a/lib/result/CloudFetchResultHandler.ts +++ b/lib/result/CloudFetchResultHandler.ts @@ -6,6 +6,7 @@ import IResultsProvider, { ResultsProviderFetchNextOptions } from './IResultsPro import { ArrowBatch } from './utils'; import { LZ4 } from '../utils'; import { LogLevel } from '../contracts/IDBSQLLogger'; +import { safeEmit } from '../telemetry/telemetryUtils'; export default class CloudFetchResultHandler implements IResultsProvider { private readonly context: IClientContext; @@ -14,18 +15,24 @@ export default class CloudFetchResultHandler implements IResultsProvider = []; private downloadTasks: Array> = []; + private chunkIndex: number = 0; + constructor( context: IClientContext, source: IResultsProvider, - { lz4Compressed }: TGetResultSetMetadataResp, + metadata: TGetResultSetMetadataResp, + statementId?: string, ) { this.context = context; this.source = source; - this.isLZ4Compressed = lz4Compressed ?? false; + this.isLZ4Compressed = metadata.lz4Compressed ?? false; + this.statementId = statementId; if (this.isLZ4Compressed && !LZ4()) { throw new HiveDriverError('Cannot handle LZ4 compressed result: module `lz4` not installed'); @@ -106,6 +113,10 @@ export default class CloudFetchResultHandler implements IResultsProvider { + emitter.emitCloudFetchChunk({ + statementId: this.statementId, + chunkIndex, + latencyMs, + bytes, + compressed: this.isLZ4Compressed, + }); + }); + } } diff --git a/lib/result/RowSetProvider.ts b/lib/result/RowSetProvider.ts index f3fa4213..076bf4ce 100644 --- a/lib/result/RowSetProvider.ts +++ b/lib/result/RowSetProvider.ts @@ -2,6 +2,7 @@ import Int64 from 'node-int64'; import { TFetchOrientation, TFetchResultsResp, TOperationHandle, TRowSet } from '../../thrift/TCLIService_types'; import Status from '../dto/Status'; import IClientContext from '../contracts/IClientContext'; +import { safeEmit } from '../telemetry/telemetryUtils'; import IResultsProvider, { ResultsProviderFetchNextOptions } from './IResultsProvider'; import { getColumnValue } from './utils'; @@ -26,6 +27,10 @@ export default class RowSetProvider implements IResultsProvider, returnOnlyPrefetchedResults: boolean, + statementId?: string, ) { this.context = context; this.operationHandle = operationHandle; + this.statementId = statementId; prefetchedResults.forEach((item) => { if (item) { this.prefetchedResults.push(item); @@ -83,16 +90,47 @@ export default class RowSetProvider implements IResultsProvider { + let bytes = 0; + const arrowBatches = response.results?.arrowBatches; + if (arrowBatches) { + for (const batch of arrowBatches) { + bytes += batch.batch?.length ?? 0; + } + } + + emitter.emitCloudFetchChunk({ + statementId: this.statementId, + chunkIndex: this.chunkIndex, + latencyMs, + bytes, + }); + this.chunkIndex += 1; + }); + } + public async hasMore() { // If there are prefetched results available - return `true` regardless of // the actual state of `hasMoreRows` flag (because we actually have some data) diff --git a/lib/telemetry/DatabricksTelemetryExporter.ts b/lib/telemetry/DatabricksTelemetryExporter.ts index eeeb5eea..74e73737 100644 --- a/lib/telemetry/DatabricksTelemetryExporter.ts +++ b/lib/telemetry/DatabricksTelemetryExporter.ts @@ -289,12 +289,16 @@ export default class DatabricksTelemetryExporter { } private async getAuthHeaders(): Promise> { - if (!this.authProvider) { + // Prefer the explicitly-injected auth provider; fall back to the context + // (used when a shared TelemetryClient resolves auth through its FIFO of + // registered DBSQLClients). + const authProvider = this.authProvider ?? this.context.getAuthProvider?.(); + if (!authProvider) { return {}; } const logger = this.context.getLogger(); try { - return normalizeHeaders(await this.authProvider.authenticate()); + return normalizeHeaders(await authProvider.authenticate()); } catch (error: any) { logger.log(LogLevel.debug, `Telemetry: auth provider threw: ${error?.message ?? error}`); return {}; @@ -340,22 +344,27 @@ export default class DatabricksTelemetryExporter { }, }; - if (metric.metricType === 'connection' && metric.driverConfig && includeCorrelation) { - // system_configuration is a high-entropy client fingerprint (OS, arch, - // locale, process, runtime). Only ship on the authenticated path. - log.entry.sql_driver_log.system_configuration = { - driver_version: metric.driverConfig.driverVersion, - driver_name: metric.driverConfig.driverName, - runtime_name: 'Node.js', - runtime_version: metric.driverConfig.nodeVersion, - runtime_vendor: metric.driverConfig.runtimeVendor, - os_name: metric.driverConfig.platform, - os_version: metric.driverConfig.osVersion, - os_arch: metric.driverConfig.osArch, - locale_name: metric.driverConfig.localeName, - char_set_encoding: metric.driverConfig.charSetEncoding, - process_name: sanitizeProcessName(metric.driverConfig.processName) || undefined, - }; + if (metric.metricType === 'connection') { + if (metric.latencyMs !== undefined) { + log.entry.sql_driver_log.operation_latency_ms = metric.latencyMs; + } + if (metric.driverConfig && includeCorrelation) { + // system_configuration is a high-entropy client fingerprint (OS, arch, + // locale, process, runtime). Only ship on the authenticated path. + log.entry.sql_driver_log.system_configuration = { + driver_version: metric.driverConfig.driverVersion, + driver_name: metric.driverConfig.driverName, + runtime_name: 'Node.js', + runtime_version: metric.driverConfig.nodeVersion, + runtime_vendor: metric.driverConfig.runtimeVendor, + os_name: metric.driverConfig.platform, + os_version: metric.driverConfig.osVersion, + os_arch: metric.driverConfig.osArch, + locale_name: metric.driverConfig.localeName, + char_set_encoding: metric.driverConfig.charSetEncoding, + process_name: sanitizeProcessName(metric.driverConfig.processName) || undefined, + }; + } } else if (metric.metricType === 'statement') { log.entry.sql_driver_log.operation_latency_ms = metric.latencyMs; @@ -364,10 +373,13 @@ export default class DatabricksTelemetryExporter { execution_result: metric.resultFormat, }; - if (metric.chunkCount && metric.chunkCount > 0) { + if ((metric.chunkCount ?? 0) > 0) { log.entry.sql_driver_log.sql_operation.chunk_details = { total_chunks_present: metric.chunkCount, total_chunks_iterated: metric.chunkCount, + initial_chunk_latency_millis: metric.chunkInitialLatencyMs, + slowest_chunk_latency_millis: metric.chunkSlowestLatencyMs, + sum_chunks_download_time_millis: metric.chunkSumLatencyMs, }; } } diff --git a/lib/telemetry/FeatureFlagCache.ts b/lib/telemetry/FeatureFlagCache.ts index 06bc59a9..47fc9e8d 100644 --- a/lib/telemetry/FeatureFlagCache.ts +++ b/lib/telemetry/FeatureFlagCache.ts @@ -202,11 +202,15 @@ export default class FeatureFlagCache { } private async getAuthHeaders(): Promise> { - if (!this.authProvider) { + // Prefer the explicitly-injected auth provider; fall back to the context + // (used when a shared TelemetryClient resolves auth through its FIFO of + // registered DBSQLClients). Mirrors DatabricksTelemetryExporter.getAuthHeaders. + const authProvider = this.authProvider ?? this.context.getAuthProvider?.(); + if (!authProvider) { return {}; } try { - return normalizeHeaders(await this.authProvider.authenticate()); + return normalizeHeaders(await authProvider.authenticate()); } catch (error: any) { this.context.getLogger().log(LogLevel.debug, `Feature flag auth failed: ${error?.message ?? error}`); return {}; diff --git a/lib/telemetry/MetricsAggregator.ts b/lib/telemetry/MetricsAggregator.ts index d160db10..2bcc69fe 100644 --- a/lib/telemetry/MetricsAggregator.ts +++ b/lib/telemetry/MetricsAggregator.ts @@ -31,6 +31,9 @@ interface StatementTelemetryDetails { bytesDownloaded: number; pollCount: number; compressionEnabled?: boolean; + chunkInitialLatencyMs?: number; + chunkSlowestLatencyMs?: number; + chunkSumLatencyMs: number; errors: TelemetryEvent[]; } @@ -49,6 +52,11 @@ export default class MetricsAggregator { private flushTimer: NodeJS.Timeout | null = null; + // Single in-flight flush serializer. Concurrent triggers (batch hit, periodic + // tick, terminal-error, manual flush) all share one HTTP POST so the user's + // socket pool can't be starved by a slow telemetry endpoint. + private flushInFlight: Promise | null = null; + private closed = false; private closing = false; @@ -63,6 +71,8 @@ export default class MetricsAggregator { private statementTtlMs: number; + private maxStatementMetrics: number; + constructor(private context: IClientContext, private exporter: DatabricksTelemetryExporter) { try { const config = context.getConfig(); @@ -72,6 +82,8 @@ export default class MetricsAggregator { this.maxErrorsPerStatement = config.telemetryMaxErrorsPerStatement ?? DEFAULT_TELEMETRY_CONFIG.maxErrorsPerStatement; this.statementTtlMs = config.telemetryStatementTtlMs ?? DEFAULT_TELEMETRY_CONFIG.statementTtlMs; + this.maxStatementMetrics = + config.telemetryMaxStatementMetrics ?? DEFAULT_TELEMETRY_CONFIG.maxStatementMetrics; this.startFlushTimer(); } catch (error: any) { @@ -83,6 +95,7 @@ export default class MetricsAggregator { this.maxPendingMetrics = DEFAULT_TELEMETRY_CONFIG.maxPendingMetrics; this.maxErrorsPerStatement = DEFAULT_TELEMETRY_CONFIG.maxErrorsPerStatement; this.statementTtlMs = DEFAULT_TELEMETRY_CONFIG.statementTtlMs; + this.maxStatementMetrics = DEFAULT_TELEMETRY_CONFIG.maxStatementMetrics; } } @@ -92,7 +105,12 @@ export default class MetricsAggregator { try { if (event.eventType === TelemetryEventType.CONNECTION_OPEN) { - this.processConnectionEvent(event); + this.processConnectionEvent(event, 'CREATE_SESSION'); + return; + } + + if (event.eventType === TelemetryEventType.CONNECTION_CLOSE) { + this.processConnectionEvent(event, 'DELETE_SESSION'); return; } @@ -109,13 +127,15 @@ export default class MetricsAggregator { } } - private processConnectionEvent(event: TelemetryEvent): void { + private processConnectionEvent(event: TelemetryEvent, operationType: 'CREATE_SESSION' | 'DELETE_SESSION'): void { const metric: TelemetryMetric = { metricType: 'connection', timestamp: event.timestamp, sessionId: event.sessionId, workspaceId: event.workspaceId, driverConfig: event.driverConfig, + operationType, + latencyMs: event.latencyMs, }; this.addPendingMetric(metric); @@ -182,9 +202,19 @@ export default class MetricsAggregator { case TelemetryEventType.STATEMENT_COMPLETE: details.executionLatencyMs = event.latencyMs; details.resultFormat = event.resultFormat; - details.chunkCount = event.chunkCount ?? 0; - details.bytesDownloaded = event.bytesDownloaded ?? 0; - details.pollCount = event.pollCount ?? 0; + // STATEMENT_COMPLETE may not carry chunk counts (the operation + // doesn't always know them at close time); only override when the + // emit explicitly supplied a value, otherwise the values accumulated + // from CLOUDFETCH_CHUNK survive. + if (event.chunkCount !== undefined) { + details.chunkCount = event.chunkCount; + } + if (event.bytesDownloaded !== undefined) { + details.bytesDownloaded = event.bytesDownloaded; + } + if (event.pollCount !== undefined) { + details.pollCount = event.pollCount; + } break; case TelemetryEventType.CLOUDFETCH_CHUNK: @@ -193,6 +223,17 @@ export default class MetricsAggregator { if (event.compressed !== undefined) { details.compressionEnabled = event.compressed; } + // Per-chunk timing aggregation. Only record positive latencies — keeps + // prefetched/cached pages out of the timing stats. + if (event.latencyMs !== undefined && event.latencyMs > 0) { + if (details.chunkInitialLatencyMs === undefined) { + details.chunkInitialLatencyMs = event.latencyMs; + } + if (details.chunkSlowestLatencyMs === undefined || event.latencyMs > details.chunkSlowestLatencyMs) { + details.chunkSlowestLatencyMs = event.latencyMs; + } + details.chunkSumLatencyMs += event.latencyMs; + } break; default: @@ -204,6 +245,12 @@ export default class MetricsAggregator { const statementId = event.statementId!; if (!this.statementMetrics.has(statementId)) { + // Hard cap on map size — abandoned operations or buggy upstreams that + // emit errors with random fresh statementIds would otherwise grow this + // map unbounded for up to statementTtlMs. + if (this.statementMetrics.size >= this.maxStatementMetrics) { + this.evictOldestStatement(); + } this.statementMetrics.set(statementId, { statementId, sessionId: event.sessionId!, @@ -212,6 +259,7 @@ export default class MetricsAggregator { chunkCount: 0, bytesDownloaded: 0, pollCount: 0, + chunkSumLatencyMs: 0, errors: [], }); } @@ -219,6 +267,34 @@ export default class MetricsAggregator { return this.statementMetrics.get(statementId)!; } + /** + * Drop the oldest entry by insertion order to make room. Emits its buffered + * errors as standalone metrics first so the first-failure signal survives. + * Map iteration order is insertion order in JS. + */ + private evictOldestStatement(): void { + const oldest = this.statementMetrics.keys().next(); + if (oldest.done) return; + const id = oldest.value; + const details = this.statementMetrics.get(id)!; + for (const errorEvent of details.errors) { + this.addPendingMetric({ + metricType: 'error', + timestamp: errorEvent.timestamp, + sessionId: details.sessionId, + statementId: details.statementId, + workspaceId: details.workspaceId, + errorName: errorEvent.errorName, + errorMessage: errorEvent.errorMessage, + errorStack: errorEvent.errorStack, + }); + } + this.statementMetrics.delete(id); + this.context + .getLogger() + .log(LogLevel.debug, `MetricsAggregator: evicted oldest statement ${id} (max=${this.maxStatementMetrics})`); + } + /** * Drop entries older than `statementTtlMs`, emitting their buffered error * events as standalone metrics first so the first-failure signal survives @@ -269,11 +345,16 @@ export default class MetricsAggregator { sessionId: details.sessionId, statementId: details.statementId, workspaceId: details.workspaceId, + operationType: details.operationType, latencyMs: details.executionLatencyMs, resultFormat: details.resultFormat, chunkCount: details.chunkCount, + chunkInitialLatencyMs: details.chunkInitialLatencyMs, + chunkSlowestLatencyMs: details.chunkSlowestLatencyMs, + chunkSumLatencyMs: details.chunkSumLatencyMs > 0 ? details.chunkSumLatencyMs : undefined, bytesDownloaded: details.bytesDownloaded, pollCount: details.pollCount, + compressed: details.compressionEnabled, }; this.addPendingMetric(metric); @@ -344,6 +425,20 @@ export default class MetricsAggregator { * `process.exit()` after `client.close()` doesn't truncate the POST. */ async flush(resetTimer: boolean = true): Promise { + // Coalesce concurrent flush callers onto the in-flight promise so we + // never run two HTTP POSTs in parallel against the telemetry endpoint. + // Pending metrics arriving while flushInFlight is set will be picked up + // by the next caller. + if (this.flushInFlight) { + return this.flushInFlight; + } + this.flushInFlight = this.runFlush(resetTimer).finally(() => { + this.flushInFlight = null; + }); + return this.flushInFlight; + } + + private async runFlush(resetTimer: boolean): Promise { const logger = this.context.getLogger(); let exportPromise: Promise | null = null; @@ -426,7 +521,37 @@ export default class MetricsAggregator { } this.closed = true; - await this.flush(false); + // Cap the wait on the final flush so a flapping telemetry endpoint + // can't hold up the user's process.exit(0). The in-flight POST is + // abandoned past the deadline; data loss is preferable to a hung exit. + const timeoutMs = this.context.getConfig().telemetryCloseTimeoutMs ?? 2000; + let timeoutHandle: NodeJS.Timeout | null = null; + const timeoutPromise = new Promise((resolve) => { + timeoutHandle = setTimeout(() => { + logger.log(LogLevel.debug, `MetricsAggregator.close: flush timed out after ${timeoutMs}ms`); + resolve(); + }, timeoutMs); + timeoutHandle.unref?.(); + }); + // Drain pattern: if a batch-trigger flush was already in-flight when + // close() ran, it captured a snapshot before completeStatement above + // appended the close-time metrics. Wait for that to finish and then + // run a fresh flush that picks up whatever's still in pendingMetrics. + const drain = async (): Promise => { + if (this.flushInFlight) { + await this.flushInFlight; + } + if (this.pendingMetrics.length > 0) { + await this.flush(false); + } + }; + try { + await Promise.race([drain(), timeoutPromise]); + } finally { + if (timeoutHandle) { + clearTimeout(timeoutHandle); + } + } } catch (error: any) { logger.log(LogLevel.debug, `MetricsAggregator.close error: ${error.message}`); } diff --git a/lib/telemetry/TelemetryClient.ts b/lib/telemetry/TelemetryClient.ts index d74d1417..e7356046 100644 --- a/lib/telemetry/TelemetryClient.ts +++ b/lib/telemetry/TelemetryClient.ts @@ -14,51 +14,240 @@ * limitations under the License. */ -import IClientContext from '../contracts/IClientContext'; -import { LogLevel } from '../contracts/IDBSQLLogger'; +import IClientContext, { ClientConfig } from '../contracts/IClientContext'; +import IDBSQLLogger, { LogLevel } from '../contracts/IDBSQLLogger'; +import IConnectionProvider from '../connection/contracts/IConnectionProvider'; +import IThriftClient from '../contracts/IThriftClient'; +import IDriver from '../contracts/IDriver'; +import IAuthentication from '../connection/contracts/IAuthentication'; +import { CircuitBreakerRegistry } from './CircuitBreaker'; +import DatabricksTelemetryExporter from './DatabricksTelemetryExporter'; +import MetricsAggregator from './MetricsAggregator'; +import FeatureFlagCache from './FeatureFlagCache'; /** - * Telemetry client for a specific host. - * Managed by TelemetryClientProvider with reference counting. - * One client instance is shared across all connections to the same host. + * Per-host telemetry resource owner. Held by `TelemetryClientProvider` + * (process-wide singleton) and shared across every `DBSQLClient` that + * connects to the same host. + * + * Owns the host-scoped triad — `MetricsAggregator`, `DatabricksTelemetryExporter`, + * `CircuitBreakerRegistry`, `FeatureFlagCache` — and implements `IClientContext` + * itself so those owned components have a stable context that outlives any + * single `DBSQLClient`. The first registered `DBSQLClient`'s logger and config + * are snapshotted; subsequent registrants donate their connection providers + * and auth providers, and the `TelemetryClient` falls through them in + * registration order when the current head closes. + * + * Why share at host granularity: + * - Circuit-breaker state for `host` is correct only if all clients hitting + * the same endpoint share counters (5 failures means 5 actual failures, not + * 5×N for N independent `DBSQLClient` instances). + * - Feature-flag cache has a per-host TTL; deduping the GET prevents + * thundering-herd on cold cache. + * - Metric batches mix events from every active client to the same host — + * one HTTP POST per `flushIntervalMs` instead of N. */ -class TelemetryClient { +class TelemetryClient implements IClientContext { private closed: boolean = false; - constructor(private context: IClientContext, private host: string) { - const logger = context.getLogger(); - logger.log(LogLevel.debug, `Created TelemetryClient for host: ${host}`); + private readonly logger: IDBSQLLogger; + + private readonly config: ClientConfig; + + private readonly circuitBreakerRegistry: CircuitBreakerRegistry; + + private readonly featureFlagCache: FeatureFlagCache; + + private readonly exporter: DatabricksTelemetryExporter; + + private readonly aggregator: MetricsAggregator; + + // FIFO so the first-registered client's connection/auth providers are tried + // first. Falls through to later registrants when an earlier one is closed. + private contexts: IClientContext[] = []; + + private authProviders: IAuthentication[] = []; + + constructor(initialContext: IClientContext, public readonly host: string) { + this.logger = initialContext.getLogger(); + // Snapshot config at first registration. Subsequent clients with + // divergent telemetry knobs (`telemetryBatchSize` etc.) inherit the + // first-registrant's tuning — documented invariant. + this.config = initialContext.getConfig(); + this.contexts.push(initialContext); + const auth = initialContext.getAuthProvider?.(); + if (auth) { + this.authProviders.push(auth); + } + + this.circuitBreakerRegistry = new CircuitBreakerRegistry(this); + this.featureFlagCache = new FeatureFlagCache(this); + // Register this host with the feature-flag cache so isTelemetryEnabled() + // does not short-circuit to false. close() releases via releaseContext(). + this.featureFlagCache.getOrCreateContext(host); + this.exporter = new DatabricksTelemetryExporter(this, host, this.circuitBreakerRegistry); + this.aggregator = new MetricsAggregator(this, this.exporter); + + this.logger.log(LogLevel.debug, `Created TelemetryClient for host: ${host}`); } /** - * Gets the host associated with this client. + * Add another `DBSQLClient`'s context to the pool. Tracked in registration + * order; `getConnectionProvider()` / `getAuthProvider()` walk the list and + * use the first entry that's still usable. */ - getHost(): string { - return this.host; + registerContext(context: IClientContext): void { + if (!this.contexts.includes(context)) { + this.contexts.push(context); + // Warn when subsequent registrants pass telemetry knobs that diverge + // from the first-registrant's snapshot — those values are silently + // ignored. Privacy-relevant for telemetryAuthenticatedExport. + this.warnOnConfigDivergence(context.getConfig()); + } + const auth = context.getAuthProvider?.(); + if (auth && !this.authProviders.includes(auth)) { + this.authProviders.push(auth); + } + } + + private warnOnConfigDivergence(other: ClientConfig): void { + const keys: Array = [ + 'telemetryAuthenticatedExport', + 'telemetryBatchSize', + 'telemetryFlushIntervalMs', + 'telemetryMaxRetries', + 'telemetryCircuitBreakerThreshold', + 'telemetryCircuitBreakerTimeout', + // Privacy-relevant: User-Agent is snapshotted from the first registrant + // and shared across the host. Multi-tenant SaaS layers with per-tenant + // userAgentEntry values would otherwise silently ship under tenant-1's UA. + 'userAgentEntry', + ]; + const diverged = keys.filter((k) => other[k] !== undefined && other[k] !== this.config[k]); + if (diverged.length > 0) { + this.logger.log( + LogLevel.warn, + `TelemetryClient(${this.host}): registered context's telemetry settings ` + + `[${diverged.join(', ')}] differ from the first registrant's; the new values will be ignored.`, + ); + } } /** - * Checks if the client has been closed. + * Remove a `DBSQLClient`'s context from the pool. Called by + * `TelemetryClientProvider.releaseClient` before refcount decrement so the + * exporter doesn't keep trying to use a closed context. */ + unregisterContext(context: IClientContext): void { + this.contexts = this.contexts.filter((c) => c !== context); + const auth = context.getAuthProvider?.(); + if (auth) { + this.authProviders = this.authProviders.filter((a) => a !== auth); + } + } + + // -- IClientContext -- + + getConfig(): ClientConfig { + return this.config; + } + + getLogger(): IDBSQLLogger { + return this.logger; + } + + async getConnectionProvider(): Promise { + let lastErr: unknown; + for (const ctx of this.contexts) { + try { + // Sequential fall-through is intentional — each context returns the + // same shared connection provider; we try the next registrant only + // when the current head is unusable. + // eslint-disable-next-line no-await-in-loop + return await ctx.getConnectionProvider(); + } catch (err) { + lastErr = err; + } + } + throw lastErr instanceof Error + ? lastErr + : new Error(`TelemetryClient: no connection provider available for host ${this.host}`); + } + + async getClient(): Promise { + if (this.contexts.length === 0) { + throw new Error(`TelemetryClient: no client available for host ${this.host}`); + } + return this.contexts[0].getClient(); + } + + async getDriver(): Promise { + if (this.contexts.length === 0) { + throw new Error(`TelemetryClient: no driver available for host ${this.host}`); + } + return this.contexts[0].getDriver(); + } + + getAuthProvider(): IAuthentication | undefined { + return this.authProviders[0]; + } + + getTelemetryEmitter(): undefined { + // The shared TelemetryClient holds the aggregator; emitters remain + // per-DBSQLClient so each can respect its own `telemetryEnabled` flag. + return undefined; + } + + getTelemetryAggregator(): MetricsAggregator { + return this.aggregator; + } + + // -- shared resource accessors -- + + getExporter(): DatabricksTelemetryExporter { + return this.exporter; + } + + getAggregator(): MetricsAggregator { + return this.aggregator; + } + + getFeatureFlagCache(): FeatureFlagCache { + return this.featureFlagCache; + } + + getHost(): string { + return this.host; + } + isClosed(): boolean { return this.closed; } /** - * Closes the telemetry client and releases resources. - * Should only be called by TelemetryClientProvider when reference count reaches zero. + * Drain pending metrics and tear down owned resources. Called by + * `TelemetryClientProvider.releaseClient` when refCount hits zero. */ - close(): void { - if (this.closed) { - return; + async close(): Promise { + if (this.closed) return; + this.closed = true; + this.logger.log(LogLevel.debug, `Closing TelemetryClient for host: ${this.host}`); + try { + await this.aggregator.close(); + } catch (err) { + this.logger.log(LogLevel.debug, `TelemetryClient aggregator close error: ${(err as Error).message}`); + } + try { + this.exporter.dispose(); + } catch (err) { + this.logger.log(LogLevel.debug, `TelemetryClient exporter dispose error: ${(err as Error).message}`); } try { - this.context.getLogger().log(LogLevel.debug, `Closing TelemetryClient for host: ${this.host}`); - } catch { - // swallow - } finally { - this.closed = true; + this.featureFlagCache.releaseContext(this.host); + } catch (err) { + this.logger.log(LogLevel.debug, `TelemetryClient FFCache release error: ${(err as Error).message}`); } + this.logger.log(LogLevel.debug, `Closed TelemetryClient for host: ${this.host}`); } } diff --git a/lib/telemetry/TelemetryClientProvider.ts b/lib/telemetry/TelemetryClientProvider.ts index c0da29f0..49d570ff 100644 --- a/lib/telemetry/TelemetryClientProvider.ts +++ b/lib/telemetry/TelemetryClientProvider.ts @@ -18,50 +18,56 @@ import IClientContext from '../contracts/IClientContext'; import { LogLevel } from '../contracts/IDBSQLLogger'; import TelemetryClient from './TelemetryClient'; -/** - * Holds a telemetry client and its reference count. - * The reference count tracks how many connections are using this client. - */ interface TelemetryClientHolder { client: TelemetryClient; refCount: number; } // Soft cap on distinct host entries. Above this the provider warns once so a -// misconfigured caller (per-request hosts, unnormalized aliases) is visible in -// logs rather than silently growing the map. +// misconfigured caller (per-request hosts, unnormalized aliases) is visible +// in logs rather than silently growing the map. const MAX_CLIENTS_SOFT_LIMIT = 128; /** - * Manages one telemetry client per host. - * Prevents rate limiting by sharing clients across connections to the same host. - * Instance-based (not singleton), stored in DBSQLClient. + * Process-wide registry of `TelemetryClient`s, one per host. Multiple + * `DBSQLClient` instances connecting to the same host share the same + * `TelemetryClient`, which owns the host-scoped circuit breaker, feature + * flag cache, exporter, and aggregator. * - * Reference counts are incremented and decremented synchronously, and - * `close()` is sync today, so there is no await between map mutation and - * client teardown. The map entry is removed before `close()` runs so a - * concurrent `getOrCreateClient` call for the same host gets a fresh - * instance rather than receiving this closing one. When `close()` becomes - * async (e.g. HTTP flush in [5/7]) the flow will need to `await` after the - * delete to preserve the same invariant. + * Singleton because the resources we're sharing — circuit-breaker counters, + * batched HTTP exports — are correct only at process scope. Per-`DBSQLClient` + * provider scope (the previous design) deduplicates nothing. + * + * `getOrCreateClient` is sync: caller increments refcount, registers its + * context+auth, and walks away. `releaseClient` is `async` because the + * underlying `TelemetryClient.close()` awaits the final flush. */ class TelemetryClientProvider { - private clients: Map; + private static instance: TelemetryClientProvider | undefined; + + private clients = new Map(); private softLimitWarned = false; - constructor(private context: IClientContext) { - this.clients = new Map(); - const logger = context.getLogger(); - logger.log(LogLevel.debug, 'Created TelemetryClientProvider'); + // Production code should use `TelemetryClientProvider.getInstance()` for + // the process-wide singleton. The constructor remains public so unit tests + // can build an isolated provider with its own map. Deliberate no-op body — + // initial state is set inline on the field declarations above. + + static getInstance(): TelemetryClientProvider { + if (!TelemetryClientProvider.instance) { + TelemetryClientProvider.instance = new TelemetryClientProvider(); + } + return TelemetryClientProvider.instance; } /** - * Canonicalize host so aliases (scheme, default port, trailing slash, case, - * trailing dot, surrounding whitespace) map to the same entry. Kept to a - * lightweight lexical normalization — `buildTelemetryUrl` still performs - * the strict security validation when a request is actually built. + * @internal Reset for tests. Production code should never call this. */ + static resetInstance(): void { + TelemetryClientProvider.instance = undefined; + } + private static normalizeHostKey(host: string): string { return host .trim() @@ -73,23 +79,17 @@ class TelemetryClientProvider { } /** - * Gets or creates a telemetry client for the specified host. - * Increments the reference count for the client. - * - * @param host The host identifier (e.g., "workspace.cloud.databricks.com") - * @returns The telemetry client for the host + * Get or create the `TelemetryClient` for `host`, registering `context` as + * a participant. Increments refcount. */ - getOrCreateClient(host: string): TelemetryClient { - const logger = this.context.getLogger(); + getOrCreateClient(context: IClientContext, host: string): TelemetryClient { + const logger = context.getLogger(); const key = TelemetryClientProvider.normalizeHostKey(host); let holder = this.clients.get(key); if (!holder) { - const client = new TelemetryClient(this.context, key); - holder = { - client, - refCount: 0, - }; + const client = new TelemetryClient(context, key); + holder = { client, refCount: 0 }; this.clients.set(key, holder); logger.log(LogLevel.debug, `Created new TelemetryClient for host: ${host}`); @@ -100,22 +100,21 @@ class TelemetryClientProvider { `TelemetryClientProvider has ${this.clients.size} distinct hosts — possible alias or leak`, ); } + } else { + holder.client.registerContext(context); } holder.refCount += 1; logger.log(LogLevel.debug, `TelemetryClient reference count for ${host}: ${holder.refCount}`); - return holder.client; } /** - * Releases a telemetry client for the specified host. - * Decrements the reference count and closes the client when it reaches zero. - * - * @param host The host identifier + * Release `context`'s registration. When the last `DBSQLClient` releases, + * the underlying `TelemetryClient.close()` runs and the entry is removed. */ - releaseClient(host: string): void { - const logger = this.context.getLogger(); + async releaseClient(context: IClientContext, host: string): Promise { + const logger = context.getLogger(); const key = TelemetryClientProvider.normalizeHostKey(host); const holder = this.clients.get(key); @@ -124,24 +123,22 @@ class TelemetryClientProvider { return; } - // Guard against double-release: a caller releasing more times than it got - // would otherwise drive refCount negative and close a client another - // caller is still holding. Warn loudly and refuse to decrement further. if (holder.refCount <= 0) { logger.log(LogLevel.warn, `Unbalanced release for TelemetryClient host: ${host}`); return; } + holder.client.unregisterContext(context); holder.refCount -= 1; logger.log(LogLevel.debug, `TelemetryClient reference count for ${host}: ${holder.refCount}`); - // Close and remove client when reference count reaches zero. - // Remove from map before calling close so a concurrent getOrCreateClient - // creates a fresh client rather than receiving this closing one. if (holder.refCount <= 0) { + // Remove from map BEFORE awaiting close so a concurrent + // getOrCreateClient creates a fresh instance rather than receiving + // this closing one. this.clients.delete(key); try { - holder.client.close(); + await holder.client.close(); logger.log(LogLevel.debug, `Closed and removed TelemetryClient for host: ${host}`); } catch (error) { const msg = error instanceof Error ? error.message : String(error); @@ -150,17 +147,13 @@ class TelemetryClientProvider { } } - /** - * @internal Exposed for testing only. - */ + /** @internal Exposed for testing only. */ getRefCount(host: string): number { const holder = this.clients.get(TelemetryClientProvider.normalizeHostKey(host)); return holder ? holder.refCount : 0; } - /** - * @internal Exposed for testing only. - */ + /** @internal Exposed for testing only. */ getActiveClients(): Map { const result = new Map(); for (const [host, holder] of this.clients.entries()) { diff --git a/lib/telemetry/TelemetryEventEmitter.ts b/lib/telemetry/TelemetryEventEmitter.ts index bbb0b757..8b9e1ef8 100644 --- a/lib/telemetry/TelemetryEventEmitter.ts +++ b/lib/telemetry/TelemetryEventEmitter.ts @@ -45,7 +45,12 @@ export default class TelemetryEventEmitter extends EventEmitter { * * @param data Connection event data including sessionId, workspaceId, and driverConfig */ - emitConnectionOpen(data: { sessionId: string; workspaceId: string; driverConfig: DriverConfiguration }): void { + emitConnectionOpen(data: { + sessionId: string; + workspaceId: string; + driverConfig: DriverConfiguration; + latencyMs: number; + }): void { if (!this.enabled) return; const logger = this.context.getLogger(); @@ -56,6 +61,7 @@ export default class TelemetryEventEmitter extends EventEmitter { sessionId: data.sessionId, workspaceId: data.workspaceId, driverConfig: data.driverConfig, + latencyMs: data.latencyMs, }; this.emit(TelemetryEventType.CONNECTION_OPEN, event); } catch (error: any) { @@ -64,6 +70,29 @@ export default class TelemetryEventEmitter extends EventEmitter { } } + /** + * Emit a connection close event. + * + * @param data Connection close event data including sessionId and latencyMs + */ + emitConnectionClose(data: { sessionId: string; latencyMs: number }): void { + if (!this.enabled) return; + + const logger = this.context.getLogger(); + try { + const event: TelemetryEvent = { + eventType: TelemetryEventType.CONNECTION_CLOSE, + timestamp: Date.now(), + sessionId: data.sessionId, + latencyMs: data.latencyMs, + }; + this.emit(TelemetryEventType.CONNECTION_CLOSE, event); + } catch (error: any) { + // Swallow all exceptions - log at debug level only + logger.log(LogLevel.debug, `Error emitting connection close event: ${error.message}`); + } + } + /** * Emit a statement start event. * diff --git a/lib/telemetry/telemetryTypeMappers.ts b/lib/telemetry/telemetryTypeMappers.ts new file mode 100644 index 00000000..d022739d --- /dev/null +++ b/lib/telemetry/telemetryTypeMappers.ts @@ -0,0 +1,71 @@ +/** + * Copyright (c) 2025 Databricks Contributors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { TOperationType, TSparkRowSetType } from '../../thrift/TCLIService_types'; + +/** + * Map Thrift TOperationType to telemetry Operation.Type enum string. + * Returns 'TYPE_UNSPECIFIED' if operationType is undefined or unknown. + */ +export function mapOperationTypeToTelemetryType(operationType?: TOperationType): string { + if (operationType === undefined) { + return 'TYPE_UNSPECIFIED'; + } + + switch (operationType) { + case TOperationType.EXECUTE_STATEMENT: + return 'EXECUTE_STATEMENT'; + case TOperationType.GET_TYPE_INFO: + return 'LIST_TYPE_INFO'; + case TOperationType.GET_CATALOGS: + return 'LIST_CATALOGS'; + case TOperationType.GET_SCHEMAS: + return 'LIST_SCHEMAS'; + case TOperationType.GET_TABLES: + return 'LIST_TABLES'; + case TOperationType.GET_TABLE_TYPES: + return 'LIST_TABLE_TYPES'; + case TOperationType.GET_COLUMNS: + return 'LIST_COLUMNS'; + case TOperationType.GET_FUNCTIONS: + return 'LIST_FUNCTIONS'; + case TOperationType.UNKNOWN: + default: + return 'TYPE_UNSPECIFIED'; + } +} + +/** + * Map Thrift TSparkRowSetType to telemetry ExecutionResult.Format enum string. + */ +export function mapResultFormatToTelemetryType(resultFormat?: TSparkRowSetType): string | undefined { + if (resultFormat === undefined) { + return undefined; + } + + switch (resultFormat) { + case TSparkRowSetType.ARROW_BASED_SET: + return 'INLINE_ARROW'; + case TSparkRowSetType.COLUMN_BASED_SET: + return 'COLUMNAR_INLINE'; + case TSparkRowSetType.ROW_BASED_SET: + return 'INLINE_JSON'; + case TSparkRowSetType.URL_BASED_SET: + return 'EXTERNAL_LINKS'; + default: + return 'FORMAT_UNSPECIFIED'; + } +} diff --git a/lib/telemetry/telemetryUtils.ts b/lib/telemetry/telemetryUtils.ts index 6326aedf..f232af71 100644 --- a/lib/telemetry/telemetryUtils.ts +++ b/lib/telemetry/telemetryUtils.ts @@ -129,6 +129,12 @@ const SECRET_PATTERNS: Array<[RegExp, string]> = [ /\b(token|password|client_secret|refresh_token|access_token|id_token|secret|api[_-]?key|apikey)=[^\s&"']+/gi, '$1=', ], + // OS-username-bearing filesystem paths in error stacks. `/home//`, + // `/Users//`, `C:\Users\\`. Stack traces routinely echo the + // running user's home directory which leaks both the OS account name and + // application directory layout. + [/\/(?:home|Users)\/[^/\s)\]]+\//g, '//'], + [/[A-Za-z]:\\Users\\[^\\\s)\]]+\\/g, '\\'], ]; /** @@ -228,3 +234,33 @@ export function sanitizeProcessName(name: string | undefined): string { const lastSep = Math.max(firstToken.lastIndexOf('/'), firstToken.lastIndexOf('\\')); return lastSep < 0 ? firstToken : firstToken.slice(lastSep + 1); } + +/** + * Run a telemetry emit at a call site, swallowing all exceptions and logging + * at debug level. Replaces the copy-pasted try/catch + getTelemetryEmitter?.() + * scaffold at every emit site. Telemetry must never break the driver. + */ +export function safeEmit( + // The any below is intentional: this helper is consumed by both `IClientContext` + // and the narrower per-call surface used in tests. Keeping it loose avoids + // a dependency cycle through IClientContext / TelemetryEventEmitter. + // eslint-disable-next-line @typescript-eslint/no-explicit-any + context: { getTelemetryEmitter?: () => any; getLogger: () => { log: (level: any, msg: string) => void } }, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + fn: (emitter: any) => void, +): void { + try { + const emitter = context.getTelemetryEmitter?.(); + if (emitter) { + fn(emitter); + } + } catch (err: any) { + try { + // LogLevel.debug = 'debug' — see lib/contracts/IDBSQLLogger.ts. Avoid + // an import here to keep this helper free of cyclic deps. + context.getLogger().log('debug', `Telemetry emit error: ${err?.message ?? err}`); + } catch { + // swallow logger errors too — telemetry never breaks the driver + } + } +} diff --git a/lib/telemetry/types.ts b/lib/telemetry/types.ts index 6a4a25a9..9a422258 100644 --- a/lib/telemetry/types.ts +++ b/lib/telemetry/types.ts @@ -24,6 +24,7 @@ export const DRIVER_NAME = 'nodejs-sql-driver'; */ export enum TelemetryEventType { CONNECTION_OPEN = 'connection.open', + CONNECTION_CLOSE = 'connection.close', STATEMENT_START = 'statement.start', STATEMENT_COMPLETE = 'statement.complete', CLOUDFETCH_CHUNK = 'cloudfetch.chunk', @@ -72,13 +73,24 @@ export interface TelemetryConfiguration { /** TTL in ms after which abandoned statement aggregations are evicted */ statementTtlMs?: number; + + /** + * Maximum wall-clock time `close()` will wait for the final flush HTTP POST + * before abandoning it and returning. Bounds shutdown latency so callers + * doing `await client.close(); process.exit(0)` are not held up by a + * misbehaving telemetry endpoint. + */ + closeTimeoutMs?: number; + + /** Hard cap on per-statement aggregation map size; oldest evicted on overflow. */ + maxStatementMetrics?: number; } /** * Default telemetry configuration values */ export const DEFAULT_TELEMETRY_CONFIG: Readonly> = Object.freeze({ - enabled: false, + enabled: true, // Enabled by default, gated by feature flag batchSize: 100, flushIntervalMs: 5000, maxRetries: 3, @@ -91,6 +103,8 @@ export const DEFAULT_TELEMETRY_CONFIG: Readonly maxPendingMetrics: 500, maxErrorsPerStatement: 50, statementTtlMs: 60 * 60 * 1000, // 1 hour + closeTimeoutMs: 2000, // 2s — caps client.close() shutdown latency + maxStatementMetrics: 5000, // hard cap for the per-statement aggregation map }); /** @@ -178,24 +192,39 @@ export interface TelemetryMetric { /** Workspace ID */ workspaceId?: string; - /** Driver configuration (for connection metrics) */ + /** Driver configuration (included in all metrics for context) */ driverConfig?: DriverConfiguration; /** Execution latency in milliseconds */ latencyMs?: number; + /** Type of operation (SELECT, INSERT, etc.) */ + operationType?: string; + /** Result format (inline, cloudfetch, arrow) */ resultFormat?: string; /** Number of result chunks */ chunkCount?: number; + /** Latency of the first chunk fetch in milliseconds */ + chunkInitialLatencyMs?: number; + + /** Latency of the slowest chunk fetch in milliseconds */ + chunkSlowestLatencyMs?: number; + + /** Sum of all chunk fetch latencies in milliseconds */ + chunkSumLatencyMs?: number; + /** Total bytes downloaded */ bytesDownloaded?: number; /** Number of poll operations */ pollCount?: number; + /** Whether compression was used */ + compressed?: boolean; + /** Error name/type */ errorName?: string; @@ -243,6 +272,9 @@ export interface DriverConfiguration { */ processName: string; + /** Authentication type (pat, external-browser, oauth-m2m, custom) */ + authType: string; + // Feature flags /** Whether CloudFetch is enabled */ cloudFetchEnabled: boolean; @@ -265,6 +297,13 @@ export interface DriverConfiguration { /** Number of concurrent CloudFetch downloads */ cloudFetchConcurrentDownloads: number; + + // Connection parameters for telemetry + /** HTTP path for API calls */ + httpPath?: string; + + /** Whether metric view metadata is enabled */ + enableMetricViewMetadata?: boolean; } /** @@ -298,6 +337,15 @@ export interface StatementMetrics { /** Number of CloudFetch chunks downloaded */ chunkCount: number; + /** Latency of the first chunk fetch in milliseconds */ + chunkInitialLatencyMs?: number; + + /** Latency of the slowest chunk fetch in milliseconds */ + chunkSlowestLatencyMs?: number; + + /** Sum of all chunk fetch latencies in milliseconds */ + chunkSumLatencyMs?: number; + /** Total bytes downloaded */ totalBytesDownloaded: number; diff --git a/lib/telemetry/urlUtils.ts b/lib/telemetry/urlUtils.ts deleted file mode 100644 index 4dd8535e..00000000 --- a/lib/telemetry/urlUtils.ts +++ /dev/null @@ -1,31 +0,0 @@ -/** - * Copyright (c) 2025 Databricks Contributors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -/** - * Build full URL from host and path, handling protocol correctly. - * @param host The hostname (with or without protocol) - * @param path The path to append (should start with /) - * @returns Full URL with protocol - */ -// eslint-disable-next-line import/prefer-default-export -export function buildUrl(host: string, path: string): string { - // Check if host already has protocol - if (host.startsWith('http://') || host.startsWith('https://')) { - return `${host}${path}`; - } - // Add https:// if no protocol present - return `https://${host}${path}`; -} diff --git a/tests/e2e/telemetry/telemetry-integration.test.ts b/tests/e2e/telemetry/telemetry-integration.test.ts new file mode 100644 index 00000000..bdcce4cc --- /dev/null +++ b/tests/e2e/telemetry/telemetry-integration.test.ts @@ -0,0 +1,381 @@ +/** + * Copyright (c) 2025 Databricks Contributors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { expect } from 'chai'; +import sinon from 'sinon'; +import { DBSQLClient } from '../../../lib'; +import config from '../utils/config'; +import FeatureFlagCache from '../../../lib/telemetry/FeatureFlagCache'; +import TelemetryClientProvider from '../../../lib/telemetry/TelemetryClientProvider'; +import TelemetryEventEmitter from '../../../lib/telemetry/TelemetryEventEmitter'; +import MetricsAggregator from '../../../lib/telemetry/MetricsAggregator'; + +describe('Telemetry Integration', () => { + // Reset the process-wide singleton between tests so refcount + cached + // feature flags from one test don't leak into the next. + afterEach(() => { + TelemetryClientProvider.resetInstance(); + sinon.restore(); + }); + + describe('Initialization', () => { + it('should initialize telemetry when telemetryEnabled is true', async function () { + this.timeout(30000); + + const client = new DBSQLClient(); + + // Spy on initialization components + const featureFlagCacheSpy = sinon.spy(FeatureFlagCache.prototype, 'getOrCreateContext'); + const telemetryProviderSpy = sinon.spy(TelemetryClientProvider.prototype, 'getOrCreateClient'); + + try { + await client.connect({ + host: config.host, + path: config.path, + token: config.token, + telemetryEnabled: true, + }); + + // Verify telemetry components were initialized + expect(featureFlagCacheSpy.called).to.be.true; + + await client.close(); + } finally { + featureFlagCacheSpy.restore(); + telemetryProviderSpy.restore(); + } + }); + + it('should not initialize telemetry when telemetryEnabled is false', async function () { + this.timeout(30000); + + const client = new DBSQLClient(); + + const featureFlagCacheSpy = sinon.spy(FeatureFlagCache.prototype, 'getOrCreateContext'); + + try { + await client.connect({ + host: config.host, + path: config.path, + token: config.token, + telemetryEnabled: false, + }); + + // Verify telemetry was not initialized + expect(featureFlagCacheSpy.called).to.be.false; + + await client.close(); + } finally { + featureFlagCacheSpy.restore(); + } + }); + + it('should respect feature flag when telemetry is enabled', async function () { + this.timeout(30000); + + const client = new DBSQLClient(); + + // Stub feature flag to return false + const featureFlagStub = sinon.stub(FeatureFlagCache.prototype, 'isTelemetryEnabled').resolves(false); + + try { + await client.connect({ + host: config.host, + path: config.path, + token: config.token, + telemetryEnabled: true, + }); + + // Verify feature flag was checked + expect(featureFlagStub.called).to.be.true; + + await client.close(); + } finally { + featureFlagStub.restore(); + } + }); + }); + + describe('Reference Counting', () => { + it('should share telemetry client across multiple connections to same host', async function () { + this.timeout(60000); + + const client1 = new DBSQLClient(); + const client2 = new DBSQLClient(); + + const getOrCreateClientSpy = sinon.spy(TelemetryClientProvider.prototype, 'getOrCreateClient'); + const releaseClientSpy = sinon.spy(TelemetryClientProvider.prototype, 'releaseClient'); + + try { + // Enable telemetry for both clients + await client1.connect({ + host: config.host, + path: config.path, + token: config.token, + telemetryEnabled: true, + }); + + await client2.connect({ + host: config.host, + path: config.path, + token: config.token, + telemetryEnabled: true, + }); + + // Both clients should get the same telemetry client for the host + expect(getOrCreateClientSpy.callCount).to.be.at.least(2); + + // Close first client + await client1.close(); + expect(releaseClientSpy.callCount).to.be.at.least(1); + + // Close second client + await client2.close(); + expect(releaseClientSpy.callCount).to.be.at.least(2); + } finally { + getOrCreateClientSpy.restore(); + releaseClientSpy.restore(); + } + }); + + it('should cleanup telemetry on close', async function () { + this.timeout(30000); + + const client = new DBSQLClient(); + + const releaseClientSpy = sinon.spy(TelemetryClientProvider.prototype, 'releaseClient'); + const releaseContextSpy = sinon.spy(FeatureFlagCache.prototype, 'releaseContext'); + const flushSpy = sinon.spy(MetricsAggregator.prototype, 'flush'); + + try { + await client.connect({ + host: config.host, + path: config.path, + token: config.token, + telemetryEnabled: true, + }); + + await client.close(); + + // Verify cleanup was called + expect(releaseClientSpy.called || flushSpy.called || releaseContextSpy.called).to.be.true; + } finally { + releaseClientSpy.restore(); + releaseContextSpy.restore(); + flushSpy.restore(); + } + }); + }); + + describe('Error Handling', () => { + it('should continue driver operation when telemetry initialization fails', async function () { + this.timeout(30000); + + const client = new DBSQLClient(); + + // Stub feature flag to throw an error + const featureFlagStub = sinon + .stub(FeatureFlagCache.prototype, 'isTelemetryEnabled') + .rejects(new Error('Feature flag fetch failed')); + + try { + // Connection should succeed even if telemetry fails + await client.connect({ + host: config.host, + path: config.path, + token: config.token, + telemetryEnabled: true, + }); + + // Should be able to open a session + const session = await client.openSession({ + initialCatalog: config.catalog, + initialSchema: config.schema, + }); + + // Should be able to execute a query + const operation = await session.executeStatement('SELECT 1 AS test'); + const result = await operation.fetchAll(); + + expect(result).to.have.lengthOf(1); + expect(result[0]).to.deep.equal({ test: 1 }); + + await session.close(); + await client.close(); + } finally { + featureFlagStub.restore(); + } + }); + + it('should continue driver operation when feature flag fetch fails', async function () { + this.timeout(30000); + + const client = new DBSQLClient(); + + // Stub getOrCreateContext to throw + const contextStub = sinon + .stub(FeatureFlagCache.prototype, 'getOrCreateContext') + .throws(new Error('Context creation failed')); + + try { + // Connection should succeed even if telemetry fails + await client.connect({ + host: config.host, + path: config.path, + token: config.token, + telemetryEnabled: true, + }); + + // Should be able to open a session + const session = await client.openSession({ + initialCatalog: config.catalog, + initialSchema: config.schema, + }); + + await session.close(); + await client.close(); + } finally { + contextStub.restore(); + } + }); + + it('should not throw exceptions due to telemetry errors', async function () { + this.timeout(30000); + + const client = new DBSQLClient(); + + // Stub multiple telemetry methods to throw + const emitterStub = sinon + .stub(TelemetryEventEmitter.prototype, 'emitConnectionOpen') + .throws(new Error('Emitter failed')); + const aggregatorStub = sinon + .stub(MetricsAggregator.prototype, 'processEvent') + .throws(new Error('Aggregator failed')); + + try { + // Connection should not throw + await client.connect({ + host: config.host, + path: config.path, + token: config.token, + telemetryEnabled: true, + }); + + // Driver operations should work normally + const session = await client.openSession({ + initialCatalog: config.catalog, + initialSchema: config.schema, + }); + + await session.close(); + await client.close(); + } finally { + emitterStub.restore(); + aggregatorStub.restore(); + } + }); + }); + + describe('Configuration', () => { + it('should read telemetry config from ClientConfig', async function () { + this.timeout(30000); + + const client = new DBSQLClient(); + const clientConfig = client.getConfig(); + + // Verify default telemetry config exists + expect(clientConfig).to.have.property('telemetryEnabled'); + expect(clientConfig).to.have.property('telemetryBatchSize'); + expect(clientConfig).to.have.property('telemetryFlushIntervalMs'); + expect(clientConfig).to.have.property('telemetryMaxRetries'); + expect(clientConfig).to.have.property('telemetryAuthenticatedExport'); + expect(clientConfig).to.have.property('telemetryCircuitBreakerThreshold'); + expect(clientConfig).to.have.property('telemetryCircuitBreakerTimeout'); + + // Verify default values. telemetryEnabled defaults to true (gated by + // remote feature flag and DATABRICKS_TELEMETRY_DISABLED env var). + expect(clientConfig.telemetryEnabled).to.equal(true); + expect(clientConfig.telemetryBatchSize).to.equal(100); + expect(clientConfig.telemetryFlushIntervalMs).to.equal(5000); + expect(clientConfig.telemetryMaxRetries).to.equal(3); + expect(clientConfig.telemetryAuthenticatedExport).to.equal(true); + expect(clientConfig.telemetryCircuitBreakerThreshold).to.equal(5); + expect(clientConfig.telemetryCircuitBreakerTimeout).to.equal(60000); + }); + + it('should allow override via ConnectionOptions', async function () { + this.timeout(30000); + + const client = new DBSQLClient(); + + // Default is true; verify explicit override to false. + expect(client.getConfig().telemetryEnabled).to.equal(true); + + try { + await client.connect({ + host: config.host, + path: config.path, + token: config.token, + telemetryEnabled: false, + }); + + // Config should reflect the override + expect(client.getConfig().telemetryEnabled).to.equal(false); + + await client.close(); + } catch (error) { + // Clean up even if test fails + await client.close(); + throw error; + } + }); + }); + + describe('End-to-End Telemetry Flow', () => { + it('should emit events during driver operations when telemetry is enabled', async function () { + this.timeout(30000); + + const client = new DBSQLClient(); + + const emitSpy = sinon.spy(TelemetryEventEmitter.prototype, 'emit'); + + try { + await client.connect({ + host: config.host, + path: config.path, + token: config.token, + telemetryEnabled: true, + }); + + const session = await client.openSession({ + initialCatalog: config.catalog, + initialSchema: config.schema, + }); + + const operation = await session.executeStatement('SELECT 1 AS test'); + await operation.fetchAll(); + + // Events may or may not be emitted depending on feature flag + // But the driver should work regardless + + await session.close(); + await client.close(); + } finally { + emitSpy.restore(); + } + }); + }); +}); diff --git a/tests/unit/.stubs/ClientContextStub.ts b/tests/unit/.stubs/ClientContextStub.ts index 519316ff..5bd48da9 100644 --- a/tests/unit/.stubs/ClientContextStub.ts +++ b/tests/unit/.stubs/ClientContextStub.ts @@ -3,7 +3,10 @@ import IConnectionProvider from '../../../lib/connection/contracts/IConnectionPr import IDriver from '../../../lib/contracts/IDriver'; import IThriftClient from '../../../lib/contracts/IThriftClient'; import IDBSQLLogger from '../../../lib/contracts/IDBSQLLogger'; +import IAuthentication from '../../../lib/connection/contracts/IAuthentication'; import DBSQLClient from '../../../lib/DBSQLClient'; +import TelemetryEventEmitter from '../../../lib/telemetry/TelemetryEventEmitter'; +import MetricsAggregator from '../../../lib/telemetry/MetricsAggregator'; import LoggerStub from './LoggerStub'; import ThriftClientStub from './ThriftClientStub'; @@ -21,6 +24,12 @@ export default class ClientContextStub implements IClientContext { public connectionProvider = new ConnectionProviderStub(); + // Tests can assign a stub emitter / aggregator to assert on emit calls. + // Defaults to undefined so optional-chaining `?.()` no-ops in unrelated tests. + public telemetryEmitter?: TelemetryEventEmitter; + + public telemetryAggregator?: MetricsAggregator; + constructor(configOverrides: Partial = {}) { this.configOverrides = configOverrides; } @@ -48,4 +57,16 @@ export default class ClientContextStub implements IClientContext { public async getDriver(): Promise { return this.driver; } + + public getAuthProvider(): IAuthentication | undefined { + return undefined; + } + + public getTelemetryEmitter(): TelemetryEventEmitter | undefined { + return this.telemetryEmitter; + } + + public getTelemetryAggregator(): MetricsAggregator | undefined { + return this.telemetryAggregator; + } } diff --git a/tests/unit/DBSQLOperation.test.ts b/tests/unit/DBSQLOperation.test.ts index b5f142ba..9ddd669f 100644 --- a/tests/unit/DBSQLOperation.test.ts +++ b/tests/unit/DBSQLOperation.test.ts @@ -1177,4 +1177,81 @@ describe('DBSQLOperation', () => { expect(driver.getResultSetMetadata.callCount).to.equal(1); }); }); + + describe('error-telemetry wrapper (F10/F17)', () => { + function makeContextWithEmitter() { + const context = new ClientContextStub(); + const emitError = sinon.stub(); + context.telemetryEmitter = { + emitError, + emitStatementStart: sinon.stub(), + emitStatementComplete: sinon.stub(), + } as any; + return { context, emitError }; + } + + it('emits ERROR telemetry when cancel rejects', async () => { + const { context, emitError } = makeContextWithEmitter(); + context.driver.cancelOperationResp.status.statusCode = TStatusCode.ERROR_STATUS; + const operation = new DBSQLOperation({ handle: operationHandleStub({ hasResultSet: true }), context }); + + await expectFailure(() => operation.cancel()); + + expect(emitError.calledOnce, 'emitError should fire once on rejected cancel').to.be.true; + expect(emitError.firstCall.args[0].errorName).to.match(/Status\s?Error/); + }); + + it('emits ERROR telemetry when close rejects', async () => { + const { context, emitError } = makeContextWithEmitter(); + context.driver.closeOperationResp.status.statusCode = TStatusCode.ERROR_STATUS; + const operation = new DBSQLOperation({ handle: operationHandleStub({ hasResultSet: true }), context }); + + await expectFailure(() => operation.close()); + + expect(emitError.calledOnce).to.be.true; + }); + + it('emits ERROR telemetry when getMetadata rejects', async () => { + const { context, emitError } = makeContextWithEmitter(); + // Force the driver to fail metadata fetch + sinon.stub(context.driver, 'getResultSetMetadata').rejects(new Error('metadata-failed')); + const operation = new DBSQLOperation({ handle: operationHandleStub({ hasResultSet: true }), context }); + + await expectFailure(() => operation.getMetadata()); + + expect(emitError.calledOnce).to.be.true; + expect(emitError.firstCall.args[0].errorMessage).to.equal('metadata-failed'); + }); + + it('emits ERROR telemetry from finished/getSchema/hasMoreRows when called on a closed op (F10 — newly wrapped)', async () => { + // Smaller invariant: closed-op rejection (synchronous error) goes through + // the new wrappers' telemetry path. Avoids the polling loops that happen + // when state is ERROR_STATE — those would require driver-status mock + // sequencing that isn't worth the complexity for this assertion. + for (const fn of ['finished', 'getSchema', 'hasMoreRows'] as const) { + const { context, emitError } = makeContextWithEmitter(); + const operation = new DBSQLOperation({ handle: operationHandleStub({ hasResultSet: true }), context }); + await operation.close(); + emitError.resetHistory(); // close() also emits telemetry; we only want to assert on the new wrapper + if (fn === 'hasMoreRows') { + // hasMoreRows short-circuits to false on closed without throwing + const r = await operation.hasMoreRows(); + expect(r).to.equal(false); + } else { + await expectFailure(() => (operation as any)[fn]()); + expect(emitError.called, `${fn}() should fire emitError on closed op`).to.be.true; + } + } + }); + + it('does not throw when no telemetry emitter is wired', async () => { + const context = new ClientContextStub(); + // emitter intentionally left undefined + context.driver.cancelOperationResp.status.statusCode = TStatusCode.ERROR_STATUS; + const operation = new DBSQLOperation({ handle: operationHandleStub({ hasResultSet: true }), context }); + + // The cancel still rejects, but no extra error from the missing emitter + await expectFailure(() => operation.cancel()); + }); + }); }); diff --git a/tests/unit/result/RowSetProvider.test.ts b/tests/unit/result/RowSetProvider.test.ts new file mode 100644 index 00000000..12fa3c57 --- /dev/null +++ b/tests/unit/result/RowSetProvider.test.ts @@ -0,0 +1,112 @@ +/** + * Copyright (c) 2025 Databricks Contributors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + */ + +import { expect } from 'chai'; +import sinon from 'sinon'; +import Int64 from 'node-int64'; +import { TOperationHandle, TOperationType, THandleIdentifier } from '../../../thrift/TCLIService_types'; +import RowSetProvider from '../../../lib/result/RowSetProvider'; +import ClientContextStub from '../.stubs/ClientContextStub'; + +function makeOperationHandle(): TOperationHandle { + return { + operationId: { + guid: Buffer.from([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]), + secret: Buffer.from([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]), + } as THandleIdentifier, + operationType: TOperationType.EXECUTE_STATEMENT, + hasResultSet: true, + }; +} + +describe('RowSetProvider', () => { + describe('chunk telemetry emission', () => { + it('emits CLOUDFETCH_CHUNK with monotonic chunkIndex per fetch', async () => { + const context = new ClientContextStub(); + const emitCloudFetchChunk = sinon.stub(); + context.telemetryEmitter = { emitCloudFetchChunk } as any; + + const handle = makeOperationHandle(); + const provider = new RowSetProvider(context, handle, [], false, 'stmt-1'); + + await provider.fetchNext({ limit: 100 }); + await provider.fetchNext({ limit: 100 }); + + expect(emitCloudFetchChunk.calledTwice).to.be.true; + expect(emitCloudFetchChunk.firstCall.args[0].chunkIndex).to.equal(0); + expect(emitCloudFetchChunk.secondCall.args[0].chunkIndex).to.equal(1); + expect(emitCloudFetchChunk.firstCall.args[0].statementId).to.equal('stmt-1'); + }); + + it('does not emit when statementId is undefined', async () => { + const context = new ClientContextStub(); + const emitCloudFetchChunk = sinon.stub(); + context.telemetryEmitter = { emitCloudFetchChunk } as any; + + const provider = new RowSetProvider(context, makeOperationHandle(), [], false /* no statementId */); + + await provider.fetchNext({ limit: 100 }); + + expect(emitCloudFetchChunk.called).to.be.false; + }); + + it('does not emit when telemetry emitter is undefined', async () => { + const context = new ClientContextStub(); + // emitter left undefined → safeEmit short-circuits + + const provider = new RowSetProvider(context, makeOperationHandle(), [], false, 'stmt-1'); + + // No assertion on rejection — should resolve normally without emitter + const result = await provider.fetchNext({ limit: 100 }); + expect(result).to.exist; + }); + + it('swallows emitter exceptions and does not throw to the caller', async () => { + const context = new ClientContextStub(); + const logSpy = sinon.spy(context.logger, 'log'); + const emitCloudFetchChunk = sinon.stub().throws(new Error('boom')); + context.telemetryEmitter = { emitCloudFetchChunk } as any; + + const provider = new RowSetProvider(context, makeOperationHandle(), [], false, 'stmt-1'); + + // Should not throw — telemetry never breaks the driver + const result = await provider.fetchNext({ limit: 100 }); + expect(result).to.exist; + + // Should log at debug + const debugCall = logSpy.getCalls().find((c) => /Telemetry emit error/.test(c.args[1] as string)); + expect(debugCall, 'should log a debug-level emit-error line').to.exist; + }); + + it('sums arrowBatches lengths into the bytes field', async () => { + const context = new ClientContextStub(); + const emitCloudFetchChunk = sinon.stub(); + context.telemetryEmitter = { emitCloudFetchChunk } as any; + + // Override the driver's response with arrow batches + context.driver.fetchResultsResp = { + ...context.driver.fetchResultsResp, + results: { + ...context.driver.fetchResultsResp.results!, + arrowBatches: [ + { batch: Buffer.from(new Uint8Array(100)), rowCount: new Int64(10) }, + { batch: Buffer.from(new Uint8Array(50)), rowCount: new Int64(5) }, + ], + }, + }; + + const provider = new RowSetProvider(context, makeOperationHandle(), [], false, 'stmt-1'); + await provider.fetchNext({ limit: 100 }); + + expect(emitCloudFetchChunk.calledOnce).to.be.true; + expect(emitCloudFetchChunk.firstCall.args[0].bytes).to.equal(150); + }); + }); +}); diff --git a/tests/unit/telemetry/TelemetryClient.test.ts b/tests/unit/telemetry/TelemetryClient.test.ts index 4f4c3f1d..91c4a7a3 100644 --- a/tests/unit/telemetry/TelemetryClient.test.ts +++ b/tests/unit/telemetry/TelemetryClient.test.ts @@ -142,11 +142,121 @@ describe('TelemetryClient', () => { const logSpy = sinon.spy(context.logger, 'log'); const client = new TelemetryClient(context, HOST); - client.close(); + await client.close(); logSpy.getCalls().forEach((call) => { - expect(call.args[0]).to.equal(LogLevel.debug); + expect(call.args[0]).to.satisfy((lvl: string) => lvl === LogLevel.debug || lvl === LogLevel.warn); }); }); }); + + describe('feature-flag context registration (F1)', () => { + it('should register the host with the feature-flag cache on construction', () => { + const context = new ClientContextStub(); + const client = new TelemetryClient(context, HOST); + + // Without F1's wiring, isTelemetryEnabled returns false because the + // contexts map is empty. With F1, the constructor registers the host + // so the cache is ready to fetch the flag. + const cache = client.getFeatureFlagCache(); + // Internal access for assertion only — tests on getInstance/resetInstance + // would otherwise leak across the singleton. + const ctx = (cache as any).contexts.get(HOST); + expect(ctx, 'context should exist after TelemetryClient construction').to.exist; + expect(ctx.refCount, 'refCount should be 1 after registration').to.equal(1); + }); + + it('should release the feature-flag context on close', async () => { + const context = new ClientContextStub(); + const client = new TelemetryClient(context, HOST); + const cache = client.getFeatureFlagCache(); + + await client.close(); + + const ctx = (cache as any).contexts.get(HOST); + expect(ctx, 'context should be removed on close (refCount → 0)').to.be.undefined; + }); + }); + + describe('multi-context FIFO', () => { + it('registerContext appends contexts in registration order', () => { + const ctxA = new ClientContextStub(); + const ctxB = new ClientContextStub(); + const client = new TelemetryClient(ctxA, HOST); + + client.registerContext(ctxB); + + const internal = client as any; + expect(internal.contexts).to.have.lengthOf(2); + expect(internal.contexts[0]).to.equal(ctxA); + expect(internal.contexts[1]).to.equal(ctxB); + }); + + it('registerContext is idempotent for the same context', () => { + const ctxA = new ClientContextStub(); + const client = new TelemetryClient(ctxA, HOST); + + client.registerContext(ctxA); + client.registerContext(ctxA); + + expect((client as any).contexts).to.have.lengthOf(1); + }); + + it('unregisterContext removes the context', () => { + const ctxA = new ClientContextStub(); + const ctxB = new ClientContextStub(); + const client = new TelemetryClient(ctxA, HOST); + client.registerContext(ctxB); + + client.unregisterContext(ctxA); + + const internal = client as any; + expect(internal.contexts).to.have.lengthOf(1); + expect(internal.contexts[0]).to.equal(ctxB); + }); + + it('warns when a registered context has divergent telemetry config (F12)', () => { + const ctxA = new ClientContextStub({ telemetryAuthenticatedExport: true }); + const ctxB = new ClientContextStub({ telemetryAuthenticatedExport: false }); + const client = new TelemetryClient(ctxA, HOST); + const logSpy = sinon.spy(ctxA.logger, 'log'); + + client.registerContext(ctxB); + + const warnCall = logSpy + .getCalls() + .find((c) => c.args[0] === LogLevel.warn && /telemetry settings .* differ/.test(c.args[1] as string)); + expect(warnCall, 'should warn about divergent telemetryAuthenticatedExport').to.exist; + }); + }); + + describe('async close()', () => { + it('returns a Promise that resolves only after aggregator close', async () => { + const context = new ClientContextStub(); + const client = new TelemetryClient(context, HOST); + const aggregator = client.getAggregator(); + const aggCloseStub = sinon.stub(aggregator, 'close').callsFake( + () => + new Promise((resolve) => { + setTimeout(resolve, 5); + }), + ); + + const closePromise = client.close(); + expect(client.isClosed(), 'closed flag is set synchronously').to.be.true; + expect(aggCloseStub.calledOnce).to.be.true; + await closePromise; + }); + + it('is idempotent — second close awaits without re-running aggregator close', async () => { + const context = new ClientContextStub(); + const client = new TelemetryClient(context, HOST); + const aggCloseStub = sinon.stub(client.getAggregator(), 'close').resolves(); + + await client.close(); + await client.close(); + + expect(aggCloseStub.calledOnce, 'aggregator.close should run exactly once').to.be.true; + }); + }); }); diff --git a/tests/unit/telemetry/TelemetryClientProvider.test.ts b/tests/unit/telemetry/TelemetryClientProvider.test.ts index 59ae3b99..00f638fe 100644 --- a/tests/unit/telemetry/TelemetryClientProvider.test.ts +++ b/tests/unit/telemetry/TelemetryClientProvider.test.ts @@ -28,28 +28,19 @@ describe('TelemetryClientProvider', () => { describe('Constructor', () => { it('should create provider with empty client map', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); expect(provider.getActiveClients().size).to.equal(0); }); - - it('should log creation at debug level', () => { - const context = new ClientContextStub(); - const logSpy = sinon.spy(context.logger, 'log'); - - new TelemetryClientProvider(context); - - expect(logSpy.calledWith(LogLevel.debug, sinon.match(/created.*telemetryclientprovider/i))).to.be.true; - }); }); describe('getOrCreateClient', () => { it('should create one client per host', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const client1 = provider.getOrCreateClient(HOST1); - const client2 = provider.getOrCreateClient(HOST2); + const client1 = provider.getOrCreateClient(context, HOST1); + const client2 = provider.getOrCreateClient(context, HOST2); expect(client1).to.be.instanceOf(TelemetryClient); expect(client2).to.be.instanceOf(TelemetryClient); @@ -59,11 +50,11 @@ describe('TelemetryClientProvider', () => { it('should share client across multiple connections to same host', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const client1 = provider.getOrCreateClient(HOST1); - const client2 = provider.getOrCreateClient(HOST1); - const client3 = provider.getOrCreateClient(HOST1); + const client1 = provider.getOrCreateClient(context, HOST1); + const client2 = provider.getOrCreateClient(context, HOST1); + const client3 = provider.getOrCreateClient(context, HOST1); expect(client1).to.equal(client2); expect(client2).to.equal(client3); @@ -72,34 +63,34 @@ describe('TelemetryClientProvider', () => { it('should increment reference count on each call', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - provider.getOrCreateClient(HOST1); + provider.getOrCreateClient(context, HOST1); expect(provider.getRefCount(HOST1)).to.equal(1); - provider.getOrCreateClient(HOST1); + provider.getOrCreateClient(context, HOST1); expect(provider.getRefCount(HOST1)).to.equal(2); - provider.getOrCreateClient(HOST1); + provider.getOrCreateClient(context, HOST1); expect(provider.getRefCount(HOST1)).to.equal(3); }); it('should log client creation at debug level', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); const logSpy = sinon.spy(context.logger, 'log'); - provider.getOrCreateClient(HOST1); + provider.getOrCreateClient(context, HOST1); expect(logSpy.calledWith(LogLevel.debug, sinon.match(/created new telemetryclient/i))).to.be.true; }); it('should log reference count at debug level', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); const logSpy = sinon.spy(context.logger, 'log'); - provider.getOrCreateClient(HOST1); + provider.getOrCreateClient(context, HOST1); expect(logSpy.calledWith(LogLevel.debug, sinon.match(/reference count for/i).and(sinon.match(/: 1$/)))).to.be .true; @@ -107,9 +98,9 @@ describe('TelemetryClientProvider', () => { it('should pass context to TelemetryClient', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const client = provider.getOrCreateClient(HOST1); + const client = provider.getOrCreateClient(context, HOST1); expect(client.getHost()).to.equal(HOST1); }); @@ -118,28 +109,28 @@ describe('TelemetryClientProvider', () => { describe('releaseClient', () => { it('should decrement reference count on release', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - provider.getOrCreateClient(HOST1); - provider.getOrCreateClient(HOST1); - provider.getOrCreateClient(HOST1); + provider.getOrCreateClient(context, HOST1); + provider.getOrCreateClient(context, HOST1); + provider.getOrCreateClient(context, HOST1); expect(provider.getRefCount(HOST1)).to.equal(3); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); expect(provider.getRefCount(HOST1)).to.equal(2); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); expect(provider.getRefCount(HOST1)).to.equal(1); }); it('should close client when reference count reaches zero', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const client = provider.getOrCreateClient(HOST1); + const client = provider.getOrCreateClient(context, HOST1); const closeSpy = sinon.spy(client, 'close'); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); expect(closeSpy.calledOnce).to.be.true; expect(client.isClosed()).to.be.true; @@ -147,12 +138,12 @@ describe('TelemetryClientProvider', () => { it('should remove client from map when reference count reaches zero', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - provider.getOrCreateClient(HOST1); + provider.getOrCreateClient(context, HOST1); expect(provider.getActiveClients().size).to.equal(1); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); expect(provider.getActiveClients().size).to.equal(0); expect(provider.getRefCount(HOST1)).to.equal(0); @@ -160,14 +151,14 @@ describe('TelemetryClientProvider', () => { it('should NOT close client while other connections exist', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const client = provider.getOrCreateClient(HOST1); - provider.getOrCreateClient(HOST1); - provider.getOrCreateClient(HOST1); + const client = provider.getOrCreateClient(context, HOST1); + provider.getOrCreateClient(context, HOST1); + provider.getOrCreateClient(context, HOST1); const closeSpy = sinon.spy(client, 'close'); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); expect(closeSpy.called).to.be.false; expect(client.isClosed()).to.be.false; @@ -176,23 +167,23 @@ describe('TelemetryClientProvider', () => { it('should handle releasing non-existent client gracefully', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); const logSpy = sinon.spy(context.logger, 'log'); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); expect(logSpy.calledWith(LogLevel.debug, sinon.match(/no telemetryclient found/i))).to.be.true; }); it('should log reference count decrease at debug level', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); const logSpy = sinon.spy(context.logger, 'log'); - provider.getOrCreateClient(HOST1); - provider.getOrCreateClient(HOST1); + provider.getOrCreateClient(context, HOST1); + provider.getOrCreateClient(context, HOST1); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); expect(logSpy.calledWith(LogLevel.debug, sinon.match(/reference count for/i).and(sinon.match(/: 1$/)))).to.be .true; @@ -200,25 +191,25 @@ describe('TelemetryClientProvider', () => { it('should log client closure at debug level', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); const logSpy = sinon.spy(context.logger, 'log'); - provider.getOrCreateClient(HOST1); - provider.releaseClient(HOST1); + provider.getOrCreateClient(context, HOST1); + await provider.releaseClient(context, HOST1); expect(logSpy.calledWith(LogLevel.debug, sinon.match(/closed and removed telemetryclient/i))).to.be.true; }); it('should swallow Error during client closure', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const client = provider.getOrCreateClient(HOST1); + const client = provider.getOrCreateClient(context, HOST1); const error = new Error('Close error'); sinon.stub(client, 'close').throws(error); const logSpy = sinon.spy(context.logger, 'log'); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); expect( logSpy.calledWith( @@ -230,9 +221,9 @@ describe('TelemetryClientProvider', () => { it('should swallow non-Error throws during client closure', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const client = provider.getOrCreateClient(HOST1); + const client = provider.getOrCreateClient(context, HOST1); // Non-Error throws — string, null, undefined — must not escape the catch. sinon.stub(client, 'close').callsFake(() => { // eslint-disable-next-line no-throw-literal @@ -240,7 +231,7 @@ describe('TelemetryClientProvider', () => { }); const logSpy = sinon.spy(context.logger, 'log'); - expect(() => provider.releaseClient(HOST1)).to.not.throw(); + expect(() => provider.releaseClient(context, HOST1)).to.not.throw(); expect( logSpy.calledWith( LogLevel.debug, @@ -249,29 +240,29 @@ describe('TelemetryClientProvider', () => { ).to.be.true; }); - it('should not throw or corrupt state on double-release', () => { + it('should not throw or corrupt state on double-release', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); // One get, two releases — the second must not throw and must not // leave the provider in a state where refCount is negative. - provider.getOrCreateClient(HOST1); - provider.releaseClient(HOST1); + provider.getOrCreateClient(context, HOST1); + await provider.releaseClient(context, HOST1); - expect(() => provider.releaseClient(HOST1)).to.not.throw(); + expect(() => provider.releaseClient(context, HOST1)).to.not.throw(); expect(provider.getRefCount(HOST1)).to.equal(0); expect(provider.getActiveClients().size).to.equal(0); }); - it('should return a fresh non-closed client after full release', () => { + it('should return a fresh non-closed client after full release', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const first = provider.getOrCreateClient(HOST1); - provider.releaseClient(HOST1); + const first = provider.getOrCreateClient(context, HOST1); + await provider.releaseClient(context, HOST1); expect(first.isClosed()).to.be.true; - const second = provider.getOrCreateClient(HOST1); + const second = provider.getOrCreateClient(context, HOST1); expect(second).to.not.equal(first); expect(second.isClosed()).to.be.false; expect(provider.getRefCount(HOST1)).to.equal(1); @@ -281,13 +272,13 @@ describe('TelemetryClientProvider', () => { describe('Host normalization', () => { it('should treat scheme, case, port and trailing slash as the same host', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const a = provider.getOrCreateClient('workspace.cloud.databricks.com'); - const b = provider.getOrCreateClient('https://workspace.cloud.databricks.com'); - const c = provider.getOrCreateClient('https://WorkSpace.CLOUD.databricks.com/'); - const d = provider.getOrCreateClient('workspace.cloud.databricks.com:443'); - const e = provider.getOrCreateClient(' workspace.cloud.databricks.com. '); + const a = provider.getOrCreateClient(context, 'workspace.cloud.databricks.com'); + const b = provider.getOrCreateClient(context, 'https://workspace.cloud.databricks.com'); + const c = provider.getOrCreateClient(context, 'https://WorkSpace.CLOUD.databricks.com/'); + const d = provider.getOrCreateClient(context, 'workspace.cloud.databricks.com:443'); + const e = provider.getOrCreateClient(context, ' workspace.cloud.databricks.com. '); expect(a).to.equal(b); expect(a).to.equal(c); @@ -297,12 +288,12 @@ describe('TelemetryClientProvider', () => { expect(provider.getRefCount('workspace.cloud.databricks.com')).to.equal(5); }); - it('should release under an alias correctly', () => { + it('should release under an alias correctly', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - provider.getOrCreateClient('example.com'); - provider.releaseClient('HTTPS://Example.COM/'); + provider.getOrCreateClient(context, 'example.com'); + await provider.releaseClient(context, 'HTTPS://Example.COM/'); expect(provider.getRefCount('example.com')).to.equal(0); expect(provider.getActiveClients().size).to.equal(0); @@ -312,18 +303,18 @@ describe('TelemetryClientProvider', () => { describe('Reference counting', () => { it('should track reference counts independently per host', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - provider.getOrCreateClient(HOST1); - provider.getOrCreateClient(HOST1); - provider.getOrCreateClient(HOST2); - provider.getOrCreateClient(HOST2); - provider.getOrCreateClient(HOST2); + provider.getOrCreateClient(context, HOST1); + provider.getOrCreateClient(context, HOST1); + provider.getOrCreateClient(context, HOST2); + provider.getOrCreateClient(context, HOST2); + provider.getOrCreateClient(context, HOST2); expect(provider.getRefCount(HOST1)).to.equal(2); expect(provider.getRefCount(HOST2)).to.equal(3); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); expect(provider.getRefCount(HOST1)).to.equal(1); expect(provider.getRefCount(HOST2)).to.equal(3); @@ -331,21 +322,21 @@ describe('TelemetryClientProvider', () => { it('should close only last connection for each host', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const client1 = provider.getOrCreateClient(HOST1); - provider.getOrCreateClient(HOST1); - const client2 = provider.getOrCreateClient(HOST2); + const client1 = provider.getOrCreateClient(context, HOST1); + provider.getOrCreateClient(context, HOST1); + const client2 = provider.getOrCreateClient(context, HOST2); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); expect(client1.isClosed()).to.be.false; expect(provider.getActiveClients().size).to.equal(2); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); expect(client1.isClosed()).to.be.true; expect(provider.getActiveClients().size).to.equal(1); - provider.releaseClient(HOST2); + await provider.releaseClient(context, HOST2); expect(client2.isClosed()).to.be.true; expect(provider.getActiveClients().size).to.equal(0); }); @@ -354,10 +345,10 @@ describe('TelemetryClientProvider', () => { describe('Per-host isolation', () => { it('should isolate clients by host', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const client1 = provider.getOrCreateClient(HOST1); - const client2 = provider.getOrCreateClient(HOST2); + const client1 = provider.getOrCreateClient(context, HOST1); + const client2 = provider.getOrCreateClient(context, HOST2); expect(client1.getHost()).to.equal(HOST1); expect(client2.getHost()).to.equal(HOST2); @@ -366,12 +357,12 @@ describe('TelemetryClientProvider', () => { it('should allow closing one host without affecting others', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const client1 = provider.getOrCreateClient(HOST1); - const client2 = provider.getOrCreateClient(HOST2); + const client1 = provider.getOrCreateClient(context, HOST1); + const client2 = provider.getOrCreateClient(context, HOST2); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); expect(client1.isClosed()).to.be.true; expect(client2.isClosed()).to.be.false; @@ -382,19 +373,19 @@ describe('TelemetryClientProvider', () => { describe('getRefCount', () => { it('should return 0 for non-existent host', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); expect(provider.getRefCount(HOST1)).to.equal(0); }); it('should return current reference count for existing host', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - provider.getOrCreateClient(HOST1); + provider.getOrCreateClient(context, HOST1); expect(provider.getRefCount(HOST1)).to.equal(1); - provider.getOrCreateClient(HOST1); + provider.getOrCreateClient(context, HOST1); expect(provider.getRefCount(HOST1)).to.equal(2); }); }); @@ -402,7 +393,7 @@ describe('TelemetryClientProvider', () => { describe('getActiveClients', () => { it('should return empty map initially', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); const clients = provider.getActiveClients(); @@ -411,10 +402,10 @@ describe('TelemetryClientProvider', () => { it('should return all active clients', () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - const client1 = provider.getOrCreateClient(HOST1); - const client2 = provider.getOrCreateClient(HOST2); + const client1 = provider.getOrCreateClient(context, HOST1); + const client2 = provider.getOrCreateClient(context, HOST2); const clients = provider.getActiveClients(); @@ -425,12 +416,12 @@ describe('TelemetryClientProvider', () => { it('should not include closed clients', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - provider.getOrCreateClient(HOST1); - provider.getOrCreateClient(HOST2); + provider.getOrCreateClient(context, HOST1); + provider.getOrCreateClient(context, HOST2); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); const clients = provider.getActiveClients(); @@ -444,9 +435,9 @@ describe('TelemetryClientProvider', () => { it('should use logger from context for all logging', () => { const context = new ClientContextStub(); const logSpy = sinon.spy(context.logger, 'log'); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); - provider.getOrCreateClient(HOST1); + provider.getOrCreateClient(context, HOST1); expect(logSpy.called).to.be.true; logSpy.getCalls().forEach((call) => { @@ -456,13 +447,13 @@ describe('TelemetryClientProvider', () => { it('should log close errors at debug level', async () => { const context = new ClientContextStub(); - const provider = new TelemetryClientProvider(context); + const provider = new TelemetryClientProvider(); const logSpy = sinon.spy(context.logger, 'log'); - const client = provider.getOrCreateClient(HOST1); + const client = provider.getOrCreateClient(context, HOST1); sinon.stub(client, 'close').throws(new Error('Test error')); - provider.releaseClient(HOST1); + await provider.releaseClient(context, HOST1); const errorLogs = logSpy.getCalls().filter((call) => /error releasing/i.test(String(call.args[1]))); expect(errorLogs.length).to.be.greaterThan(0); diff --git a/tests/unit/telemetry/TelemetryEventEmitter.test.ts b/tests/unit/telemetry/TelemetryEventEmitter.test.ts index 6fc29107..c1f86802 100644 --- a/tests/unit/telemetry/TelemetryEventEmitter.test.ts +++ b/tests/unit/telemetry/TelemetryEventEmitter.test.ts @@ -17,187 +17,723 @@ import { expect } from 'chai'; import sinon from 'sinon'; import TelemetryEventEmitter from '../../../lib/telemetry/TelemetryEventEmitter'; -import { TelemetryEventType } from '../../../lib/telemetry/types'; -import ClientContextStub from '../.stubs/ClientContextStub'; -import { LogLevel } from '../../../lib/contracts/IDBSQLLogger'; - -function makeEmitter(enabled: boolean): TelemetryEventEmitter { - const context = new ClientContextStub({ telemetryEnabled: enabled } as any); - return new TelemetryEventEmitter(context); -} +import { TelemetryEventType, TelemetryEvent, DriverConfiguration } from '../../../lib/telemetry/types'; +import IClientContext from '../../../lib/contracts/IClientContext'; +import IDBSQLLogger, { LogLevel } from '../../../lib/contracts/IDBSQLLogger'; describe('TelemetryEventEmitter', () => { - describe('when telemetry is disabled', () => { - it('should not emit any events', () => { - const emitter = makeEmitter(false); - const listener = sinon.stub(); - emitter.on(TelemetryEventType.CONNECTION_OPEN, listener); + let context: IClientContext; + let logger: IDBSQLLogger; + let emitter: TelemetryEventEmitter; + + beforeEach(() => { + logger = { + log: sinon.stub(), + }; + + context = { + getLogger: () => logger, + getConfig: () => ({ + telemetryEnabled: true, + directResultsDefaultMaxRows: 10000, + fetchChunkDefaultMaxRows: 100000, + socketTimeout: 900000, + retryMaxAttempts: 30, + retriesTimeout: 900000, + retryDelayMin: 1000, + retryDelayMax: 30000, + useCloudFetch: true, + cloudFetchConcurrentDownloads: 10, + cloudFetchSpeedThresholdMBps: 0, + useLZ4Compression: true, + }), + } as any; + + emitter = new TelemetryEventEmitter(context); + }); - emitter.emitConnectionOpen({ sessionId: 's1', workspaceId: 'w1', driverConfig: {} as any }); + afterEach(() => { + sinon.restore(); + }); + + describe('constructor', () => { + it('should create instance with telemetry enabled', () => { + expect(emitter).to.be.instanceOf(TelemetryEventEmitter); + }); - expect(listener.called).to.be.false; + it('should create instance with telemetry disabled', () => { + const disabledContext = { + getLogger: () => logger, + getConfig: () => ({ + telemetryEnabled: false, + directResultsDefaultMaxRows: 10000, + fetchChunkDefaultMaxRows: 100000, + socketTimeout: 900000, + retryMaxAttempts: 30, + retriesTimeout: 900000, + retryDelayMin: 1000, + retryDelayMax: 30000, + useCloudFetch: true, + cloudFetchConcurrentDownloads: 10, + cloudFetchSpeedThresholdMBps: 0, + useLZ4Compression: true, + }), + } as any; + + const disabledEmitter = new TelemetryEventEmitter(disabledContext); + expect(disabledEmitter).to.be.instanceOf(TelemetryEventEmitter); }); - it('should not emit statement start', () => { - const emitter = makeEmitter(false); - const listener = sinon.stub(); - emitter.on(TelemetryEventType.STATEMENT_START, listener); + it('should default to disabled when telemetryEnabled is undefined', () => { + const defaultContext = { + getLogger: () => logger, + getConfig: () => ({ + directResultsDefaultMaxRows: 10000, + fetchChunkDefaultMaxRows: 100000, + socketTimeout: 900000, + retryMaxAttempts: 30, + retriesTimeout: 900000, + retryDelayMin: 1000, + retryDelayMax: 30000, + useCloudFetch: true, + cloudFetchConcurrentDownloads: 10, + cloudFetchSpeedThresholdMBps: 0, + useLZ4Compression: true, + }), + } as any; + + const defaultEmitter = new TelemetryEventEmitter(defaultContext); + expect(defaultEmitter).to.be.instanceOf(TelemetryEventEmitter); + }); + }); - emitter.emitStatementStart({ statementId: 'st1', sessionId: 's1' }); + describe('emitConnectionOpen', () => { + it('should emit connection.open event with correct data', (done) => { + const driverConfig: DriverConfiguration = { + driverVersion: '1.0.0', + driverName: 'databricks-sql-nodejs', + nodeVersion: process.version, + platform: process.platform, + osVersion: 'test-os', + osArch: 'x64', + runtimeVendor: 'Node.js Foundation', + localeName: 'en_US', + charSetEncoding: 'UTF-8', + processName: 'node', + authType: 'pat', + cloudFetchEnabled: true, + lz4Enabled: true, + arrowEnabled: false, + directResultsEnabled: true, + socketTimeout: 900000, + retryMaxAttempts: 30, + cloudFetchConcurrentDownloads: 10, + }; + + emitter.on(TelemetryEventType.CONNECTION_OPEN, (event: TelemetryEvent) => { + expect(event.eventType).to.equal(TelemetryEventType.CONNECTION_OPEN); + expect(event.sessionId).to.equal('session-123'); + expect(event.workspaceId).to.equal('workspace-456'); + expect(event.driverConfig).to.deep.equal(driverConfig); + expect(event.timestamp).to.be.a('number'); + done(); + }); - expect(listener.called).to.be.false; + emitter.emitConnectionOpen({ + sessionId: 'session-123', + workspaceId: 'workspace-456', + driverConfig, + latencyMs: 100, + }); }); - it('should not emit error events', () => { - const emitter = makeEmitter(false); - const listener = sinon.stub(); - emitter.on(TelemetryEventType.ERROR, listener); + it('should not emit when telemetry is disabled', () => { + const disabledContext = { + getLogger: () => logger, + getConfig: () => ({ + telemetryEnabled: false, + directResultsDefaultMaxRows: 10000, + fetchChunkDefaultMaxRows: 100000, + socketTimeout: 900000, + retryMaxAttempts: 30, + retriesTimeout: 900000, + retryDelayMin: 1000, + retryDelayMax: 30000, + useCloudFetch: true, + cloudFetchConcurrentDownloads: 10, + cloudFetchSpeedThresholdMBps: 0, + useLZ4Compression: true, + }), + } as any; + + const disabledEmitter = new TelemetryEventEmitter(disabledContext); + let eventEmitted = false; + + disabledEmitter.on(TelemetryEventType.CONNECTION_OPEN, () => { + eventEmitted = true; + }); - emitter.emitError({ errorName: 'SomeError', errorMessage: 'msg', isTerminal: false }); + disabledEmitter.emitConnectionOpen({ + sessionId: 'session-123', + workspaceId: 'workspace-456', + driverConfig: {} as DriverConfiguration, + latencyMs: 100, + }); - expect(listener.called).to.be.false; + expect(eventEmitted).to.be.false; }); - }); - describe('emitConnectionOpen()', () => { - it('should emit a CONNECTION_OPEN event with correct fields', () => { - const emitter = makeEmitter(true); - const listener = sinon.stub(); - emitter.on(TelemetryEventType.CONNECTION_OPEN, listener); + it('should swallow exceptions and log at debug level', () => { + // Force an exception by emitting before adding any listeners + // Then make emit throw by adding a throwing listener + emitter.on(TelemetryEventType.CONNECTION_OPEN, () => { + throw new Error('Test error'); + }); - emitter.emitConnectionOpen({ sessionId: 's1', workspaceId: 'w1', driverConfig: {} as any }); + emitter.emitConnectionOpen({ + sessionId: 'session-123', + workspaceId: 'workspace-456', + driverConfig: {} as DriverConfiguration, + latencyMs: 100, + }); - expect(listener.calledOnce).to.be.true; - const event = listener.firstCall.args[0]; - expect(event.eventType).to.equal(TelemetryEventType.CONNECTION_OPEN); - expect(event.sessionId).to.equal('s1'); - expect(event.workspaceId).to.equal('w1'); - expect(event.timestamp).to.be.a('number'); + expect((logger.log as sinon.SinonStub).calledWith(LogLevel.debug)).to.be.true; + expect((logger.log as sinon.SinonStub).args[0][1]).to.include('Error emitting connection event'); }); - it('should swallow and log exceptions from listeners', () => { - const context = new ClientContextStub({ telemetryEnabled: true } as any); - const logSpy = sinon.spy(context.logger, 'log'); - const emitter = new TelemetryEventEmitter(context); - + it('should not log at warn or error level', () => { emitter.on(TelemetryEventType.CONNECTION_OPEN, () => { - throw new Error('listener boom'); + throw new Error('Test error'); }); - expect(() => - emitter.emitConnectionOpen({ sessionId: 's1', workspaceId: 'w1', driverConfig: {} as any }), - ).to.not.throw(); - expect(logSpy.calledWith(LogLevel.debug, sinon.match(/listener boom/))).to.be.true; + emitter.emitConnectionOpen({ + sessionId: 'session-123', + workspaceId: 'workspace-456', + driverConfig: {} as DriverConfiguration, + latencyMs: 100, + }); - logSpy.restore(); + const logStub = logger.log as sinon.SinonStub; + for (let i = 0; i < logStub.callCount; i++) { + const level = logStub.args[i][0]; + expect(level).to.not.equal(LogLevel.warn); + expect(level).to.not.equal(LogLevel.error); + } }); }); - describe('emitStatementStart()', () => { - it('should emit a STATEMENT_START event with correct fields', () => { - const emitter = makeEmitter(true); - const listener = sinon.stub(); - emitter.on(TelemetryEventType.STATEMENT_START, listener); + describe('emitStatementStart', () => { + it('should emit statement.start event with correct data', (done) => { + emitter.on(TelemetryEventType.STATEMENT_START, (event: TelemetryEvent) => { + expect(event.eventType).to.equal(TelemetryEventType.STATEMENT_START); + expect(event.statementId).to.equal('stmt-789'); + expect(event.sessionId).to.equal('session-123'); + expect(event.operationType).to.equal('SELECT'); + expect(event.timestamp).to.be.a('number'); + done(); + }); + + emitter.emitStatementStart({ + statementId: 'stmt-789', + sessionId: 'session-123', + operationType: 'SELECT', + }); + }); + + it('should emit without operationType', (done) => { + emitter.on(TelemetryEventType.STATEMENT_START, (event: TelemetryEvent) => { + expect(event.eventType).to.equal(TelemetryEventType.STATEMENT_START); + expect(event.statementId).to.equal('stmt-789'); + expect(event.sessionId).to.equal('session-123'); + expect(event.operationType).to.be.undefined; + done(); + }); + + emitter.emitStatementStart({ + statementId: 'stmt-789', + sessionId: 'session-123', + }); + }); + + it('should not emit when telemetry is disabled', () => { + const disabledContext = { + getLogger: () => logger, + getConfig: () => ({ telemetryEnabled: false }), + } as any; + + const disabledEmitter = new TelemetryEventEmitter(disabledContext); + let eventEmitted = false; + + disabledEmitter.on(TelemetryEventType.STATEMENT_START, () => { + eventEmitted = true; + }); + + disabledEmitter.emitStatementStart({ + statementId: 'stmt-789', + sessionId: 'session-123', + }); + + expect(eventEmitted).to.be.false; + }); + + it('should swallow exceptions and log at debug level', () => { + emitter.on(TelemetryEventType.STATEMENT_START, () => { + throw new Error('Test error'); + }); - emitter.emitStatementStart({ statementId: 'st1', sessionId: 's1', operationType: 'SELECT' }); + emitter.emitStatementStart({ + statementId: 'stmt-789', + sessionId: 'session-123', + }); - expect(listener.calledOnce).to.be.true; - const event = listener.firstCall.args[0]; - expect(event.eventType).to.equal(TelemetryEventType.STATEMENT_START); - expect(event.statementId).to.equal('st1'); - expect(event.operationType).to.equal('SELECT'); + expect((logger.log as sinon.SinonStub).calledWith(LogLevel.debug)).to.be.true; + expect((logger.log as sinon.SinonStub).args[0][1]).to.include('Error emitting statement start'); }); }); - describe('emitStatementComplete()', () => { - it('should emit a STATEMENT_COMPLETE event with correct fields', () => { - const emitter = makeEmitter(true); - const listener = sinon.stub(); - emitter.on(TelemetryEventType.STATEMENT_COMPLETE, listener); + describe('emitStatementComplete', () => { + it('should emit statement.complete event with all data fields', (done) => { + emitter.on(TelemetryEventType.STATEMENT_COMPLETE, (event: TelemetryEvent) => { + expect(event.eventType).to.equal(TelemetryEventType.STATEMENT_COMPLETE); + expect(event.statementId).to.equal('stmt-789'); + expect(event.sessionId).to.equal('session-123'); + expect(event.latencyMs).to.equal(1500); + expect(event.resultFormat).to.equal('cloudfetch'); + expect(event.chunkCount).to.equal(5); + expect(event.bytesDownloaded).to.equal(1024000); + expect(event.pollCount).to.equal(3); + expect(event.timestamp).to.be.a('number'); + done(); + }); emitter.emitStatementComplete({ - statementId: 'st1', - sessionId: 's1', - latencyMs: 123, - resultFormat: 'arrow', - chunkCount: 2, - bytesDownloaded: 1024, + statementId: 'stmt-789', + sessionId: 'session-123', + latencyMs: 1500, + resultFormat: 'cloudfetch', + chunkCount: 5, + bytesDownloaded: 1024000, pollCount: 3, }); + }); + + it('should emit with minimal data', (done) => { + emitter.on(TelemetryEventType.STATEMENT_COMPLETE, (event: TelemetryEvent) => { + expect(event.eventType).to.equal(TelemetryEventType.STATEMENT_COMPLETE); + expect(event.statementId).to.equal('stmt-789'); + expect(event.sessionId).to.equal('session-123'); + expect(event.latencyMs).to.be.undefined; + expect(event.resultFormat).to.be.undefined; + done(); + }); + + emitter.emitStatementComplete({ + statementId: 'stmt-789', + sessionId: 'session-123', + }); + }); + + it('should not emit when telemetry is disabled', () => { + const disabledContext = { + getLogger: () => logger, + getConfig: () => ({ telemetryEnabled: false }), + } as any; + + const disabledEmitter = new TelemetryEventEmitter(disabledContext); + let eventEmitted = false; + + disabledEmitter.on(TelemetryEventType.STATEMENT_COMPLETE, () => { + eventEmitted = true; + }); + + disabledEmitter.emitStatementComplete({ + statementId: 'stmt-789', + sessionId: 'session-123', + }); + + expect(eventEmitted).to.be.false; + }); + + it('should swallow exceptions and log at debug level', () => { + emitter.on(TelemetryEventType.STATEMENT_COMPLETE, () => { + throw new Error('Test error'); + }); - expect(listener.calledOnce).to.be.true; - const event = listener.firstCall.args[0]; - expect(event.eventType).to.equal(TelemetryEventType.STATEMENT_COMPLETE); - expect(event.latencyMs).to.equal(123); - expect(event.chunkCount).to.equal(2); + emitter.emitStatementComplete({ + statementId: 'stmt-789', + sessionId: 'session-123', + }); + + expect((logger.log as sinon.SinonStub).calledWith(LogLevel.debug)).to.be.true; + expect((logger.log as sinon.SinonStub).args[0][1]).to.include('Error emitting statement complete'); }); }); - describe('emitCloudFetchChunk()', () => { - it('should emit a CLOUDFETCH_CHUNK event with correct fields', () => { - const emitter = makeEmitter(true); - const listener = sinon.stub(); - emitter.on(TelemetryEventType.CLOUDFETCH_CHUNK, listener); + describe('emitCloudFetchChunk', () => { + it('should emit cloudfetch.chunk event with correct data', (done) => { + emitter.on(TelemetryEventType.CLOUDFETCH_CHUNK, (event: TelemetryEvent) => { + expect(event.eventType).to.equal(TelemetryEventType.CLOUDFETCH_CHUNK); + expect(event.statementId).to.equal('stmt-789'); + expect(event.chunkIndex).to.equal(2); + expect(event.latencyMs).to.equal(250); + expect(event.bytes).to.equal(204800); + expect(event.compressed).to.be.true; + expect(event.timestamp).to.be.a('number'); + done(); + }); + + emitter.emitCloudFetchChunk({ + statementId: 'stmt-789', + chunkIndex: 2, + latencyMs: 250, + bytes: 204800, + compressed: true, + }); + }); + + it('should emit without optional fields', (done) => { + emitter.on(TelemetryEventType.CLOUDFETCH_CHUNK, (event: TelemetryEvent) => { + expect(event.eventType).to.equal(TelemetryEventType.CLOUDFETCH_CHUNK); + expect(event.statementId).to.equal('stmt-789'); + expect(event.chunkIndex).to.equal(0); + expect(event.bytes).to.equal(100000); + expect(event.latencyMs).to.be.undefined; + expect(event.compressed).to.be.undefined; + done(); + }); + + emitter.emitCloudFetchChunk({ + statementId: 'stmt-789', + chunkIndex: 0, + bytes: 100000, + }); + }); + + it('should not emit when telemetry is disabled', () => { + const disabledContext = { + getLogger: () => logger, + getConfig: () => ({ telemetryEnabled: false }), + } as any; + + const disabledEmitter = new TelemetryEventEmitter(disabledContext); + let eventEmitted = false; + + disabledEmitter.on(TelemetryEventType.CLOUDFETCH_CHUNK, () => { + eventEmitted = true; + }); + + disabledEmitter.emitCloudFetchChunk({ + statementId: 'stmt-789', + chunkIndex: 0, + bytes: 100000, + }); + + expect(eventEmitted).to.be.false; + }); - emitter.emitCloudFetchChunk({ statementId: 'st1', chunkIndex: 0, bytes: 512, compressed: true }); + it('should swallow exceptions and log at debug level', () => { + emitter.on(TelemetryEventType.CLOUDFETCH_CHUNK, () => { + throw new Error('Test error'); + }); + + emitter.emitCloudFetchChunk({ + statementId: 'stmt-789', + chunkIndex: 0, + bytes: 100000, + }); - expect(listener.calledOnce).to.be.true; - const event = listener.firstCall.args[0]; - expect(event.eventType).to.equal(TelemetryEventType.CLOUDFETCH_CHUNK); - expect(event.bytes).to.equal(512); - expect(event.compressed).to.be.true; + expect((logger.log as sinon.SinonStub).calledWith(LogLevel.debug)).to.be.true; + expect((logger.log as sinon.SinonStub).args[0][1]).to.include('Error emitting cloudfetch chunk'); }); }); - describe('emitError()', () => { - it('should emit an ERROR event with correct fields', () => { - const emitter = makeEmitter(true); - const listener = sinon.stub(); - emitter.on(TelemetryEventType.ERROR, listener); + describe('emitError', () => { + it('should emit error event with all fields', (done) => { + emitter.on(TelemetryEventType.ERROR, (event: TelemetryEvent) => { + expect(event.eventType).to.equal(TelemetryEventType.ERROR); + expect(event.statementId).to.equal('stmt-789'); + expect(event.sessionId).to.equal('session-123'); + expect(event.errorName).to.equal('AuthenticationError'); + expect(event.errorMessage).to.equal('Invalid credentials'); + expect(event.isTerminal).to.be.true; + expect(event.timestamp).to.be.a('number'); + done(); + }); + + emitter.emitError({ + statementId: 'stmt-789', + sessionId: 'session-123', + errorName: 'AuthenticationError', + errorMessage: 'Invalid credentials', + isTerminal: true, + }); + }); + + it('should emit error event with minimal fields', (done) => { + emitter.on(TelemetryEventType.ERROR, (event: TelemetryEvent) => { + expect(event.eventType).to.equal(TelemetryEventType.ERROR); + expect(event.errorName).to.equal('TimeoutError'); + expect(event.errorMessage).to.equal('Request timed out'); + expect(event.isTerminal).to.be.false; + expect(event.statementId).to.be.undefined; + expect(event.sessionId).to.be.undefined; + done(); + }); + + emitter.emitError({ + errorName: 'TimeoutError', + errorMessage: 'Request timed out', + isTerminal: false, + }); + }); + + it('should not emit when telemetry is disabled', () => { + const disabledContext = { + getLogger: () => logger, + getConfig: () => ({ telemetryEnabled: false }), + } as any; + + const disabledEmitter = new TelemetryEventEmitter(disabledContext); + let eventEmitted = false; + + disabledEmitter.on(TelemetryEventType.ERROR, () => { + eventEmitted = true; + }); + + disabledEmitter.emitError({ + errorName: 'Error', + errorMessage: 'Test', + isTerminal: false, + }); + + expect(eventEmitted).to.be.false; + }); + + it('should swallow exceptions and log at debug level', () => { + emitter.on(TelemetryEventType.ERROR, () => { + throw new Error('Test error'); + }); emitter.emitError({ - statementId: 'st1', - sessionId: 's1', - errorName: 'NetworkError', - errorMessage: 'timeout', + errorName: 'Error', + errorMessage: 'Test', isTerminal: false, }); - expect(listener.calledOnce).to.be.true; - const event = listener.firstCall.args[0]; - expect(event.eventType).to.equal(TelemetryEventType.ERROR); - expect(event.errorName).to.equal('NetworkError'); - expect(event.isTerminal).to.be.false; + expect((logger.log as sinon.SinonStub).calledWith(LogLevel.debug)).to.be.true; + expect((logger.log as sinon.SinonStub).args[0][1]).to.include('Error emitting error event'); + }); + }); + + describe('exception swallowing', () => { + it('should never propagate exceptions to caller', () => { + emitter.on(TelemetryEventType.CONNECTION_OPEN, () => { + throw new Error('Critical error'); + }); + + expect(() => { + emitter.emitConnectionOpen({ + sessionId: 'session-123', + workspaceId: 'workspace-456', + driverConfig: {} as DriverConfiguration, + latencyMs: 100, + }); + }).to.not.throw(); + }); + + it('should swallow multiple listener exceptions', () => { + emitter.on(TelemetryEventType.STATEMENT_START, () => { + throw new Error('First listener error'); + }); + emitter.on(TelemetryEventType.STATEMENT_START, () => { + throw new Error('Second listener error'); + }); + + expect(() => { + emitter.emitStatementStart({ + statementId: 'stmt-789', + sessionId: 'session-123', + }); + }).to.not.throw(); }); - it('should emit a terminal ERROR event', () => { - const emitter = makeEmitter(true); - const listener = sinon.stub(); - emitter.on(TelemetryEventType.ERROR, listener); + it('should log only at debug level, never at warn or error', () => { + emitter.on(TelemetryEventType.STATEMENT_COMPLETE, () => { + throw new Error('Test error'); + }); + emitter.on(TelemetryEventType.CLOUDFETCH_CHUNK, () => { + throw new Error('Test error'); + }); + emitter.on(TelemetryEventType.ERROR, () => { + throw new Error('Test error'); + }); - emitter.emitError({ errorName: 'AuthenticationError', errorMessage: 'auth failed', isTerminal: true }); + emitter.emitStatementComplete({ + statementId: 'stmt-1', + sessionId: 'session-1', + }); + emitter.emitCloudFetchChunk({ + statementId: 'stmt-1', + chunkIndex: 0, + bytes: 1000, + }); + emitter.emitError({ + errorName: 'Error', + errorMessage: 'Test', + isTerminal: false, + }); - const event = listener.firstCall.args[0]; - expect(event.isTerminal).to.be.true; + const logStub = logger.log as sinon.SinonStub; + for (let i = 0; i < logStub.callCount; i++) { + const level = logStub.args[i][0]; + expect(level).to.equal(LogLevel.debug); + } }); }); - describe('logging level compliance', () => { - it('should never log at warn or error level', () => { - const context = new ClientContextStub({ telemetryEnabled: true } as any); - const logSpy = sinon.spy(context.logger, 'log'); - const emitter = new TelemetryEventEmitter(context); + describe('no console logging', () => { + it('should not use console.log', () => { + const consoleSpy = sinon.spy(console, 'log'); emitter.on(TelemetryEventType.CONNECTION_OPEN, () => { - throw new Error('boom'); + throw new Error('Test error'); + }); + + emitter.emitConnectionOpen({ + sessionId: 'session-123', + workspaceId: 'workspace-456', + driverConfig: {} as DriverConfiguration, + latencyMs: 100, + }); + + expect(consoleSpy.called).to.be.false; + consoleSpy.restore(); + }); + + it('should not use console.debug', () => { + const consoleSpy = sinon.spy(console, 'debug'); + + emitter.on(TelemetryEventType.STATEMENT_START, () => { + throw new Error('Test error'); + }); + + emitter.emitStatementStart({ + statementId: 'stmt-789', + sessionId: 'session-123', + }); + + expect(consoleSpy.called).to.be.false; + consoleSpy.restore(); + }); + + it('should not use console.error', () => { + const consoleSpy = sinon.spy(console, 'error'); + + emitter.on(TelemetryEventType.ERROR, () => { + throw new Error('Test error'); + }); + + emitter.emitError({ + errorName: 'Error', + errorMessage: 'Test', + isTerminal: true, + }); + + expect(consoleSpy.called).to.be.false; + consoleSpy.restore(); + }); + }); + + describe('respects telemetryEnabled flag', () => { + it('should respect flag from context.getConfig()', () => { + const customContext = { + getLogger: () => logger, + getConfig: () => ({ + telemetryEnabled: true, + directResultsDefaultMaxRows: 10000, + fetchChunkDefaultMaxRows: 100000, + socketTimeout: 900000, + retryMaxAttempts: 30, + retriesTimeout: 900000, + retryDelayMin: 1000, + retryDelayMax: 30000, + useCloudFetch: true, + cloudFetchConcurrentDownloads: 10, + cloudFetchSpeedThresholdMBps: 0, + useLZ4Compression: true, + }), + } as any; + + const customEmitter = new TelemetryEventEmitter(customContext); + let eventCount = 0; + + customEmitter.on(TelemetryEventType.CONNECTION_OPEN, () => { + eventCount++; + }); + + customEmitter.emitConnectionOpen({ + sessionId: 'session-123', + workspaceId: 'workspace-456', + driverConfig: {} as DriverConfiguration, + latencyMs: 100, }); - emitter.emitConnectionOpen({ sessionId: 's1', workspaceId: 'w1', driverConfig: {} as any }); + expect(eventCount).to.equal(1); + }); + + it('should not emit when explicitly disabled', () => { + const disabledContext = { + getLogger: () => logger, + getConfig: () => ({ + telemetryEnabled: false, + }), + } as any; + + const disabledEmitter = new TelemetryEventEmitter(disabledContext); + let eventCount = 0; + + disabledEmitter.on(TelemetryEventType.CONNECTION_OPEN, () => { + eventCount++; + }); + disabledEmitter.on(TelemetryEventType.STATEMENT_START, () => { + eventCount++; + }); + disabledEmitter.on(TelemetryEventType.STATEMENT_COMPLETE, () => { + eventCount++; + }); + disabledEmitter.on(TelemetryEventType.CLOUDFETCH_CHUNK, () => { + eventCount++; + }); + disabledEmitter.on(TelemetryEventType.ERROR, () => { + eventCount++; + }); - expect(logSpy.neverCalledWith(LogLevel.error, sinon.match.any)).to.be.true; - expect(logSpy.neverCalledWith(LogLevel.warn, sinon.match.any)).to.be.true; + disabledEmitter.emitConnectionOpen({ + sessionId: 'session-123', + workspaceId: 'workspace-456', + driverConfig: {} as DriverConfiguration, + latencyMs: 100, + }); + disabledEmitter.emitStatementStart({ + statementId: 'stmt-789', + sessionId: 'session-123', + }); + disabledEmitter.emitStatementComplete({ + statementId: 'stmt-789', + sessionId: 'session-123', + }); + disabledEmitter.emitCloudFetchChunk({ + statementId: 'stmt-789', + chunkIndex: 0, + bytes: 1000, + }); + disabledEmitter.emitError({ + errorName: 'Error', + errorMessage: 'Test', + isTerminal: false, + }); - logSpy.restore(); + expect(eventCount).to.equal(0); }); }); });