feat(bigtable): add metrics exporter for outstandingrpcs and perconnectionerror count (googleapis#13510)

lift and shift of sushanb#3
```
curl -L -O https://github.com/sushanb/google-cloud-go/commit/b9d20984c514b4632f7c613e4d6ab58b1776a939.patch
git apply b9d2098.patch
```

Author: sushanb

bigtable/bigtable.go

@@ -203,6 +203,8 @@ func NewClientWithConfig(ctx context.Context, project, instance string, config C
 			btransport.WithInstanceName(fullInstanceName),
 			btransport.WithAppProfile(config.AppProfile),
 			btransport.WithFeatureFlagsMetadata(ffMD),
+			btransport.WithMetricsReporterConfig(btopt.DefaultMetricsReporterConfig()),
+			btransport.WithMeterProvider(metricsTracerFactory.otelMeterProvider),
 		)
 
 		if err != nil {

bigtable/internal/option/option.go

@@ -252,3 +252,18 @@ func DefaultDynamicChannelPoolConfig() DynamicChannelPoolConfig {
 		MaxRemoveConns:       2, // Only Cap for removals
 	}
 }
+
+// MetricsReporterConfig for periodic reporting
+// MetricsReporterConfig holds the parameters for metrics reporting.
+type MetricsReporterConfig struct {
+	Enabled           bool
+	ReportingInterval time.Duration
+}
+
+// DefaultMetricsReporterConfig with defaults used.
+func DefaultMetricsReporterConfig() MetricsReporterConfig {
+	return MetricsReporterConfig{
+		Enabled:           true,
+		ReportingInterval: 1 * time.Minute,
+	}
+}

bigtable/internal/transport/connpool.go

@@ -50,6 +50,20 @@ const requestParamsHeader = "x-goog-request-params"
 // BigtableChannelPoolOption options for configurable
 type BigtableChannelPoolOption func(*BigtableChannelPool)
 
+// connPoolStatsSupplier callback  that returns a snapshot of connection pool statistics.
+type connPoolStatsSupplier func() []connPoolStats
+
+// connPoolStats holds a snapshot of statistics for a single connection.
+type connPoolStats struct {
+	OutstandingUnaryLoad     int32
+	OutstandingStreamingLoad int32
+	ErrorCount               int64
+	IsALTSUsed               bool
+	LBPolicy                 string
+}
+
+var _ Monitor = (*MetricsReporter)(nil)
+
 // WithAppProfile provides the appProfile
 func WithAppProfile(appProfile string) BigtableChannelPoolOption {
 	return func(p *BigtableChannelPool) {
@@ -131,6 +145,28 @@ func (bc *BigtableConn) Prime(ctx context.Context, fullInstanceName, appProfileI
 	return nil
 }
 
+// connPoolStatsSupplier returns a snapshot of the current connection pool statistics.
+func (p *BigtableChannelPool) connPoolStatsSupplier() []connPoolStats {
+	conns := p.getConns()
+	if len(conns) == 0 {
+		return nil
+	}
+
+	stats := make([]connPoolStats, len(conns))
+	lbPolicy := p.strategy.String()
+
+	for i, entry := range conns {
+		stats[i] = connPoolStats{
+			OutstandingUnaryLoad:     entry.unaryLoad.Load(),
+			OutstandingStreamingLoad: entry.streamingLoad.Load(),
+			ErrorCount:               entry.errorCount.Swap(0),
+			IsALTSUsed:               entry.isALTSUsed(),
+			LBPolicy:                 lbPolicy,
+		}
+	}
+	return stats
+}
+
 // NewBigtableConn creates a wrapped grpc Client Conn
 func NewBigtableConn(conn *grpc.ClientConn) *BigtableConn {
 	bc := &BigtableConn{
@@ -239,6 +275,16 @@ type BigtableChannelPool struct {
 	instanceName   string
 	featureFlagsMD metadata.MD
 	meterProvider  metric.MeterProvider
+	// configs
+	metricsConfig btopt.MetricsReporterConfig
+
+	// background monitors
+	monitors []Monitor
+}
+
+// WithMetricsReporterConfig attaches the relevant config for exporting the metrics
+func WithMetricsReporterConfig(config btopt.MetricsReporterConfig) BigtableChannelPoolOption {
+	return func(p *BigtableChannelPool) { p.metricsConfig = config }
 }
 
 // getConns safely loads the current slice of connections.
@@ -327,9 +373,27 @@ func NewBigtableChannelPool(ctx context.Context, connPoolSize int, strategy btop
 	}
 
 	pool.conns.Store(&initialConns)
+
+	btopt.Debugf(pool.logger, "bigtable_connpool: using load balancing strategy: %s\n", strategy)
+
+	metricsReporter, err := NewMetricsReporter(pool.metricsConfig, pool.connPoolStatsSupplier, pool.logger, pool.meterProvider)
+	if err == nil {
+		// ignore
+		pool.monitors = append(pool.monitors, metricsReporter)
+	} else {
+		btopt.Debugf(pool.logger, "bigtable_connpool: failed to create metrics reporter: %v\n", err)
+	}
+	pool.startMonitors()
 	return pool, nil
 }
 
+func (p *BigtableChannelPool) startMonitors() {
+	for _, m := range p.monitors {
+		btopt.Debugf(p.logger, "bigtable_connpool: Starting monitor %T\n", m)
+		m.Start(p.poolCtx)
+	}
+}
+
 // Num returns the number of connections in the pool.
 func (p *BigtableChannelPool) Num() int {
 	return len(p.getConns())
@@ -338,6 +402,10 @@ func (p *BigtableChannelPool) Num() int {
 // Close closes all connections in the pool.
 func (p *BigtableChannelPool) Close() error {
 	p.poolCancel() // Cancel the context for background tasks
+	// Stop all monitors.
+	for _, m := range p.monitors {
+		m.Stop()
+	}
 	conns := p.getConns()
 	var errs multiError
 

bigtable/internal/transport/connpool_test.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"io"
 	"net/url"
+	"reflect"
 	"sort"
 	"strings"
 	"sync"
@@ -1391,6 +1392,87 @@ func TestRemoveConnections(t *testing.T) {
 	})
 }
 
+func TestConnPoolStatisticsVisitor(t *testing.T) {
+	ctx := context.Background()
+	poolSize := 3
+	fake := &fakeService{}
+	addr := setupTestServer(t, fake)
+	dialFunc := func() (*BigtableConn, error) { return dialBigtableserver(addr) }
+
+	pool, err := NewBigtableChannelPool(ctx, poolSize, btopt.RoundRobin, dialFunc, poolOpts()...)
+	if err != nil {
+		t.Fatalf("Failed to create pool: %v", err)
+	}
+	defer pool.Close()
+
+	// Wait for connections to be established
+	time.Sleep(100 * time.Millisecond)
+
+	conns := pool.getConns()
+	if len(conns) != poolSize {
+		t.Fatalf("Pool size mismatch: got %d, want %d", len(conns), poolSize)
+	}
+
+	testData := []struct {
+		unary     int32
+		streaming int32
+		errors    int64
+		isALTS    bool
+	}{
+		{unary: 5, streaming: 2, errors: 10, isALTS: false},
+		{unary: 0, streaming: 1, errors: 0, isALTS: true},
+		{unary: 3, streaming: 0, errors: 5, isALTS: false},
+	}
+
+	if len(testData) != poolSize {
+		t.Fatalf("Test data size: pool size mismatch: got %d, want %d", len(testData), poolSize)
+	}
+
+	for i, data := range testData {
+		if i < len(conns) {
+			conns[i].unaryLoad.Store(data.unary)
+			conns[i].streamingLoad.Store(data.streaming)
+			conns[i].errorCount.Store(data.errors)
+			conns[i].conn.isALTSConn.Store(data.isALTS)
+		}
+	}
+
+	// Get the snapshot
+	stats := pool.connPoolStatsSupplier()
+
+	if len(stats) != poolSize {
+		t.Errorf("Snapshot size mismatch: got %d, want %d", len(stats), poolSize)
+	}
+
+	expectedStats := make([]connPoolStats, poolSize)
+	for i, data := range testData {
+		expectedStats[i] = connPoolStats{
+			OutstandingUnaryLoad:     data.unary,
+			OutstandingStreamingLoad: data.streaming,
+			ErrorCount:               data.errors,
+			IsALTSUsed:               data.isALTS,
+			LBPolicy:                 btopt.RoundRobin.String(),
+		}
+	}
+
+	sort.Slice(stats, func(i, j int) bool { return stats[i].OutstandingUnaryLoad < stats[j].OutstandingUnaryLoad })
+	sort.Slice(expectedStats, func(i, j int) bool {
+		return expectedStats[i].OutstandingUnaryLoad < expectedStats[j].OutstandingUnaryLoad
+	})
+
+	if !reflect.DeepEqual(stats, expectedStats) {
+		t.Errorf("Snapshot data mismatch:\ngot:  %v\nwant: %v", stats, expectedStats)
+	}
+
+	// Verify error counts are reset
+	connsAfter := pool.getConns()
+	for i, entry := range connsAfter {
+		if entry.errorCount.Load() != 0 {
+			t.Errorf("entry[%d].errorCount was not reset: got %d, want 0", i, entry.errorCount.Load())
+		}
+	}
+}
+
 // --- Benchmarks ---
 
 func createBenchmarkFake() *fakeService {

bigtable/internal/transport/metrics_reporter.go

@@ -0,0 +1,157 @@
+// Copyright 2025 Google LLC
+//
+// 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.
+
+package internal
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	btopt "cloud.google.com/go/bigtable/internal/option"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+const (
+	// outstandingRPCsMetricName is the name for the outstanding RPCs histogram.
+	outstandingRPCsMetricName = "connection_pool/outstanding_rpcs"
+	// perConnErrorCountMetricName is the name for the per-connection error count histogram.
+	perConnErrorCountMetricName = "per_connection_error_count"
+	// clientMeterName is the prefix for metrics name
+	clientMeterName = "bigtable.googleapis.com/internal/client/"
+)
+
+// MetricsReporter periodically collects and reports metrics for the connection pool.
+type MetricsReporter struct {
+	config                btopt.MetricsReporterConfig
+	connPoolStatsSupplier connPoolStatsSupplier
+	logger                *log.Logger
+	ticker                *time.Ticker
+	done                  chan struct{}
+	stopOnce              sync.Once
+
+	// OpenTelemetry metric instruments
+	meterProvider                    metric.MeterProvider
+	outstandingRPCsHistogram         metric.Float64Histogram
+	perConnectionErrorCountHistogram metric.Float64Histogram
+}
+
+// NewMetricsReporter starts a monitor to export periodic metrics
+func NewMetricsReporter(config btopt.MetricsReporterConfig, connPoolStatsSupplier connPoolStatsSupplier, logger *log.Logger, mp metric.MeterProvider) (*MetricsReporter, error) {
+	mr := &MetricsReporter{
+		config:                config,
+		connPoolStatsSupplier: connPoolStatsSupplier,
+		logger:                logger,
+		done:                  make(chan struct{}),
+		meterProvider:         mp,
+	}
+
+	if !config.Enabled {
+		// Stats Disabled, return error during ctor
+		return nil, fmt.Errorf("bigtable_connpool: MetricsReporterConfig.Enabled is false")
+	}
+
+	if mp == nil {
+		// State: Stats Enabled but MeterProvider is nil, return error during ctor
+		return nil, fmt.Errorf("bigtable_connpool: MetricsReporterConfig.Enabled is true, but MeterProvider is nil")
+	}
+
+	// create meter
+	meter := mp.Meter(clientMeterName)
+	var err error
+
+	// fail if meter cannot be created.
+	mr.outstandingRPCsHistogram, err = meter.Float64Histogram(outstandingRPCsMetricName, metric.WithDescription("Distribution of outstanding RPCs per connection."), metric.WithUnit("1"))
+	if err != nil {
+		return nil, fmt.Errorf("bigtable_connpool: failed to create %s histogram: %w", outstandingRPCsMetricName, err)
+	}
+
+	mr.perConnectionErrorCountHistogram, err = meter.Float64Histogram(perConnErrorCountMetricName, metric.WithDescription("Distribution of errors per connection per minute."), metric.WithUnit("1"))
+	if err != nil {
+		return nil, fmt.Errorf("bigtable_connpool: failed to create %s histogram: %w", perConnErrorCountMetricName, err)
+	}
+
+	return mr, nil
+}
+
+// Start starts the reporter.
+func (mr *MetricsReporter) Start(ctx context.Context) {
+	// config.Enabled should ensure that all relevant thing (meterProvider and meter should exists)
+	if !mr.config.Enabled {
+		return
+	}
+
+	mr.ticker = time.NewTicker(mr.config.ReportingInterval)
+	go func() {
+		defer mr.ticker.Stop()
+		for {
+			select {
+			case <-mr.ticker.C:
+				mr.snapshotAndRecordMetrics(ctx)
+			case <-mr.done:
+				return
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+}
+
+// Stop stops the reporter gracefully
+func (mr *MetricsReporter) Stop() {
+	mr.stopOnce.Do(func() {
+		if mr.config.Enabled {
+			close(mr.done)
+		}
+	})
+}
+
+// snapshotAndRecordMetrics collects and records metrics for the current state of the connection pool.
+func (mr *MetricsReporter) snapshotAndRecordMetrics(ctx context.Context) {
+	stats := mr.connPoolStatsSupplier()
+	if len(stats) == 0 {
+		return
+	}
+
+	for _, stat := range stats {
+		// Record per-connection error count for the interval
+		if mr.perConnectionErrorCountHistogram != nil {
+			mr.perConnectionErrorCountHistogram.Record(ctx, float64(stat.ErrorCount))
+		}
+		transportType := "cloudpath"
+		if stat.IsALTSUsed {
+			transportType = "directpath"
+		}
+
+		// Common attributes for this connection
+		baseAttrs := []attribute.KeyValue{
+			attribute.String("transport_type", transportType),
+			attribute.String("lb_policy", stat.LBPolicy),
+		}
+
+		if mr.outstandingRPCsHistogram != nil {
+			// Record distribution sample for unary load
+			unaryAttrs := attribute.NewSet(append(baseAttrs, attribute.Bool("streaming", false))...)
+			mr.outstandingRPCsHistogram.Record(ctx, float64(stat.OutstandingUnaryLoad), metric.WithAttributeSet(unaryAttrs))
+
+			// Record distribution sample for streaming load
+			streamingAttrs := attribute.NewSet(append(baseAttrs, attribute.Bool("streaming", true))...)
+			mr.outstandingRPCsHistogram.Record(ctx, float64(stat.OutstandingStreamingLoad), metric.WithAttributeSet(streamingAttrs))
+		}
+
+	}
+}