[PECOBLR-1382] Implement telemetry Phase 7: Driver integration

This commit implements Phase 7 (driver integration) for the telemetry system,
completing the full telemetry pipeline from driver operations to export.

Phase 7: Driver Integration
- Add telemetry configuration to UserConfig
  - EnableTelemetry: User opt-in flag (respects server feature flags)
  - ForceEnableTelemetry: Force enable flag (bypasses server checks)
  - DSN parameter parsing in ParseDSN()
  - DeepCopy support for telemetry fields

- Add telemetry support to connection
  - Add telemetry field to conn struct (*telemetry.Interceptor)
  - Initialize telemetry in connector.Connect()
  - Release telemetry resources in conn.Close()
  - Graceful shutdown with pending metric flush

- Export telemetry types for driver use
  - Export Interceptor type (was interceptor)
  - Export GetInterceptor() method (was getInterceptor)
  - Export Close() method (was close)

- Create driver integration helper (driver_integration.go)
  - InitializeForConnection(): One-stop initialization
  - ReleaseForConnection(): Resource cleanup
  - Encapsulates feature flag checks and client management
  - Reference counting for per-host resources

Integration Flow:
1. User sets enableTelemetry=true or forceEnableTelemetry=true in DSN
2. connector.Connect() calls telemetry.InitializeForConnection()
3. Telemetry checks feature flags and returns Interceptor if enabled
4. Connection uses Interceptor for metric collection (Phase 8)
5. conn.Close() releases telemetry resources

Architecture:
- Per-connection: Interceptor instance
- Per-host (shared): telemetryClient, aggregator, exporter
- Global (singleton): clientManager, featureFlagCache, circuitBreakerManager

Opt-In Priority (5 levels):
1. forceEnableTelemetry=true - Always enabled (testing/internal)
2. enableTelemetry=false - Always disabled (explicit opt-out)
3. enableTelemetry=true + server flag - User opt-in with server control
4. Server flag only - Default Databricks-controlled behavior
5. Default - Disabled (fail-safe)

Testing:
- All 70+ telemetry tests passing
- No breaking changes to existing driver tests
- Compilation verified across all packages
- Graceful handling when telemetry disabled

Note: Statement hooks (beforeExecute/afterExecute) will be added in follow-up
for actual metric collection during query execution.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: samikshya-db

connection.go

@@ -22,14 +22,16 @@ import (
 	"github.com/databricks/databricks-sql-go/internal/sentinel"
 	"github.com/databricks/databricks-sql-go/internal/thrift_protocol"
 	"github.com/databricks/databricks-sql-go/logger"
+	"github.com/databricks/databricks-sql-go/telemetry"
 	"github.com/pkg/errors"
 )
 
 type conn struct {
-	id      string
-	cfg     *config.Config
-	client  cli_service.TCLIService
-	session *cli_service.TOpenSessionResp
+	id        string
+	cfg       *config.Config
+	client    cli_service.TCLIService
+	session   *cli_service.TOpenSessionResp
+	telemetry *telemetry.Interceptor // Optional telemetry interceptor
 }
 
 // Prepare prepares a statement with the query bound to this connection.
@@ -49,6 +51,12 @@ func (c *conn) Close() error {
 	log := logger.WithContext(c.id, "", "")
 	ctx := driverctx.NewContextWithConnId(context.Background(), c.id)
 
+	// Close telemetry and release resources
+	if c.telemetry != nil {
+		_ = c.telemetry.Close(ctx)
+		telemetry.ReleaseForConnection(c.cfg.Host)
+	}
+
 	_, err := c.client.CloseSession(ctx, &cli_service.TCloseSessionReq{
 		SessionHandle: c.session.SessionHandle,
 	})

connector.go

@@ -20,6 +20,7 @@ import (
 	"github.com/databricks/databricks-sql-go/internal/config"
 	dbsqlerrint "github.com/databricks/databricks-sql-go/internal/errors"
 	"github.com/databricks/databricks-sql-go/logger"
+	"github.com/databricks/databricks-sql-go/telemetry"
 )
 
 type connector struct {
@@ -75,6 +76,20 @@ func (c *connector) Connect(ctx context.Context) (driver.Conn, error) {
 	}
 	log := logger.WithContext(conn.id, driverctx.CorrelationIdFromContext(ctx), "")
 
+	// Initialize telemetry if configured
+	if c.cfg.EnableTelemetry || c.cfg.ForceEnableTelemetry {
+		conn.telemetry = telemetry.InitializeForConnection(
+			ctx,
+			c.cfg.Host,
+			c.client,
+			c.cfg.EnableTelemetry,
+			c.cfg.ForceEnableTelemetry,
+		)
+		if conn.telemetry != nil {
+			log.Debug().Msg("telemetry initialized for connection")
+		}
+	}
+
 	log.Info().Msgf("connect: host=%s port=%d httpPath=%s serverProtocolVersion=0x%X", c.cfg.Host, c.cfg.Port, c.cfg.HTTPPath, session.ServerProtocolVersion)
 
 	return conn, nil

internal/config/config.go

@@ -98,6 +98,9 @@ type UserConfig struct {
 	RetryWaitMin             time.Duration
 	RetryWaitMax             time.Duration
 	RetryMax                 int
+	// Telemetry configuration
+	EnableTelemetry          bool   // Opt-in for telemetry (respects server feature flags)
+	ForceEnableTelemetry     bool   // Force enable telemetry (bypasses server checks)
 	Transport                http.RoundTripper
 	UseLz4Compression        bool
 	EnableMetricViewMetadata bool
@@ -144,6 +147,8 @@ func (ucfg UserConfig) DeepCopy() UserConfig {
 		UseLz4Compression:        ucfg.UseLz4Compression,
 		EnableMetricViewMetadata: ucfg.EnableMetricViewMetadata,
 		CloudFetchConfig:         ucfg.CloudFetchConfig,
+		EnableTelemetry:          ucfg.EnableTelemetry,
+		ForceEnableTelemetry:     ucfg.ForceEnableTelemetry,
 	}
 }
 
@@ -282,6 +287,21 @@ func ParseDSN(dsn string) (UserConfig, error) {
 		ucfg.EnableMetricViewMetadata = enableMetricViewMetadata
 	}
 
+	// Telemetry parameters
+	if enableTelemetry, ok, err := params.extractAsBool("enableTelemetry"); ok {
+		if err != nil {
+			return UserConfig{}, err
+		}
+		ucfg.EnableTelemetry = enableTelemetry
+	}
+
+	if forceEnableTelemetry, ok, err := params.extractAsBool("forceEnableTelemetry"); ok {
+		if err != nil {
+			return UserConfig{}, err
+		}
+		ucfg.ForceEnableTelemetry = forceEnableTelemetry
+	}
+
 	// for timezone we do a case insensitive key match.
 	// We use getNoCase because we want to leave timezone in the params so that it will also
 	// be used as a session param.

telemetry/client.go

@@ -80,8 +80,9 @@ func (c *telemetryClient) close() error {
 	return c.aggregator.close(ctx)
 }
 
-// getInterceptor returns a new interceptor for a connection.
+// GetInterceptor returns a new interceptor for a connection.
 // Each connection gets its own interceptor, but they all share the same aggregator.
-func (c *telemetryClient) getInterceptor(enabled bool) *interceptor {
+// Exported for use by the driver package.
+func (c *telemetryClient) GetInterceptor(enabled bool) *Interceptor {
 	return newInterceptor(c.aggregator, enabled)
 }

telemetry/driver_integration.go

@@ -0,0 +1,63 @@
+package telemetry
+
+import (
+	"context"
+	"net/http"
+)
+
+// InitializeForConnection initializes telemetry for a database connection.
+// Returns an Interceptor if telemetry is enabled, nil otherwise.
+// This function handles all the logic for checking feature flags and creating the interceptor.
+//
+// Parameters:
+//   - ctx: Context for the initialization
+//   - host: Databricks host
+//   - httpClient: HTTP client for making requests
+//   - enableTelemetry: User opt-in flag
+//   - forceEnableTelemetry: Force enable flag (bypasses server checks)
+//
+// Returns:
+//   - *Interceptor: Telemetry interceptor if enabled, nil otherwise
+func InitializeForConnection(
+	ctx context.Context,
+	host string,
+	httpClient *http.Client,
+	enableTelemetry bool,
+	forceEnableTelemetry bool,
+) *Interceptor {
+	// Create telemetry config
+	cfg := DefaultConfig()
+	cfg.EnableTelemetry = enableTelemetry
+	cfg.ForceEnableTelemetry = forceEnableTelemetry
+
+	// Check if telemetry should be enabled
+	if !isTelemetryEnabled(ctx, cfg, host, httpClient) {
+		return nil
+	}
+
+	// Get or create telemetry client for this host
+	clientMgr := getClientManager()
+	telemetryClient := clientMgr.getOrCreateClient(host, httpClient, cfg)
+
+	// Get feature flag cache context (for reference counting)
+	flagCache := getFeatureFlagCache()
+	flagCache.getOrCreateContext(host)
+
+	// Return interceptor
+	return telemetryClient.GetInterceptor(true)
+}
+
+// ReleaseForConnection releases telemetry resources for a connection.
+// Should be called when the connection is closed.
+//
+// Parameters:
+//   - host: Databricks host
+func ReleaseForConnection(host string) {
+	// Release client manager reference
+	clientMgr := getClientManager()
+	_ = clientMgr.releaseClient(host)
+
+	// Release feature flag cache reference
+	flagCache := getFeatureFlagCache()
+	flagCache.releaseContext(host)
+}

telemetry/interceptor.go

@@ -5,8 +5,9 @@ import (
 	"time"
 )
 
-// interceptor wraps driver operations to collect metrics.
-type interceptor struct {
+// Interceptor wraps driver operations to collect metrics.
+// Exported for use by the driver package.
+type Interceptor struct {
 	aggregator *metricsAggregator
 	enabled    bool
 }
@@ -23,8 +24,8 @@ type contextKey int
 const metricContextKey contextKey = 0
 
 // newInterceptor creates a new telemetry interceptor.
-func newInterceptor(aggregator *metricsAggregator, enabled bool) *interceptor {
-	return &interceptor{
+func newInterceptor(aggregator *metricsAggregator, enabled bool) *Interceptor {
+	return &Interceptor{
 		aggregator: aggregator,
 		enabled:    enabled,
 	}
@@ -45,7 +46,7 @@ func getMetricContext(ctx context.Context) *metricContext {
 
 // beforeExecute is called before statement execution.
 // Returns a new context with metric tracking attached.
-func (i *interceptor) beforeExecute(ctx context.Context, statementID string) context.Context {
+func (i *Interceptor) beforeExecute(ctx context.Context, statementID string) context.Context {
 	if !i.enabled {
 		return ctx
 	}
@@ -61,7 +62,7 @@ func (i *interceptor) beforeExecute(ctx context.Context, statementID string) con
 
 // afterExecute is called after statement execution.
 // Records the metric with timing and error information.
-func (i *interceptor) afterExecute(ctx context.Context, err error) {
+func (i *Interceptor) afterExecute(ctx context.Context, err error) {
 	if !i.enabled {
 		return
 	}
@@ -96,7 +97,7 @@ func (i *interceptor) afterExecute(ctx context.Context, err error) {
 }
 
 // addTag adds a tag to the current metric context.
-func (i *interceptor) addTag(ctx context.Context, key string, value interface{}) {
+func (i *Interceptor) addTag(ctx context.Context, key string, value interface{}) {
 	if !i.enabled {
 		return
 	}
@@ -108,7 +109,7 @@ func (i *interceptor) addTag(ctx context.Context, key string, value interface{})
 }
 
 // recordConnection records a connection event.
-func (i *interceptor) recordConnection(ctx context.Context, tags map[string]interface{}) {
+func (i *Interceptor) recordConnection(ctx context.Context, tags map[string]interface{}) {
 	if !i.enabled {
 		return
 	}
@@ -129,16 +130,17 @@ func (i *interceptor) recordConnection(ctx context.Context, tags map[string]inte
 }
 
 // completeStatement marks a statement as complete and flushes aggregated metrics.
-func (i *interceptor) completeStatement(ctx context.Context, statementID string, failed bool) {
+func (i *Interceptor) completeStatement(ctx context.Context, statementID string, failed bool) {
 	if !i.enabled {
 		return
 	}
 
 	i.aggregator.completeStatement(ctx, statementID, failed)
 }
 
-// close shuts down the interceptor and flushes pending metrics.
-func (i *interceptor) close(ctx context.Context) error {
+// Close shuts down the interceptor and flushes pending metrics.
+// Exported for use by the driver package.
+func (i *Interceptor) Close(ctx context.Context) error {
 	if !i.enabled {
 		return nil
 	}