Pass in metrics collector

Author: danieljbruce

src/row.ts

@@ -32,6 +32,10 @@ import {google} from '../protos/protos';
 import {RowDataUtils, RowProperties} from './row-data-utils';
 import {TabularApiSurface} from './tabular-api-surface';
 import {getRowsInternal} from './utils/getRowsInternal';
+import {
+  MethodName,
+  StreamingState,
+} from './client-side-metrics/client-side-metrics-attributes';
 
 export interface Rule {
   column: string;
@@ -667,31 +671,42 @@ export class Row {
       filter,
     });
 
-    void getRowsInternal(this.table, true, getRowsOptions, (err, rows) => {
-      if (err) {
-        callback(err);
-        return;
-      }
+    const metricsCollector =
+      this.table.bigtable._metricsConfigManager.createOperation(
+        MethodName.READ_ROW,
+        StreamingState.UNARY,
+        this.table,
+      );
+    void getRowsInternal(
+      this.table,
+      metricsCollector,
+      getRowsOptions,
+      (err, rows) => {
+        if (err) {
+          callback(err);
+          return;
+        }
 
-      const row = rows![0];
+        const row = rows![0];
 
-      if (!row) {
-        const e = new RowError(this.id);
-        callback(e);
-        return;
-      }
+        if (!row) {
+          const e = new RowError(this.id);
+          callback(e);
+          return;
+        }
 
-      this.data = row.data;
+        this.data = row.data;
 
-      // If the user specifies column names, we'll return back the row data
-      // we received. Otherwise, we'll return the row "this" in a typical
-      // GrpcServiceObject#get fashion.
-      if (columns.length > 0) {
-        callback(null, row.data);
-      } else {
-        (callback as {} as GetRowCallback<Row>)(null, this);
-      }
-    });
+        // If the user specifies column names, we'll return back the row data
+        // we received. Otherwise, we'll return the row "this" in a typical
+        // GrpcServiceObject#get fashion.
+        if (columns.length > 0) {
+          callback(null, row.data);
+        } else {
+          (callback as {} as GetRowCallback<Row>)(null, this);
+        }
+      },
+    );
   }
 
   getMetadata(options?: GetRowOptions): Promise<GetRowMetadataResponse>;

src/tabular-api-surface.ts

@@ -33,6 +33,10 @@ import * as is from 'is';
 import {GoogleInnerError} from './table';
 import {createReadStreamInternal} from './utils/createReadStreamInternal';
 import {getRowsInternal} from './utils/getRowsInternal';
+import {
+  MethodName,
+  StreamingState,
+} from './client-side-metrics/client-side-metrics-attributes';
 
 // See protos/google/rpc/code.proto
 // (4=DEADLINE_EXCEEDED, 8=RESOURCE_EXHAUSTED, 10=ABORTED, 14=UNAVAILABLE)
@@ -206,7 +210,13 @@ Please use the format 'prezzy' or '${instance.name}/tables/prezzy'.`);
    * @param opts
    */
   createReadStream(opts?: GetRowsOptions) {
-    return createReadStreamInternal(this, false, opts);
+    const metricsCollector =
+      this.bigtable._metricsConfigManager.createOperation(
+        MethodName.READ_ROWS,
+        StreamingState.STREAMING,
+        this,
+      );
+    return createReadStreamInternal(this, metricsCollector, opts);
   }
 
   getRows(options?: GetRowsOptions): Promise<GetRowsResponse>;
@@ -234,7 +244,13 @@ Please use the format 'prezzy' or '${instance.name}/tables/prezzy'.`);
     optionsOrCallback?: GetRowsOptions | GetRowsCallback,
     cb?: GetRowsCallback,
   ): void | Promise<GetRowsResponse> {
-    return getRowsInternal(this, false, optionsOrCallback, cb);
+    const metricsCollector =
+      this.bigtable._metricsConfigManager.createOperation(
+        MethodName.READ_ROWS,
+        StreamingState.STREAMING,
+        this,
+      );
+    return getRowsInternal(this, metricsCollector, optionsOrCallback, cb);
   }
 
   insert(

src/utils/createReadStreamInternal.ts

@@ -41,6 +41,7 @@ import {
   RETRYABLE_STATUS_CODES,
   TabularApiSurface,
 } from '../tabular-api-surface';
+import {OperationMetricsCollector} from '../client-side-metrics/operation-metrics-collector';
 
 /**
  * Creates a readable stream of rows from a Bigtable table or authorized view.
@@ -50,7 +51,7 @@ import {
  * be used to create a stream for either a whole table or an authorized view.
  *
  * @param {Table} table The Table instance to read rows from.
- * @param {boolean} singleRow boolean to check if the request is for a single row.
+ * @param metricsCollector
  * @param {GetRowsOptions} [opts] Optional configuration for the read operation.
  * @param {boolean} [opts.decode=true] If set to `false` it will not decode
  *     Buffer values returned from Bigtable.
@@ -68,13 +69,12 @@ import {
  *     match.
  * @param {object[]} [opts.ranges] A list of key ranges.
  * @param {string} [opts.start] Start value for key range.
- * @param {string} [viewName] The name of the authorized view, if applicable.
  * @returns {stream} A readable stream of {@link Row} objects.
  *
  */
 export function createReadStreamInternal(
   table: TabularApiSurface,
-  singleRow: boolean,
+  metricsCollector: OperationMetricsCollector,
   opts?: GetRowsOptions,
 ) {
   const options = opts || {};
@@ -201,11 +201,6 @@ export function createReadStreamInternal(
     }
     return originalEnd(chunk, encoding, cb);
   };
-  const metricsCollector = table.bigtable._metricsConfigManager.createOperation(
-    singleRow ? MethodName.READ_ROW : MethodName.READ_ROWS,
-    singleRow ? StreamingState.UNARY : StreamingState.STREAMING,
-    table,
-  );
   metricsCollector.onOperationStart();
   const makeNewRequest = () => {
     metricsCollector.onAttemptStart();

src/utils/getRowsInternal.ts

@@ -20,6 +20,7 @@ import {
 } from '../tabular-api-surface';
 import {createReadStreamInternal} from './createReadStreamInternal';
 import {Row} from '../row';
+import {OperationMetricsCollector} from '../client-side-metrics/operation-metrics-collector';
 // eslint-disable-next-line @typescript-eslint/no-var-requires
 const concat = require('concat-stream');
 
@@ -31,8 +32,7 @@ const concat = require('concat-stream');
  * via {@link Table#createReadStream}.
  *
  * @param {TabularApiSurface} table The table instance to get rows from.
- * @param {boolean} singleRow Boolean to check if the request is for a single row.
- * @param {string} [viewName] The name of the authorized view, if applicable.
+ * @param metricsCollector
  * @param {object} [optionsOrCallback] Configuration object. See
  *     {@link Table#createReadStream} for a complete list of options.
  * @param {object} [optionsOrCallback.gaxOptions] Request configuration options, outlined
@@ -48,15 +48,15 @@ const concat = require('concat-stream');
  */
 export function getRowsInternal(
   table: TabularApiSurface,
-  singleRow: boolean,
+  metricsCollector: OperationMetricsCollector,
   optionsOrCallback?: GetRowsOptions | GetRowsCallback,
   cb?: GetRowsCallback,
 ): void | Promise<GetRowsResponse> {
   const callback =
     typeof optionsOrCallback === 'function' ? optionsOrCallback : cb!;
   const options =
     typeof optionsOrCallback === 'object' ? optionsOrCallback : {};
-  createReadStreamInternal(table, singleRow, options)
+  createReadStreamInternal(table, metricsCollector, options)
     .on('error', callback)
     .pipe(
       concat((rows: Row[]) => {