databricks
diff --git a/‎CHANGELOG.md‎
Lines changed: 7 additions & 0 deletions b/‎CHANGELOG.md‎
Lines changed: 7 additions & 0 deletions
diff --git a/‎lib/DBSQLClient.ts‎
Lines changed: 54 additions & 26 deletions b/‎lib/DBSQLClient.ts‎
Lines changed: 54 additions & 26 deletions
diff --git a/‎lib/DBSQLOperation.ts‎
Lines changed: 57 additions & 73 deletions b/‎lib/DBSQLOperation.ts‎
Lines changed: 57 additions & 73 deletions
@@ -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
 
 
@@ -36,6 +36,7 @@ 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 {
@@ -429,6 +430,19 @@ 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;
@@ -440,14 +454,22 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
     }
 
     // Override telemetry config if provided in options
-    if (options.telemetryEnabled !== undefined) {
-      this.config.telemetryEnabled = options.telemetryEnabled;
-    }
-    if (options.telemetryBatchSize !== undefined) {
-      this.config.telemetryBatchSize = options.telemetryBatchSize;
-    }
-    if (options.telemetryAuthenticatedExport !== undefined) {
-      this.config.telemetryAuthenticatedExport = options.telemetryAuthenticatedExport;
+    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
@@ -489,8 +511,15 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
       this.emit('timeout');
     });
 
-    // Initialize telemetry if enabled
-    if (this.config.telemetryEnabled) {
+    // 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();
     }
 
@@ -534,22 +563,18 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
     this.sessions.add(session);
 
     // Emit connection.open telemetry event
-    if (this.telemetryEmitter && this.host) {
-      try {
-        const latencyMs = Date.now() - startTime;
-        const workspaceId = this.extractWorkspaceId(this.host);
-        const driverConfig = this.buildDriverConfiguration();
-        this.telemetryEmitter.emitConnectionOpen({
-          sessionId: session.id,
-          workspaceId,
-          driverConfig,
-          latencyMs,
-        });
-      } catch (error: any) {
-        // CRITICAL: All telemetry exceptions swallowed
-        this.logger.log(LogLevel.debug, `Error emitting connection.open event: ${error.message}`);
-      }
-    }
+    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;
   }
@@ -578,6 +603,9 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
       }
       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;
 
@@ -36,6 +36,7 @@ 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;
@@ -181,12 +182,7 @@ export default class DBSQLOperation implements IOperation {
    * const result = await queryOperation.fetchChunk({maxRows: 1000});
    */
   public async fetchChunk(options?: FetchOptions): Promise<Array<object>> {
-    try {
-      return await this.fetchChunkInternal(options);
-    } catch (err: any) {
-      this.emitErrorEvent(err);
-      throw err;
-    }
+    return this.withErrorTelemetry(() => this.fetchChunkInternal(options));
   }
 
   private async fetchChunkInternal(options?: FetchOptions): Promise<Array<object>> {
@@ -266,12 +262,7 @@ export default class DBSQLOperation implements IOperation {
    * @throws {StatusError}
    */
   public async cancel(): Promise<Status> {
-    try {
-      return await this.cancelInternal();
-    } catch (err: any) {
-      this.emitErrorEvent(err);
-      throw err;
-    }
+    return this.withErrorTelemetry(() => this.cancelInternal());
   }
 
   private async cancelInternal(): Promise<Status> {
@@ -299,12 +290,7 @@ export default class DBSQLOperation implements IOperation {
    * @throws {StatusError}
    */
   public async close(): Promise<Status> {
-    try {
-      return await this.closeInternal();
-    } catch (err: any) {
-      this.emitErrorEvent(err);
-      throw err;
-    }
+    return this.withErrorTelemetry(() => this.closeInternal());
   }
 
   private async closeInternal(): Promise<Status> {
@@ -332,46 +318,63 @@ export default class DBSQLOperation implements IOperation {
   }
 
   public async finished(options?: FinishedOptions): Promise<void> {
-    await this.failIfClosed();
-    await this.waitUntilReady(options);
+    return this.withErrorTelemetry(async () => {
+      await this.failIfClosed();
+      await this.waitUntilReady(options);
+    });
   }
 
   public async hasMoreRows(): Promise<boolean> {
-    // 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<TTableSchema | null> {
-    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<TGetResultSetMetadataResp> {
-    try {
+    return this.withErrorTelemetry(async () => {
       await this.failIfClosed();
       await this.waitUntilReady();
-      return await this.fetchMetadata();
+      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<T>(fn: () => Promise<T>): Promise<T> {
+    try {
+      return await fn();
     } catch (err: any) {
       this.emitErrorEvent(err);
       throw err;
@@ -540,33 +543,23 @@ export default class DBSQLOperation implements IOperation {
    * CRITICAL: All exceptions swallowed and logged at LogLevel.debug ONLY.
    */
   private emitStatementStart(): void {
-    try {
-      const telemetryEmitter = this.context.getTelemetryEmitter?.();
-      if (!telemetryEmitter) {
-        return;
-      }
-
-      telemetryEmitter.emitStatementStart({
+    safeEmit(this.context, (emitter) => {
+      emitter.emitStatementStart({
         statementId: this.id,
         sessionId: this.sessionId || '',
         operationType: mapOperationTypeToTelemetryType(this.operationHandle.operationType),
       });
-    } catch (error: any) {
-      this.context.getLogger().log(LogLevel.debug, `Error emitting statement.start event: ${error.message}`);
-    }
+    });
   }
 
   /**
    * Emit statement.complete telemetry event and complete aggregation.
    * CRITICAL: All exceptions swallowed and logged at LogLevel.debug ONLY.
    */
   private async emitStatementComplete(): Promise<void> {
-    try {
-      const telemetryEmitter = this.context.getTelemetryEmitter?.();
-      const telemetryAggregator = this.context.getTelemetryAggregator?.();
-      if (!telemetryEmitter || !telemetryAggregator) {
-        return;
-      }
+    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
@@ -575,22 +568,18 @@ export default class DBSQLOperation implements IOperation {
       // would then fire spurious error telemetry from `getMetadata`'s error
       // wrapper.
       const resultFormat = mapResultFormatToTelemetryType(this.metadata?.resultFormat);
-
       const latencyMs = Date.now() - this.startTime;
 
-      telemetryEmitter.emitStatementComplete({
+      emitter.emitStatementComplete({
         statementId: this.id,
         sessionId: this.sessionId || '',
         latencyMs,
         resultFormat,
         pollCount: this.pollCount,
       });
 
-      // Complete statement aggregation
-      telemetryAggregator.completeStatement(this.id);
-    } catch (error: any) {
-      this.context.getLogger().log(LogLevel.debug, `Error emitting statement.complete event: ${error.message}`);
-    }
+      aggregator.completeStatement(this.id);
+    });
   }
 
   /**
@@ -601,20 +590,15 @@ export default class DBSQLOperation implements IOperation {
    * debug level — telemetry must never break the driver.
    */
   private emitErrorEvent(error: Error): void {
-    try {
-      const telemetryEmitter = this.context.getTelemetryEmitter?.();
-      if (!telemetryEmitter) return;
-
-      telemetryEmitter.emitError({
+    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),
       });
-    } catch (emitError: any) {
-      this.context.getLogger().log(LogLevel.debug, `Error emitting error event: ${emitError.message}`);
-    }
+    });
   }
 }