feat: Add client side metrics for readModifyWriteRow calls (#1656)

## Description

This PR collects client side metrics for readModifyWriteRow calls and collects all the right types of metrics for these unary calls. They leverage the plumbing provided for unary calls to add interceptors to those calls so that we can record metrics for the right zone/cluster/server time.

## Impact

This PR collects client side metrics for readModifyWriteRow calls.

## Testing

For each readRows/mutateRows/mutateRow test we add a corresponding test for readModifyWriteRow.

Author: danieljbruce

handwritten/bigtable/src/interceptor.ts …lient-side-metrics/metric-interceptor.tshandwritten/bigtable/src/interceptor.ts renamed to handwritten/bigtable/src/client-side-metrics/metric-interceptor.ts handwritten/bigtable/src/interceptor.ts renamed to handwritten/bigtable/src/client-side-metrics/metric-interceptor.ts

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 import {CallOptions} from 'google-gax';
-import {OperationMetricsCollector} from './client-side-metrics/operation-metrics-collector';
+import {OperationMetricsCollector} from './operation-metrics-collector';
 
 // Mock Server Implementation
 import * as grpcJs from '@grpc/grpc-js';
@@ -49,16 +49,21 @@ function createMetricsInterceptorProvider(
             collector.onStatusMetadataReceived(
               status as unknown as ServerStatus,
             );
+            collector.onAttemptComplete(status.code);
             nextStat(status);
           },
         };
         next(metadata, newListener);
       },
+      sendMessage: function (message, next) {
+        collector.onAttemptStart();
+        next(message);
+      },
     });
   };
 }
 
-export function withInterceptors(
+export function withMetricInterceptors(
   gaxOptions: CallOptions,
   metricsCollector?: OperationMetricsCollector,
 ) {

handwritten/bigtable/src/client-side-metrics/operation-metrics-collector.ts

@@ -286,8 +286,15 @@ export class OperationMetricsCollector {
     finalOperationStatus: grpc.status,
     applicationLatency?: number,
   ) {
-    this.onAttemptComplete(finalOperationStatus);
     withMetricsDebug(() => {
+      if (
+        this.state ===
+          MetricsCollectorState.OPERATION_STARTED_ATTEMPT_IN_PROGRESS_NO_ROWS_YET ||
+        this.state ===
+          MetricsCollectorState.OPERATION_STARTED_ATTEMPT_IN_PROGRESS_SOME_ROWS_RECEIVED
+      ) {
+        this.onAttemptComplete(finalOperationStatus);
+      }
       checkState(this.state, [
         MetricsCollectorState.OPERATION_STARTED_ATTEMPT_NOT_IN_PROGRESS,
       ]);

handwritten/bigtable/src/row-data-utils.ts

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+import {OperationMetricsCollector} from './client-side-metrics/operation-metrics-collector';
+
 const dotProp = require('dot-prop');
 import {Filter, RawFilter} from './filter';
 import {
@@ -30,6 +32,11 @@ import {TabularApiSurface} from './tabular-api-surface';
 import arrify = require('arrify');
 import {Bigtable} from './index';
 import {CallOptions} from 'google-gax';
+import {
+  MethodName,
+  StreamingState,
+} from './client-side-metrics/client-side-metrics-attributes';
+import {withMetricInterceptors} from './client-side-metrics/metric-interceptor';
 
 interface TabularApiSurfaceRequest {
   tableName?: string;
@@ -193,14 +200,31 @@ class RowDataUtils {
       properties.reqOpts,
     );
     properties.requestData.data = {};
+    // 1. Create a metrics collector.
+    const metricsCollector = new OperationMetricsCollector(
+      properties.requestData.table,
+      MethodName.READ_MODIFY_WRITE_ROW,
+      StreamingState.UNARY,
+      (
+        properties.requestData.table as any
+      ).bigtable._metricsConfigManager!.metricsHandlers,
+    );
+    // 2. Tell the metrics collector an attempt has been started.
+    metricsCollector.onOperationStart();
+    // 3. Make a unary call with gax options that include interceptors. The
+    // interceptors are built from a method that hooks them up to the
+    // metrics collector
     properties.requestData.bigtable.request<google.bigtable.v2.IReadModifyWriteRowResponse>(
       {
         client: 'BigtableClient',
         method: 'readModifyWriteRow',
         reqOpts,
-        gaxOpts: gaxOptions,
+        gaxOpts: withMetricInterceptors(gaxOptions, metricsCollector),
+      },
+      (err, ...args) => {
+        metricsCollector.onOperationComplete(err ? err.code : 0);
+        callback(err, ...args);
       },
-      callback,
     );
   }
 

handwritten/bigtable/src/utils/createReadStreamInternal.ts

@@ -420,7 +420,6 @@ export function createReadStreamInternal(
       })
       .on('end', () => {
         activeRequestStream = null;
-        const applicationLatency = userStream.getTotalDurationMs();
         metricsCollector.onOperationComplete(
           grpc.status.OK,
           userStream.getTotalDurationMs(),