[4/7] Telemetry Event Emission and Aggregation

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,14 @@ import IDBSQLLogger, { LogLevel } from './contracts/IDBSQLLogger';
 import DBSQLLogger from './DBSQLLogger';
 import CloseableCollection from './utils/CloseableCollection';
 import IConnectionProvider from './connection/contracts/IConnectionProvider';
+import FeatureFlagCache from './telemetry/FeatureFlagCache';
+import TelemetryClientProvider from './telemetry/TelemetryClientProvider';
+import TelemetryEventEmitter from './telemetry/TelemetryEventEmitter';
+import MetricsAggregator from './telemetry/MetricsAggregator';
+import DatabricksTelemetryExporter from './telemetry/DatabricksTelemetryExporter';
+import { CircuitBreakerRegistry } from './telemetry/CircuitBreaker';
+import { DriverConfiguration, DRIVER_NAME } from './telemetry/types';
+import driverVersion from './version';
 
 function prependSlash(str: string): string {
   if (str.length > 0 && str.charAt(0) !== '/') {
@@ -75,6 +84,23 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
 
   private readonly sessions = new CloseableCollection<DBSQLSession>();
 
+  // Telemetry components (instance-based, NOT singletons)
+  private host?: string;
+
+  private httpPath?: string;
+
+  private authType?: string;
+
+  private featureFlagCache?: FeatureFlagCache;
+
+  private telemetryClientProvider?: TelemetryClientProvider;
+
+  private telemetryEmitter?: TelemetryEventEmitter;
+
+  private telemetryAggregator?: MetricsAggregator;
+
+  private circuitBreakerRegistry?: CircuitBreakerRegistry;
+
   private static getDefaultLogger(): IDBSQLLogger {
     if (!this.defaultLogger) {
       this.defaultLogger = new DBSQLLogger();
@@ -101,6 +127,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 +247,214 @@ 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 {
+    if (options.authType === 'databricks-oauth') {
+      // Check if M2M (has client secret) or U2M (no client secret)
+      return options.oauthClientSecret === undefined
+        ? 'external-browser' // U2M OAuth (User-to-Machine)
+        : 'oauth-m2m'; // M2M OAuth (Machine-to-Machine)
+    }
+
+    if (options.authType === 'custom') {
+      return 'custom'; // Custom auth provider
+    }
+
+    // 'access-token' or undefined
+    return 'pat'; // Personal Access Token
+  }
+
+  /**
+   * 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<void> {
+    if (!this.host) {
+      return;
+    }
+
+    try {
+      // Create circuit breaker registry (shared by feature flags and telemetry)
+      this.circuitBreakerRegistry = new CircuitBreakerRegistry(this);
+
+      // Create feature flag cache instance with circuit breaker protection
+      this.featureFlagCache = new FeatureFlagCache(this, this.circuitBreakerRegistry);
+      this.featureFlagCache.getOrCreateContext(this.host);
+
+      // Check if telemetry enabled via feature flag
+      const enabled = await this.featureFlagCache.isTelemetryEnabled(this.host);
+
+      if (!enabled) {
+        this.logger.log(LogLevel.debug, 'Telemetry: disabled');
+        return;
+      }
+
+      // Create telemetry components (all instance-based)
+      this.telemetryClientProvider = new TelemetryClientProvider(this);
+      this.telemetryEmitter = new TelemetryEventEmitter(this);
+
+      // Get or create telemetry client for this host (increments refCount)
+      this.telemetryClientProvider.getOrCreateClient(this.host);
+
+      // Create telemetry exporter with shared circuit breaker registry
+      const exporter = new DatabricksTelemetryExporter(this, this.host, this.circuitBreakerRegistry);
+      this.telemetryAggregator = new MetricsAggregator(this, exporter);
+
+      // Wire up event listeners
+      this.telemetryEmitter.on('connection.open', (event) => {
+        try {
+          this.telemetryAggregator?.processEvent(event);
+        } catch (error: any) {
+          this.logger.log(LogLevel.debug, `Error processing connection.open event: ${error.message}`);
+        }
+      });
+
+      this.telemetryEmitter.on('connection.close', (event) => {
+        try {
+          this.telemetryAggregator?.processEvent(event);
+        } catch (error: any) {
+          this.logger.log(LogLevel.debug, `Error processing connection.close event: ${error.message}`);
+        }
+      });
+
+      this.telemetryEmitter.on('statement.start', (event) => {
+        try {
+          this.telemetryAggregator?.processEvent(event);
+        } catch (error: any) {
+          this.logger.log(LogLevel.debug, `Error processing statement.start event: ${error.message}`);
+        }
+      });
+
+      this.telemetryEmitter.on('statement.complete', (event) => {
+        try {
+          this.telemetryAggregator?.processEvent(event);
+        } catch (error: any) {
+          this.logger.log(LogLevel.debug, `Error processing statement.complete event: ${error.message}`);
+        }
+      });
+
+      this.telemetryEmitter.on('cloudfetch.chunk', (event) => {
+        try {
+          this.telemetryAggregator?.processEvent(event);
+        } catch (error: any) {
+          this.logger.log(LogLevel.debug, `Error processing cloudfetch.chunk event: ${error.message}`);
+        }
+      });
+
+      this.telemetryEmitter.on('error', (event) => {
+        try {
+          this.telemetryAggregator?.processEvent(event);
+        } catch (error: any) {
+          this.logger.log(LogLevel.debug, `Error processing error event: ${error.message}`);
+        }
+      });
+
+      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 +476,27 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
       }
     }
 
+    // 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
+    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;
+    }
+
     // 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 +536,11 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
       this.emit('timeout');
     });
 
+    // Initialize telemetry if enabled
+    if (this.config.telemetryEnabled) {
+      await this.initializeTelemetry();
+    }
+
     return this;
   }
 
@@ -290,6 +554,9 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
    * const session = await client.openSession();
    */
   public async openSession(request: OpenSessionRequest = {}): Promise<IDBSQLSession> {
+    // Track connection open latency
+    const startTime = Date.now();
+
     // Prepare session configuration
     const configuration = request.configuration ? { ...request.configuration } : {};
 
@@ -312,12 +579,54 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
       serverProtocolVersion: response.serverProtocolVersion,
     });
     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}`);
+      }
+    }
+
     return session;
   }
 
   public async close(): Promise<void> {
     await this.sessions.closeAll();
 
+    // Cleanup telemetry
+    if (this.host) {
+      try {
+        // Step 1: Close aggregator (stops timer, completes statements, final flush)
+        if (this.telemetryAggregator) {
+          this.telemetryAggregator.close();
+        }
+
+        // Step 2: Release telemetry client (decrements ref count, closes if last)
+        if (this.telemetryClientProvider) {
+          await this.telemetryClientProvider.releaseClient(this.host);
+        }
+
+        // Step 3: Release feature flag context (decrements ref count)
+        if (this.featureFlagCache) {
+          this.featureFlagCache.releaseContext(this.host);
+        }
+      } catch (error: any) {
+        // Swallow all telemetry cleanup errors
+        this.logger.log(LogLevel.debug, `Telemetry cleanup error: ${error.message}`);
+      }
+    }
+
     this.client = undefined;
     this.connectionProvider = undefined;
     this.authProvider = undefined;