[PECOBLR-1381] Implement telemetry Phase 6: Metric collection & aggregation

This commit implements Phase 6 (metric collection and aggregation) for the
telemetry system.

Phase 6: Metric Collection & Aggregation
- Implement error classification (errors.go)
  - isTerminalError() for identifying non-retryable errors
  - classifyError() for categorizing errors for telemetry
  - HTTP error handling utilities

- Implement telemetry interceptor (interceptor.go)
  - beforeExecute() / afterExecute() hooks for statement execution
  - Context-based metric tracking with metricContext
  - Latency measurement and tag collection
  - Connection event recording
  - Error swallowing with panic recovery

- Implement metrics aggregator (aggregator.go)
  - Statement-level metric aggregation
  - Batch size and flush interval logic
  - Background flush goroutine with ticker
  - Thread-safe metric recording with mutex protection
  - Immediate flush for connection and terminal errors
  - Aggregated counts (chunks, bytes, polls)

- Update telemetryClient (client.go)
  - Wire up aggregator with exporter
  - Automatic aggregator start in constructor
  - Graceful shutdown with 5s timeout
  - getInterceptor() for per-connection interceptors

Architecture:
- Each connection gets its own interceptor instance
- All interceptors share the same aggregator (per host)
- Aggregator batches metrics and flushes periodically
- Exporter sends batched metrics to Databricks
- Circuit breaker protects against endpoint failures

Testing:
- All 70+ existing tests continue to pass
- Compilation verified, no breaking changes

Note: Phase 7 (driver integration) will be completed separately to allow
careful review and testing of hooks in connection.go and statement.go.

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

Author: samikshya-db

telemetry/aggregator.go

@@ -0,0 +1,226 @@
+package telemetry
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+// metricsAggregator aggregates metrics by statement and batches for export.
+type metricsAggregator struct {
+	mu sync.RWMutex
+
+	statements map[string]*statementMetrics
+	batch      []*telemetryMetric
+	exporter   *telemetryExporter
+
+	batchSize     int
+	flushInterval time.Duration
+	stopCh        chan struct{}
+	flushTimer    *time.Ticker
+}
+
+// statementMetrics holds aggregated metrics for a statement.
+type statementMetrics struct {
+	statementID     string
+	sessionID       string
+	totalLatency    time.Duration
+	chunkCount      int
+	bytesDownloaded int64
+	pollCount       int
+	errors          []string
+	tags            map[string]interface{}
+}
+
+// newMetricsAggregator creates a new metrics aggregator.
+func newMetricsAggregator(exporter *telemetryExporter, cfg *Config) *metricsAggregator {
+	agg := &metricsAggregator{
+		statements:    make(map[string]*statementMetrics),
+		batch:         make([]*telemetryMetric, 0, cfg.BatchSize),
+		exporter:      exporter,
+		batchSize:     cfg.BatchSize,
+		flushInterval: cfg.FlushInterval,
+		stopCh:        make(chan struct{}),
+	}
+
+	// Start background flush timer
+	go agg.flushLoop()
+
+	return agg
+}
+
+// recordMetric records a metric for aggregation.
+func (agg *metricsAggregator) recordMetric(ctx context.Context, metric *telemetryMetric) {
+	// Swallow all errors
+	defer func() {
+		if r := recover(); r != nil {
+			// Log at trace level only
+			// logger.Trace().Msgf("telemetry: recordMetric panic: %v", r)
+		}
+	}()
+
+	agg.mu.Lock()
+	defer agg.mu.Unlock()
+
+	switch metric.metricType {
+	case "connection":
+		// Emit connection events immediately
+		agg.batch = append(agg.batch, metric)
+		if len(agg.batch) >= agg.batchSize {
+			agg.flushUnlocked(ctx)
+		}
+
+	case "statement":
+		// Aggregate by statement ID
+		stmt, exists := agg.statements[metric.statementID]
+		if !exists {
+			stmt = &statementMetrics{
+				statementID: metric.statementID,
+				sessionID:   metric.sessionID,
+				tags:        make(map[string]interface{}),
+			}
+			agg.statements[metric.statementID] = stmt
+		}
+
+		// Update aggregated values
+		stmt.totalLatency += time.Duration(metric.latencyMs) * time.Millisecond
+		if chunkCount, ok := metric.tags["chunk_count"].(int); ok {
+			stmt.chunkCount += chunkCount
+		}
+		if bytes, ok := metric.tags["bytes_downloaded"].(int64); ok {
+			stmt.bytesDownloaded += bytes
+		}
+		if pollCount, ok := metric.tags["poll_count"].(int); ok {
+			stmt.pollCount += pollCount
+		}
+
+		// Store error if present
+		if metric.errorType != "" {
+			stmt.errors = append(stmt.errors, metric.errorType)
+		}
+
+		// Merge tags
+		for k, v := range metric.tags {
+			stmt.tags[k] = v
+		}
+
+	case "error":
+		// Check if terminal error
+		if metric.errorType != "" && isTerminalError(&simpleError{msg: metric.errorType}) {
+			// Flush terminal errors immediately
+			agg.batch = append(agg.batch, metric)
+			agg.flushUnlocked(ctx)
+		} else {
+			// Buffer non-terminal errors with statement
+			if stmt, exists := agg.statements[metric.statementID]; exists {
+				stmt.errors = append(stmt.errors, metric.errorType)
+			}
+		}
+	}
+}
+
+// completeStatement marks a statement as complete and emits aggregated metric.
+func (agg *metricsAggregator) completeStatement(ctx context.Context, statementID string, failed bool) {
+	defer func() {
+		if r := recover(); r != nil {
+			// Log at trace level only
+		}
+	}()
+
+	agg.mu.Lock()
+	defer agg.mu.Unlock()
+
+	stmt, exists := agg.statements[statementID]
+	if !exists {
+		return
+	}
+	delete(agg.statements, statementID)
+
+	// Create aggregated metric
+	metric := &telemetryMetric{
+		metricType:  "statement",
+		timestamp:   time.Now(),
+		statementID: stmt.statementID,
+		sessionID:   stmt.sessionID,
+		latencyMs:   stmt.totalLatency.Milliseconds(),
+		tags:        stmt.tags,
+	}
+
+	// Add aggregated counts
+	metric.tags["chunk_count"] = stmt.chunkCount
+	metric.tags["bytes_downloaded"] = stmt.bytesDownloaded
+	metric.tags["poll_count"] = stmt.pollCount
+
+	// Add error information if failed
+	if failed && len(stmt.errors) > 0 {
+		// Use the first error as the primary error type
+		metric.errorType = stmt.errors[0]
+	}
+
+	agg.batch = append(agg.batch, metric)
+
+	// Flush if batch full
+	if len(agg.batch) >= agg.batchSize {
+		agg.flushUnlocked(ctx)
+	}
+}
+
+// flushLoop runs periodic flush in background.
+func (agg *metricsAggregator) flushLoop() {
+	agg.flushTimer = time.NewTicker(agg.flushInterval)
+	defer agg.flushTimer.Stop()
+
+	for {
+		select {
+		case <-agg.flushTimer.C:
+			agg.flush(context.Background())
+		case <-agg.stopCh:
+			return
+		}
+	}
+}
+
+// flush flushes pending metrics to exporter.
+func (agg *metricsAggregator) flush(ctx context.Context) {
+	agg.mu.Lock()
+	defer agg.mu.Unlock()
+	agg.flushUnlocked(ctx)
+}
+
+// flushUnlocked flushes without locking (caller must hold lock).
+func (agg *metricsAggregator) flushUnlocked(ctx context.Context) {
+	if len(agg.batch) == 0 {
+		return
+	}
+
+	// Copy batch and clear
+	metrics := make([]*telemetryMetric, len(agg.batch))
+	copy(metrics, agg.batch)
+	agg.batch = agg.batch[:0]
+
+	// Export asynchronously
+	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				// Log at trace level only
+			}
+		}()
+		agg.exporter.export(ctx, metrics)
+	}()
+}
+
+// close stops the aggregator and flushes pending metrics.
+func (agg *metricsAggregator) close(ctx context.Context) error {
+	close(agg.stopCh)
+	agg.flush(ctx)
+	return nil
+}
+
+// simpleError is a simple error implementation for testing.
+type simpleError struct {
+	msg string
+}
+
+func (e *simpleError) Error() string {
+	return e.msg
+}

telemetry/client.go

@@ -1,8 +1,10 @@
 package telemetry
 
 import (
+	"context"
 	"net/http"
 	"sync"
+	"time"
 )
 
 // telemetryClient represents a client for sending telemetry data to Databricks.
@@ -11,44 +13,75 @@ import (
 // - One telemetryClient instance is shared across ALL connections to the same host
 // - This prevents rate limiting by consolidating telemetry from multiple connections
 // - The client MUST be fully thread-safe as it will be accessed concurrently
-// - All methods (start, close, and future export methods) must use proper synchronization
+// - All methods (start, close, and export methods) use proper synchronization
 //
-// The mu mutex protects the started and closed flags. Future implementations in Phase 4
-// will need to ensure thread-safety for batch operations and flushing.
-//
-// This is a minimal stub implementation that will be fully implemented in Phase 4.
+// The mu mutex protects the started and closed flags.
+// The aggregator handles thread-safe metric collection and batching.
 type telemetryClient struct {
 	host       string
 	httpClient *http.Client
 	cfg        *Config
+
+	exporter   *telemetryExporter
+	aggregator *metricsAggregator
+
 	mu         sync.Mutex // Protects started and closed flags
 	started    bool
 	closed     bool
 }
 
 // newTelemetryClient creates a new telemetry client for the given host.
 func newTelemetryClient(host string, httpClient *http.Client, cfg *Config) *telemetryClient {
+	// Create exporter
+	exporter := newTelemetryExporter(host, httpClient, cfg)
+
+	// Create aggregator with exporter
+	aggregator := newMetricsAggregator(exporter, cfg)
+
 	return &telemetryClient{
 		host:       host,
 		httpClient: httpClient,
 		cfg:        cfg,
+		exporter:   exporter,
+		aggregator: aggregator,
 	}
 }
 
 // start starts the telemetry client's background operations.
-// This is a stub implementation that will be fully implemented in Phase 4.
+// The aggregator starts its background flush timer automatically.
 func (c *telemetryClient) start() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
+
+	if c.started {
+		return nil
+	}
+
 	c.started = true
+	// Aggregator already started in newTelemetryClient
 	return nil
 }
 
 // close stops the telemetry client and flushes any pending data.
-// This is a stub implementation that will be fully implemented in Phase 4.
+// Provides graceful shutdown with a timeout to flush pending metrics.
 func (c *telemetryClient) close() error {
 	c.mu.Lock()
-	defer c.mu.Unlock()
+	if c.closed {
+		c.mu.Unlock()
+		return nil
+	}
 	c.closed = true
-	return nil
+	c.mu.Unlock()
+
+	// Flush pending metrics with timeout
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	return c.aggregator.close(ctx)
+}
+
+// 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 {
+	return newInterceptor(c.aggregator, enabled)
 }