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

src/interceptor.ts …lient-side-metrics/metric-interceptor.tssrc/interceptor.ts renamed to src/client-side-metrics/metric-interceptor.ts src/interceptor.ts renamed to 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,
 ) {

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,
  ]);

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,
  );
  }
 

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(),