feat(bigtable): Add application blocking latencies (#11349)

bhshkh · web-flow · commit 8f145ed17ceb · 2025-06-23T16:02:59.000-04:00
diff --git a/bigtable/bigtable.go b/bigtable/bigtable.go
@@ -1079,8 +1079,12 @@ func (t *Table) readRows(ctx context.Context, arg RowSet, f func(Row) bool, mt *
 					continue
 				}
 				prevRowKey = row.Key()
+
+				appBlockingLatencyStart := time.Now()
 				continueReading := f(row)
 				numRowsRead++
+				mt.currOp.incrementAppBlockingLatency(convertToMs(time.Since(appBlockingLatencyStart)))
+
 				if !continueReading {
 					// Cancel and drain stream.
 					cancel()
@@ -2246,6 +2250,10 @@ func recordOperationCompletion(mt *builtinMetricsTracer) {
 		// 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:
diff --git a/bigtable/metrics.go b/bigtable/metrics.go
@@ -57,12 +57,13 @@ const (
 	metricLabelKeyClientUID          = "client_uid"
 
 	// Metric names
-	metricNameOperationLatencies = "operation_latencies"
-	metricNameAttemptLatencies   = "attempt_latencies"
-	metricNameServerLatencies    = "server_latencies"
-	metricNameRetryCount         = "retry_count"
-	metricNameDebugTags          = "debug_tags"
-	metricNameConnErrCount       = "connectivity_error_count"
+	metricNameOperationLatencies   = "operation_latencies"
+	metricNameAttemptLatencies     = "attempt_latencies"
+	metricNameServerLatencies      = "server_latencies"
+	metricNameAppBlockingLatencies = "application_latencies"
+	metricNameRetryCount           = "retry_count"
+	metricNameDebugTags            = "debug_tags"
+	metricNameConnErrCount         = "connectivity_error_count"
 
 	// Metric units
 	metricUnitMS    = "ms"
@@ -108,6 +109,7 @@ var (
 			},
 			recordedPerAttempt: true,
 		},
+		metricNameAppBlockingLatencies: {},
 		metricNameRetryCount: {
 			additionalAttrs: []string{
 				metricLabelKeyStatus,
@@ -164,12 +166,13 @@ type builtinMetricsTracerFactory struct {
 	// do not change across different function calls on client
 	clientAttributes []attribute.KeyValue
 
-	operationLatencies metric.Float64Histogram
-	serverLatencies    metric.Float64Histogram
-	attemptLatencies   metric.Float64Histogram
-	retryCount         metric.Int64Counter
-	connErrCount       metric.Int64Counter
-	debugTags          metric.Int64Counter
+	operationLatencies   metric.Float64Histogram
+	serverLatencies      metric.Float64Histogram
+	attemptLatencies     metric.Float64Histogram
+	appBlockingLatencies metric.Float64Histogram
+	retryCount           metric.Int64Counter
+	connErrCount         metric.Int64Counter
+	debugTags            metric.Int64Counter
 }
 
 func newBuiltinMetricsTracerFactory(ctx context.Context, project, instance, appProfile string, metricsProvider MetricsProvider, opts ...option.ClientOption) (*builtinMetricsTracerFactory, error) {
@@ -269,6 +272,17 @@ func (tf *builtinMetricsTracerFactory) createInstruments(meter metric.Meter) err
 		return err
 	}
 
+	// Create application_latencies
+	tf.appBlockingLatencies, err = meter.Float64Histogram(
+		metricNameAppBlockingLatencies,
+		metric.WithDescription("The latency of the client application consuming available response data."),
+		metric.WithUnit(metricUnitMS),
+		metric.WithExplicitBucketBoundaries(bucketBounds...),
+	)
+	if err != nil {
+		return err
+	}
+
 	// Create retry_count
 	tf.retryCount, err = meter.Int64Counter(
 		metricNameRetryCount,
@@ -306,12 +320,13 @@ type builtinMetricsTracer struct {
 	// do not change across different operations on client
 	clientAttributes []attribute.KeyValue
 
-	instrumentOperationLatencies metric.Float64Histogram
-	instrumentServerLatencies    metric.Float64Histogram
-	instrumentAttemptLatencies   metric.Float64Histogram
-	instrumentRetryCount         metric.Int64Counter
-	instrumentConnErrCount       metric.Int64Counter
-	instrumentDebugTags          metric.Int64Counter
+	instrumentOperationLatencies   metric.Float64Histogram
+	instrumentServerLatencies      metric.Float64Histogram
+	instrumentAttemptLatencies     metric.Float64Histogram
+	instrumentAppBlockingLatencies metric.Float64Histogram
+	instrumentRetryCount           metric.Int64Counter
+	instrumentConnErrCount         metric.Int64Counter
+	instrumentDebugTags            metric.Int64Counter
 
 	tableName   string
 	method      string
@@ -336,6 +351,8 @@ type opTracer struct {
 	status string
 
 	currAttempt attemptTracer
+
+	appBlockingLatency float64
 }
 
 func (o *opTracer) setStartTime(t time.Time) {
@@ -350,6 +367,10 @@ func (o *opTracer) incrementAttemptCount() {
 	o.attemptCount++
 }
 
+func (o *opTracer) incrementAppBlockingLatency(latency float64) {
+	o.appBlockingLatency += latency
+}
+
 // attemptTracer is used to record metrics for each individual attempt of the operation.
 // Attempt corresponds to an attempt of an RPC.
 type attemptTracer struct {
@@ -404,12 +425,13 @@ func (tf *builtinMetricsTracerFactory) createBuiltinMetricsTracer(ctx context.Co
 		currOp:           currOpTracer,
 		clientAttributes: tf.clientAttributes,
 
-		instrumentOperationLatencies: tf.operationLatencies,
-		instrumentServerLatencies:    tf.serverLatencies,
-		instrumentAttemptLatencies:   tf.attemptLatencies,
-		instrumentRetryCount:         tf.retryCount,
-		instrumentConnErrCount:       tf.connErrCount,
-		instrumentDebugTags:          tf.debugTags,
+		instrumentOperationLatencies:   tf.operationLatencies,
+		instrumentServerLatencies:      tf.serverLatencies,
+		instrumentAttemptLatencies:     tf.attemptLatencies,
+		instrumentAppBlockingLatencies: tf.appBlockingLatencies,
+		instrumentRetryCount:           tf.retryCount,
+		instrumentConnErrCount:         tf.connErrCount,
+		instrumentDebugTags:            tf.debugTags,
 
 		tableName:   tableName,
 		isStreaming: isStreaming,
diff --git a/bigtable/metrics_test.go b/bigtable/metrics_test.go
