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

* feat(bigtable): Adding first_response_latencies and connectivity_error_count metrics

* test(bigtable): Fixing tests

Author: bhshkh

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

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

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 {