feat: Add client side metrics collection to readRows calls (#1571)

**Summary:**

This PR will collect client side metrics for readRows calls.

**Background:**

The following PRs have been added to support the plumbing for client-side metrics.
https://togithub.com/googleapis/nodejs-bigtable/pull/1569
https://togithub.com/googleapis/nodejs-bigtable/pull/1566

They didn't change the client behaviour at all. This PR starts using this plumbing for readRows calls.

The diagram below shows the CSM architecture with new pieces in green in the diagram:

![CSM With fixture - 3rd PR - ReadRows calls (1)](https://togithub.com/user-attachments/assets/7f1207e4-55c2-4bb0-8dac-796bc003720a)

**Changes:**

`src/client-side-metrics/operation-metrics-collector.ts:` Added optimizations and added guards to make sure that if the data from the server is of some unexpected form that it won't crash our clients.
`src/index.ts:` Add code to generate a uuid for each client and store the GCP metrics handler on the client. Also add piping to pass status events along and add a setting that lets users turn off client side metrics.
`src/tabular-api-surface.ts`: Use the metrics collector with the gcp metrics handler to collect client side metrics in readRows calls
`system-test/client-side-metrics-setup-table.ts:` A test utility function for setting up an instance with a few tables
`system-test/client-side-metrics.ts:` A file with some tests for capturing various scenarios where a user might be collecting client side metrics that ensures the right metrics are exported without error even when multiple tables, instances or clients are involved.
`system-test/read-rows.ts:` This test has a mock that is client side metrics incompatible so don't use client side metrics in this test.

**Next Steps:**

- Add plumbing for application blocking latencies
- Collect client side metrics for other grpc endpoints

Author: danieljbruce

src/client-side-metrics/client-side-metrics-attributes.ts

@@ -25,6 +25,7 @@ export enum StreamingState {
  * metrics, allowing for differentiation of performance by method.
  */
 export enum MethodName {
+  READ_ROW = 'Bigtable.ReadRow',
   READ_ROWS = 'Bigtable.ReadRows',
   MUTATE_ROW = 'Bigtable.MutateRow',
   CHECK_AND_MUTATE_ROW = 'Bigtable.CheckAndMutateRow',

src/client-side-metrics/exporter.ts

@@ -19,11 +19,10 @@ import {
   Histogram,
   ResourceMetrics,
 } from '@opentelemetry/sdk-metrics';
-import {grpc, ServiceError} from 'google-gax';
+import {ClientOptions, ServiceError} from 'google-gax';
 import {MetricServiceClient} from '@google-cloud/monitoring';
 import {google} from '@google-cloud/monitoring/build/protos/protos';
 import ICreateTimeSeriesRequest = google.monitoring.v3.ICreateTimeSeriesRequest;
-import {RetryOptions} from 'google-gax';
 
 export interface ExportResult {
   code: number;
@@ -119,7 +118,7 @@ function getIntegerPoints(dataPoint: DataPoint<number>) {
  * getResource gets the resource object which is used for building the timeseries
  * object that will be sent to Google Cloud Monitoring dashboard
  *
- * @param {string} metricName The backend name of the metric that we want to record
+ * @param {string} projectId The name of the project
  * @param {DataPoint} dataPoint The datapoint containing the data we wish to
  * send to the Google Cloud Monitoring dashboard
  */
@@ -184,6 +183,7 @@ function getMetric(
  * metric attributes, data points, and aggregation information, into an object
  * that conforms to the expected request format of the Cloud Monitoring API.
  *
+ * @param projectId
  * @param {ResourceMetrics} exportArgs - The OpenTelemetry metrics data to be converted. This
  *   object contains resource attributes, scope information, and a list of
  *   metrics with their associated data points.
@@ -211,14 +211,10 @@ function getMetric(
  *
  *
  */
-export function metricsToRequest(exportArgs: ResourceMetrics) {
-  type WithSyncAttributes = {_syncAttributes: {[index: string]: string}};
-  const resourcesWithSyncAttributes =
-    exportArgs.resource as unknown as WithSyncAttributes;
-  const projectId =
-    resourcesWithSyncAttributes._syncAttributes[
-      'monitored_resource.project_id'
-    ];
+export function metricsToRequest(
+  projectId: string,
+  exportArgs: ResourceMetrics,
+) {
   const timeSeriesArray = [];
   for (const scopeMetrics of exportArgs.scopeMetrics) {
     for (const scopeMetric of scopeMetrics.metrics) {
@@ -297,49 +293,33 @@ export function metricsToRequest(exportArgs: ResourceMetrics) {
  * @beta
  */
 export class CloudMonitoringExporter extends MetricExporter {
-  private monitoringClient = new MetricServiceClient();
+  private client: MetricServiceClient;
 
-  export(
+  constructor(options: ClientOptions) {
+    super();
+    if (options && options.apiEndpoint) {
+      // We want the MetricServiceClient to always hit its default endpoint.
+      delete options.apiEndpoint;
+    }
+    this.client = new MetricServiceClient(options);
+  }
+
+  async export(
     metrics: ResourceMetrics,
     resultCallback: (result: ExportResult) => void,
-  ): void {
+  ): Promise<void> {
     (async () => {
       try {
-        const request = metricsToRequest(metrics);
-        // In order to manage the "One or more points were written more
-        // frequently than the maximum sampling period configured for the
-        // metric." error we should have the metric service client retry a few
-        // times to ensure the metrics do get written.
-        //
-        // We use all the usual retry codes plus INVALID_ARGUMENT (code 3)
-        // because INVALID ARGUMENT (code 3) corresponds to the maximum
-        // sampling error.
-        const retry = new RetryOptions(
-          [
-            grpc.status.INVALID_ARGUMENT,
-            grpc.status.DEADLINE_EXCEEDED,
-            grpc.status.RESOURCE_EXHAUSTED,
-            grpc.status.ABORTED,
-            grpc.status.UNAVAILABLE,
-          ],
-          {
-            initialRetryDelayMillis: 5000,
-            retryDelayMultiplier: 2,
-            maxRetryDelayMillis: 50000,
-          },
-        );
-        await this.monitoringClient.createTimeSeries(
+        const projectId = await this.client.getProjectId();
+        const request = metricsToRequest(projectId, metrics);
+        await this.client.createServiceTimeSeries(
           request as ICreateTimeSeriesRequest,
-          {
-            retry,
-          },
         );
         // The resultCallback typically accepts a value equal to {code: x}
         // for some value x along with other info. When the code is equal to 0
         // then the operation completed successfully. When the code is not equal
-        // to 0 then the operation failed. Open telemetry logs errors to the
-        // console when the resultCallback passes in non-zero code values and
-        // logs nothing when the code is 0.
+        // to 0 then the operation failed. The resultCallback will not log
+        // anything to the console whether the error code was 0 or not.
         resultCallback({code: 0});
       } catch (error) {
         resultCallback(error as ServiceError);

src/client-side-metrics/gcp-metrics-handler.ts

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+import {CloudMonitoringExporter} from './exporter';
 import {
   IMetricsHandler,
   OnAttemptCompleteData,
@@ -20,13 +21,39 @@ import {
 import * as Resources from '@opentelemetry/resources';
 import * as ResourceUtil from '@google-cloud/opentelemetry-resource-util';
 import {PushMetricExporter, View} from '@opentelemetry/sdk-metrics';
+import {ClientOptions} from 'google-gax';
 const {
   Aggregation,
   ExplicitBucketHistogramAggregation,
   MeterProvider,
   Histogram,
   PeriodicExportingMetricReader,
 } = require('@opentelemetry/sdk-metrics');
+import * as os from 'os';
+import * as crypto from 'crypto';
+
+/**
+ * Generates a unique client identifier string.
+ *
+ * This function creates a client identifier that incorporates the hostname,
+ * process ID, and a UUID to ensure uniqueness across different client instances
+ * and processes. The identifier follows the pattern:
+ *
+ * `node-<uuid>-<pid><hostname>`
+ *
+ * where:
+ * - `<uuid>` is a randomly generated UUID (version 4).
+ * - `<pid>` is the process ID of the current Node.js process.
+ * - `<hostname>` is the hostname of the machine.
+ *
+ * @returns {string} A unique client identifier string.
+ */
+function generateClientUuid() {
+  const hostname = os.hostname() || 'localhost';
+  const currentPid = process.pid || '';
+  const uuid4 = crypto.randomUUID();
+  return `node-${uuid4}-${currentPid}${hostname}`;
+}
 
 /**
  * A collection of OpenTelemetry metric instruments used to record
@@ -47,10 +74,9 @@ interface MetricsInstruments {
  * This method gets the open telemetry instruments that will store GCP metrics
  * for a particular project.
  *
- * @param projectId The project for which the instruments will be stored.
  * @param exporter The exporter the metrics will be sent to.
  */
-function createInstruments(projectId: string, exporter: PushMetricExporter) {
+function createInstruments(exporter: PushMetricExporter): MetricsInstruments {
   const latencyBuckets = [
     0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 8.0, 10.0, 13.0, 16.0, 20.0, 25.0, 30.0,
     40.0, 50.0, 65.0, 80.0, 100.0, 130.0, 160.0, 200.0, 250.0, 300.0, 400.0,
@@ -80,7 +106,6 @@ function createInstruments(projectId: string, exporter: PushMetricExporter) {
     views: viewList,
     resource: new Resources.Resource({
       'service.name': 'Cloud Bigtable Table',
-      'monitored_resource.project_id': projectId,
     }).merge(new ResourceUtil.GcpDetectorSync().detect()),
     readers: [
       // Register the exporter
@@ -183,11 +208,8 @@ function createInstruments(projectId: string, exporter: PushMetricExporter) {
  * associating them with relevant attributes for detailed analysis in Cloud Monitoring.
  */
 export class GCPMetricsHandler implements IMetricsHandler {
-  private exporter: PushMetricExporter;
-  // The variable below is the singleton map from projects to instrument stacks
-  // which exists so that we only create one instrument stack per project. This
-  // will eliminate errors due to the maximum sampling period.
-  static instrumentsForProject: {[projectId: string]: MetricsInstruments} = {};
+  private otelInstruments: MetricsInstruments;
+  private clientUid: string;
 
   /**
    * The `GCPMetricsHandler` is responsible for managing and recording
@@ -196,33 +218,11 @@ export class GCPMetricsHandler implements IMetricsHandler {
    * (histograms and counters) and exports them to Google Cloud Monitoring
    * through the provided `PushMetricExporter`.
    *
-   * @param exporter - The `PushMetricExporter` instance to use for exporting
-   *   metrics to Google Cloud Monitoring. This exporter is responsible for
-   *   sending the collected metrics data to the monitoring backend. The provided exporter must be fully configured, for example the projectId must have been set.
    */
-  constructor(exporter: PushMetricExporter) {
-    this.exporter = exporter;
-  }
-
-  /**
-   * Initializes the OpenTelemetry metrics instruments if they haven't been already.
-   * Creates and registers metric instruments (histograms and counters) for various Bigtable client metrics.
-   * Sets up a MeterProvider and configures a PeriodicExportingMetricReader for exporting metrics to Cloud Monitoring.
-   *
-   * which will be provided to the exporter in every export call.
-   *
-   */
-  private getInstruments(projectId: string): MetricsInstruments {
-    // The projectId is needed per metrics handler because when the exporter is
-    // used it provides the project id for the name of the time series exported.
-    // ie. name: `projects/${....['monitored_resource.project_id']}`,
-    if (!GCPMetricsHandler.instrumentsForProject[projectId]) {
-      GCPMetricsHandler.instrumentsForProject[projectId] = createInstruments(
-        projectId,
-        this.exporter,
-      );
-    }
-    return GCPMetricsHandler.instrumentsForProject[projectId];
+  constructor(options: ClientOptions) {
+    this.clientUid = generateClientUuid();
+    const exporter = new CloudMonitoringExporter(options);
+    this.otelInstruments = createInstruments(exporter);
   }
 
   /**
@@ -231,11 +231,11 @@ export class GCPMetricsHandler implements IMetricsHandler {
    * @param {OnOperationCompleteData} data Data related to the completed operation.
    */
   onOperationComplete(data: OnOperationCompleteData) {
-    const otelInstruments = this.getInstruments(data.projectId);
+    const otelInstruments = this.otelInstruments;
     const commonAttributes = {
       app_profile: data.metricsCollectorData.app_profile,
       method: data.metricsCollectorData.method,
-      client_uid: data.metricsCollectorData.client_uid,
+      client_uid: this.clientUid,
       client_name: data.client_name,
       instanceId: data.metricsCollectorData.instanceId,
       table: data.metricsCollectorData.table,
@@ -271,11 +271,11 @@ export class GCPMetricsHandler implements IMetricsHandler {
    * @param {OnAttemptCompleteData} data Data related to the completed attempt.
    */
   onAttemptComplete(data: OnAttemptCompleteData) {
-    const otelInstruments = this.getInstruments(data.projectId);
+    const otelInstruments = this.otelInstruments;
     const commonAttributes = {
       app_profile: data.metricsCollectorData.app_profile,
       method: data.metricsCollectorData.method,
-      client_uid: data.metricsCollectorData.client_uid,
+      client_uid: this.clientUid,
       status: data.status,
       client_name: data.client_name,
       instanceId: data.metricsCollectorData.instanceId,

src/client-side-metrics/metrics-config-manager.ts

@@ -0,0 +1,44 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import {IMetricsHandler} from './metrics-handler';
+import {
+  ITabularApiSurface,
+  OperationMetricsCollector,
+} from './operation-metrics-collector';
+import {MethodName, StreamingState} from './client-side-metrics-attributes';
+
+/**
+ * A class for tracing and recording client-side metrics related to Bigtable operations.
+ */
+export class ClientSideMetricsConfigManager {
+  private metricsHandlers: IMetricsHandler[];
+
+  constructor(handlers: IMetricsHandler[]) {
+    this.metricsHandlers = handlers;
+  }
+
+  createOperation(
+    methodName: MethodName,
+    streaming: StreamingState,
+    table: ITabularApiSurface,
+  ): OperationMetricsCollector {
+    return new OperationMetricsCollector(
+      table,
+      methodName,
+      streaming,
+      this.metricsHandlers,
+    );
+  }
+}

src/client-side-metrics/metrics-handler.ts

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 import {MethodName, StreamingState} from './client-side-metrics-attributes';
-import {grpc} from 'google-gax';
 
 /**
  * The interfaces below use undefined instead of null to indicate a metric is
@@ -28,11 +27,9 @@ type IMetricsCollectorData = {
   zone?: string;
   app_profile?: string;
   method: MethodName;
-  client_uid: string;
 };
 
 interface StandardData {
-  projectId: string;
   metricsCollectorData: IMetricsCollectorData;
   client_name: string;
   streaming: StreamingState;