feat(bigtable): first_response_latencies and connectivity_error_count metrics (#10616)

bhshkh · web-flow · commit 3054593d1f59 · 2025-08-28T13:05:19.000-04:00
* feat(bigtable): Adding first_response_latencies and connectivity_error_count metrics

* test(bigtable): Fixing tests
diff --git a/bigtable/bigtable.go b/bigtable/bigtable.go
@@ -1019,6 +1019,7 @@ func (t *Table) readRows(ctx context.Context, arg RowSet, f func(Row) bool, mt *
 		return errNegativeRowLimit
 	}
 
+	firstResponseRecorded := false
 	err = gaxInvokeWithRecorder(ctx, mt, methodNameReadRows, func(ctx context.Context, headerMD, trailerMD *metadata.MD, _ gax.CallSettings) error {
 		if rowLimitSet && numRowsRead >= intialRowLimit {
 			return nil
@@ -1071,6 +1072,10 @@ func (t *Table) readRows(ctx context.Context, arg RowSet, f func(Row) bool, mt *
 		for {
 			proto.Reset(res)
 			err := stream.RecvMsg(res)
+			if !firstResponseRecorded && (err == nil || err == io.EOF) {
+				firstResponseRecorded = true
+				mt.currOp.setFirstRespTime(time.Now())
+			}
 			if err == io.EOF {
 				*trailerMD = stream.Trailer()
 				break
diff --git a/bigtable/integration_test.go b/bigtable/integration_test.go
@@ -1008,6 +1008,7 @@ func TestIntegration_ExportBuiltInMetrics(t *testing.T) {
 		metricNameOperationLatencies,
 		metricNameAttemptLatencies,
 		metricNameServerLatencies,
+		metricNameFirstRespLatencies,
 	}
 
 	// Try for 5m with 10s sleep between retries
diff --git a/bigtable/metrics.go b/bigtable/metrics.go
@@ -423,6 +423,9 @@ type opTracer struct {
 
 	startTime time.Time
 
+	// Only for ReadRows. Time when the response headers are received in a streaming RPC.
+	firstRespTime time.Time
+
 	// gRPC status code of last completed attempt
 	status string
 
@@ -435,6 +438,10 @@ func (o *opTracer) setStartTime(t time.Time) {
 	o.startTime = t
 }
 
+func (o *opTracer) setFirstRespTime(t time.Time) {
+	o.firstRespTime = t
+}
+
 func (o *opTracer) setStatus(status string) {
 	o.status = status
 }
@@ -666,6 +673,13 @@ func (mt *builtinMetricsTracer) recordOperationCompletion() {
 	opLatAttrs, _ := mt.toOtelMetricAttrs(metricNameOperationLatencies)
 	mt.instrumentOperationLatencies.Record(mt.ctx, elapsedTimeMs, metric.WithAttributeSet(opLatAttrs))
 
+	// Record first_reponse_latencies
+	firstRespLatAttrs, _ := mt.toOtelMetricAttrs(metricNameFirstRespLatencies)
+	if mt.method == metricMethodPrefix+methodNameReadRows {
+		elapsedTimeMs = convertToMs(mt.currOp.firstRespTime.Sub(mt.currOp.startTime))
+		mt.instrumentFirstRespLatencies.Record(mt.ctx, elapsedTimeMs, metric.WithAttributeSet(firstRespLatAttrs))
+	}
+
 	// Record retry_count
 	retryCntAttrs, _ := mt.toOtelMetricAttrs(metricNameRetryCount)
 	if mt.currOp.attemptCount > 1 {
diff --git a/bigtable/metrics_test.go b/bigtable/metrics_test.go

Original file line number	Diff line number	Diff line change
`@@ -1008,6 +1008,7 @@ func TestIntegration_ExportBuiltInMetrics(t *testing.T) {`
`1008`	`1008`	`metricNameOperationLatencies,`
`1009`	`1009`	`metricNameAttemptLatencies,`
`1010`	`1010`	`metricNameServerLatencies,`
	`1011`	`+ metricNameFirstRespLatencies,`
`1011`	`1012`	`}`
`1012`	`1013`
`1013`	`1014`	`// Try for 5m with 10s sleep between retries`