fix(bigtable): Fix NoopMetricsProvider panic (#12709)

bhshkh · web-flow · commit 8c6da0dd99ce · 2025-08-14T11:59:47.000-07:00
diff --git a/bigtable/bigtable.go b/bigtable/bigtable.go
@@ -36,8 +36,6 @@ import (
 	"cloud.google.com/go/internal/trace"
 	gax "github.com/googleapis/gax-go/v2"
 	"github.com/googleapis/gax-go/v2/apierror"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/metric"
 	"google.golang.org/api/option"
 	"google.golang.org/api/option/internaloption"
 	gtransport "google.golang.org/api/transport/grpc"
@@ -140,19 +138,7 @@ func NewClientWithConfig(ctx context.Context, project, instance string, config C
 	o = append(o, opts...)
 
 	// TODO(b/372244283): Remove after b/358175516 has been fixed
-	asyncRefreshMetricAttrs := metricsTracerFactory.clientAttributes
-	asyncRefreshMetricAttrs = append(asyncRefreshMetricAttrs,
-		attribute.String(metricLabelKeyTag, "async_refresh_dry_run"),
-		// Table, cluster and zone are unknown at this point
-		// Use default values
-		attribute.String(monitoredResLabelKeyTable, defaultTable),
-		attribute.String(monitoredResLabelKeyCluster, defaultCluster),
-		attribute.String(monitoredResLabelKeyZone, defaultZone),
-	)
-	o = append(o, internaloption.EnableAsyncRefreshDryRun(func() {
-		metricsTracerFactory.debugTags.Add(context.Background(), 1,
-			metric.WithAttributes(asyncRefreshMetricAttrs...))
-	}))
+	o = append(o, internaloption.EnableAsyncRefreshDryRun(metricsTracerFactory.newAsyncRefreshErrHandler()))
 
 	connPool, err := gtransport.DialPool(ctx, o...)
 	if err != nil {
@@ -528,11 +514,11 @@ func (c *Client) prepareStatementWithMetadata(ctx context.Context, query string,
 	defer func() { trace.EndSpan(ctx, err) }()
 
 	mt := c.newBuiltinMetricsTracer(ctx, "", false)
-	defer recordOperationCompletion(mt)
+	defer mt.recordOperationCompletion()
 
 	preparedStatement, err = c.prepareStatement(ctx, mt, query, paramTypes, opts...)
 	statusCode, statusErr := convertToGrpcStatusErr(err)
-	mt.currOp.setStatus(statusCode.String())
+	mt.setCurrOpStatus(statusCode.String())
 	return preparedStatement, statusErr
 }
 
@@ -707,11 +693,11 @@ func (bs *BoundStatement) Execute(ctx context.Context, f func(ResultRow) bool, o
 	defer func() { trace.EndSpan(ctx, err) }()
 
 	mt := bs.ps.c.newBuiltinMetricsTracer(ctx, "", true)
-	defer recordOperationCompletion(mt)
+	defer mt.recordOperationCompletion()
 
 	err = bs.execute(ctx, f, mt)
 	statusCode, statusErr := convertToGrpcStatusErr(err)
-	mt.currOp.setStatus(statusCode.String())
+	mt.setCurrOpStatus(statusCode.String())
 	return statusErr
 }
 
@@ -965,11 +951,11 @@ func (t *Table) ReadRows(ctx context.Context, arg RowSet, f func(Row) bool, opts
 	defer func() { trace.EndSpan(ctx, err) }()
 
 	mt := t.newBuiltinMetricsTracer(ctx, true)
-	defer recordOperationCompletion(mt)
+	defer mt.recordOperationCompletion()
 
 	err = t.readRows(ctx, arg, f, mt, opts...)
 	statusCode, statusErr := convertToGrpcStatusErr(err)
-	mt.currOp.setStatus(statusCode.String())
+	mt.setCurrOpStatus(statusCode.String())
 	return statusErr
 }
 
@@ -1082,7 +1068,7 @@ func (t *Table) readRows(ctx context.Context, arg RowSet, f func(Row) bool, mt *
 				appBlockingLatencyStart := time.Now()
 				continueReading := f(row)
 				numRowsRead++
-				mt.currOp.incrementAppBlockingLatency(convertToMs(time.Since(appBlockingLatencyStart)))
+				mt.incrementAppBlockingLatency(convertToMs(time.Since(appBlockingLatencyStart)))
 
 				if !continueReading {
 					// Cancel and drain stream.
@@ -1640,11 +1626,11 @@ func (t *Table) Apply(ctx context.Context, row string, m *Mutation, opts ...Appl
 	ctx = trace.StartSpan(ctx, "cloud.google.com/go/bigtable/Apply")
 	defer func() { trace.EndSpan(ctx, err) }()
 	mt := t.newBuiltinMetricsTracer(ctx, false)
-	defer recordOperationCompletion(mt)
+	defer mt.recordOperationCompletion()
 
 	err = t.apply(ctx, mt, row, m, opts...)
 	statusCode, statusErr := convertToGrpcStatusErr(err)
-	mt.currOp.setStatus(statusCode.String())
+	mt.setCurrOpStatus(statusCode.String())
 	return statusErr
 }
 
@@ -1918,7 +1904,7 @@ func (t *Table) ApplyBulk(ctx context.Context, rowKeys []string, muts []*Mutatio
 func (t *Table) applyGroup(ctx context.Context, group []*entryErr, opts ...ApplyOption) (err error) {
 	attrMap := make(map[string]interface{})
 	mt := t.newBuiltinMetricsTracer(ctx, true)
-	defer recordOperationCompletion(mt)
+	defer mt.recordOperationCompletion()
 
 	err = gaxInvokeWithRecorder(ctx, mt, "MutateRows", func(ctx context.Context, headerMD, trailerMD *metadata.MD, _ gax.CallSettings) error {
 		attrMap["rowCount"] = len(group)
@@ -1939,7 +1925,7 @@ func (t *Table) applyGroup(ctx context.Context, group []*entryErr, opts ...Apply
 	}, t.c.retryOption)
 
 	statusCode, statusErr := convertToGrpcStatusErr(err)
-	mt.currOp.setStatus(statusCode.String())
+	mt.setCurrOpStatus(statusCode.String())
 	return statusErr
 }
 
@@ -2080,11 +2066,11 @@ func (t *Table) ApplyReadModifyWrite(ctx context.Context, row string, m *ReadMod
 	ctx = mergeOutgoingMetadata(ctx, t.md)
 
 	mt := t.newBuiltinMetricsTracer(ctx, false)
-	defer recordOperationCompletion(mt)
+	defer mt.recordOperationCompletion()
 
 	updatedRow, err := t.applyReadModifyWrite(ctx, mt, row, m)
 	statusCode, statusErr := convertToGrpcStatusErr(err)
-	mt.currOp.setStatus(statusCode.String())
+	mt.setCurrOpStatus(statusCode.String())
 	return updatedRow, statusErr
 }
 
@@ -2161,11 +2147,11 @@ func (t *Table) SampleRowKeys(ctx context.Context) ([]string, error) {
 	ctx = mergeOutgoingMetadata(ctx, t.md)
 
 	mt := t.newBuiltinMetricsTracer(ctx, true)
-	defer recordOperationCompletion(mt)
+	defer mt.recordOperationCompletion()
 
 	rowKeys, err := t.sampleRowKeys(ctx, mt)
 	statusCode, statusErr := convertToGrpcStatusErr(err)
-	mt.currOp.setStatus(statusCode.String())
+	mt.setCurrOpStatus(statusCode.String())
 	return rowKeys, statusErr
 }
 
@@ -2227,34 +2213,6 @@ func (c *Client) newBuiltinMetricsTracer(ctx context.Context, table string, isSt
 	return &mt
 }
 
-// recordOperationCompletion records as many operation specific metrics as it can
-// Ignores error seen while creating metric attributes since metric can still
-// be recorded with rest of the attributes
-func recordOperationCompletion(mt *builtinMetricsTracer) {
-	if !mt.builtInEnabled {
-		return
-	}
-
-	// Calculate elapsed time
-	elapsedTimeMs := convertToMs(time.Since(mt.currOp.startTime))
-
-	// Record operation_latencies
-	opLatAttrs, _ := mt.toOtelMetricAttrs(metricNameOperationLatencies)
-	mt.instrumentOperationLatencies.Record(mt.ctx, elapsedTimeMs, metric.WithAttributeSet(opLatAttrs))
-
-	// Record retry_count
-	retryCntAttrs, _ := mt.toOtelMetricAttrs(metricNameRetryCount)
-	if mt.currOp.attemptCount > 1 {
-		// Only record when retry count is greater than 0 so the retry
-		// graph will be less confusing
-		mt.instrumentRetryCount.Add(mt.ctx, mt.currOp.attemptCount-1, metric.WithAttributeSet(retryCntAttrs))
-	}
-
-	// Record application_latencies
-	appBlockingLatAttrs, _ := mt.toOtelMetricAttrs(metricNameAppBlockingLatencies)
-	mt.instrumentAppBlockingLatencies.Record(mt.ctx, mt.currOp.appBlockingLatency, metric.WithAttributeSet(appBlockingLatAttrs))
-}
-
 // gaxInvokeWithRecorder:
 // - wraps 'f' in a new function 'callWrapper' that:
 //   - updates tracer state and records built in attempt specific metrics
@@ -2267,82 +2225,16 @@ func gaxInvokeWithRecorder(ctx context.Context, mt *builtinMetricsTracer, method
 	attempTrailerMD := metadata.New(nil)
 	mt.setMethod(method)
 
-	var callWrapper func(context.Context, gax.CallSettings) error
-	if !mt.builtInEnabled {
-		callWrapper = func(ctx context.Context, callSettings gax.CallSettings) error {
-			// f makes calls to CBT service
-			return f(ctx, &attemptHeaderMD, &attempTrailerMD, callSettings)
-		}
-	} else {
-		callWrapper = func(ctx context.Context, callSettings gax.CallSettings) error {
-			// Increment number of attempts
-			mt.currOp.incrementAttemptCount()
-
-			mt.currOp.currAttempt = attemptTracer{}
-
-			// record start time
-			mt.currOp.currAttempt.setStartTime(time.Now())
-
-			// f makes calls to CBT service
-			err := f(ctx, &attemptHeaderMD, &attempTrailerMD, callSettings)
-
-			// Set attempt status
-			statusCode, _ := convertToGrpcStatusErr(err)
-			mt.currOp.currAttempt.setStatus(statusCode.String())
+	callWrapper := func(ctx context.Context, callSettings gax.CallSettings) error {
+		mt.recordAttemptStart()
 
-			// Get location attributes from metadata and set it in tracer
-			// Ignore get location error since the metric can still be recorded with rest of the attributes
-			clusterID, zoneID, _ := extractLocation(attemptHeaderMD, attempTrailerMD)
-			mt.currOp.currAttempt.setClusterID(clusterID)
-			mt.currOp.currAttempt.setZoneID(zoneID)
+		// f makes calls to CBT service
+		err := f(ctx, &attemptHeaderMD, &attempTrailerMD, callSettings)
 
-			// Set server latency in tracer
-			serverLatency, serverLatencyErr := extractServerLatency(attemptHeaderMD, attempTrailerMD)
-			mt.currOp.currAttempt.setServerLatencyErr(serverLatencyErr)
-			mt.currOp.currAttempt.setServerLatency(serverLatency)
-
-			// Record attempt specific metrics
-			recordAttemptCompletion(mt)
-			return err
-		}
-	}
-	return gax.Invoke(ctx, callWrapper, opts...)
-}
-
-// recordAttemptCompletion records as many attempt specific metrics as it can
-// Ignore errors seen while creating metric attributes since metric can still
-// be recorded with rest of the attributes
-func recordAttemptCompletion(mt *builtinMetricsTracer) {
-	if !mt.builtInEnabled {
-		return
-	}
-
-	// Calculate elapsed time
-	elapsedTime := convertToMs(time.Since(mt.currOp.currAttempt.startTime))
-
-	// Record attempt_latencies
-	attemptLatAttrs, _ := mt.toOtelMetricAttrs(metricNameAttemptLatencies)
-	mt.instrumentAttemptLatencies.Record(mt.ctx, elapsedTime, metric.WithAttributeSet(attemptLatAttrs))
-
-	// Record server_latencies
-	serverLatAttrs, _ := mt.toOtelMetricAttrs(metricNameServerLatencies)
-	if mt.currOp.currAttempt.serverLatencyErr == nil {
-		mt.instrumentServerLatencies.Record(mt.ctx, mt.currOp.currAttempt.serverLatency, metric.WithAttributeSet(serverLatAttrs))
+		// Record attempt specific metrics
+		mt.recordAttemptCompletion(attemptHeaderMD, attempTrailerMD, err)
+		return err
 	}
 
-	// Record connectivity_error_count
-	connErrCountAttrs, _ := mt.toOtelMetricAttrs(metricNameConnErrCount)
-	// Determine if connection error should be incremented.
-	// A true connectivity error occurs only when we receive NO server-side signals.
-	// 1. Server latency (from server-timing header) is a signal, but absent in DirectPath.
-	// 2. Location (from x-goog-ext header) is a signal present in both paths.
-	// Therefore, we only count an error if BOTH signals are missing.
-	isServerLatencyEffectivelyEmpty := mt.currOp.currAttempt.serverLatencyErr != nil || mt.currOp.currAttempt.serverLatency == 0
-	isLocationEmpty := mt.currOp.currAttempt.clusterID == defaultCluster
-	if isServerLatencyEffectivelyEmpty && isLocationEmpty {
-		// This is a connectivity error: the request likely never reached Google's network.
-		mt.instrumentConnErrCount.Add(mt.ctx, 1, metric.WithAttributeSet(connErrCountAttrs))
-	} else {
-		mt.instrumentConnErrCount.Add(mt.ctx, 0, metric.WithAttributeSet(connErrCountAttrs))
-	}
+	return gax.Invoke(ctx, callWrapper, opts...)
 }
diff --git a/bigtable/integration_test.go b/bigtable/integration_test.go
@@ -896,23 +896,54 @@ func TestIntegration_HighlyConcurrentReadsAndWrites(t *testing.T) {
 	wg.Wait()
 }
 
-func TestIntegration_ExportBuiltInMetrics(t *testing.T) {
+func TestIntegration_NoopMetricsProvider(t *testing.T) {
 	ctx := context.Background()
+	testEnv, _, adminClient, _, tableName, cleanup, err := setupIntegration(ctx, t)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cleanup()
 
-	// Reduce sampling period for faster test runs
-	origSamplePeriod := defaultSamplePeriod
-	defaultSamplePeriod = time.Minute
-	defer func() {
-		defaultSamplePeriod = origSamplePeriod
-	}()
+	if testing.Short() || !testEnv.Config().UseProd {
+		t.Skip("Skip long running tests in short mode or non-prod environments")
+	}
+
+	family := "export"
+	if err := createColumnFamily(ctx, t, adminClient, tableName, family, nil); err != nil {
+		t.Fatalf("Creating column family: %v", err)
+	}
+
+	noopClient, err := testEnv.NewClientWithConfig(ClientConfig{MetricsProvider: NoopMetricsProvider{}})
+	if err != nil {
+		t.Fatalf("NewClientWithConfig: %v", err)
+	}
 
+	noopTable := noopClient.Open(tableName)
+	for i := 0; i < 10; i++ {
+		mut := NewMutation()
+		mut.Set(family, "col", 1000, []byte("test"))
+		if err := noopTable.Apply(ctx, fmt.Sprintf("row-%v", i), mut); err != nil {
+			t.Fatalf("Apply: %v", err)
+		}
+	}
+
+	err = noopTable.ReadRows(ctx, PrefixRange("row-"), func(r Row) bool {
+		return true
+	}, RowFilter(ColumnFilter("col")))
+	if err != nil {
+		t.Fatalf("ReadRows: %v", err)
+	}
+}
+
+func TestIntegration_ExportBuiltInMetrics(t *testing.T) {
 	// record start time
 	testStartTime := time.Now()
 	tsListStart := &timestamppb.Timestamp{
 		Seconds: testStartTime.Unix(),
 		Nanos:   int32(testStartTime.Nanosecond()),
 	}
 
+	ctx := context.Background()
 	testEnv, _, adminClient, table, tableName, cleanup, err := setupIntegration(ctx, t)
 	if err != nil {
 		t.Fatal(err)
@@ -935,6 +966,7 @@ func TestIntegration_ExportBuiltInMetrics(t *testing.T) {
 			t.Fatalf("Apply: %v", err)
 		}
 	}
+
 	err = table.ReadRows(ctx, PrefixRange("row-"), func(r Row) bool {
 		return true
 	}, RowFilter(ColumnFilter("col")))
@@ -6126,6 +6158,9 @@ func setupIntegration(ctx context.Context, t *testing.T) (_ IntegrationEnv, _ *C
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	_ = cancel // ignore for test
 
+	// Reduce sampling period for faster test runs
+	origSamplePeriod := defaultSamplePeriod
+	defaultSamplePeriod = time.Minute
 	client, err := testEnv.NewClient()
 	if err != nil {
 		t.Logf("Error creating client: %v", err)
@@ -6185,6 +6220,7 @@ func setupIntegration(ctx context.Context, t *testing.T) (_ IntegrationEnv, _ *C
 	}
 
 	return testEnv, client, adminClient, client.Open(tableName), tableName, func() {
+		defaultSamplePeriod = origSamplePeriod
 		if err := deleteTable(ctx, t, adminClient, tableName); err != nil {
 			t.Errorf("DeleteTable got error %v", err)
 		}
diff --git a/bigtable/metrics.go b/bigtable/metrics.go
