perf(spanner): carry MetricsTracer on call context instead of global registry

handwritten/spanner/observability-test/context-isolation.ts

@@ -165,37 +165,19 @@ describe('OpenTelemetry Context Isolation Tests', () => {
       await MetricsTracerFactory.resetInstance();
     });
 
-    it('should schedule MetricsTracerFactory cleanup setInterval in ROOT_CONTEXT', () => {
+    it('should not schedule any background cleanup setInterval', () => {
       const tracer = trace.getTracer('test');
+      const setIntervalStub = sandbox.stub(global, 'setInterval');
 
-      const setIntervalStub = sandbox
-        .stub(global, 'setInterval')
-        .callsFake(() => {
-          const activeSpan = trace.getSpan(context.active());
-
-          // Assert that the active context is ROOT_CONTEXT (i.e., no active span)
-          assert.strictEqual(
-            activeSpan,
-            undefined,
-            'setInterval scheduling must be isolated within ROOT_CONTEXT and not carry any active request span',
-          );
-          return {
-            unref: () => {},
-          } as unknown as NodeJS.Timeout;
-        });
-
-      // Start an active request context
       tracer.startActiveSpan('request-span', span => {
         try {
-          // Instantiate the singleton under a request context
           MetricsTracerFactory.getInstance('mock-project-id');
         } finally {
           span.end();
         }
       });
 
-      // Verify that the cleanup interval was scheduled
-      assert.strictEqual(setIntervalStub.callCount, 1);
+      assert.strictEqual(setIntervalStub.callCount, 0);
     });
   });
 });

handwritten/spanner/src/index.ts

@@ -175,7 +175,7 @@
   >;
   observabilityOptions?: ObservabilityOptions;
   disableBuiltInMetrics?: boolean;
   interceptors?: any[];
   sessionLabels?: {[key: string]: string};
   /**
    * The Trusted Cloud Domain (TPC) DNS of the service used to make requests.
@@ -619,12 +619,11 @@
     };
 
     const res = performTeardown();
-
     if (callback) {
       // process.nextTick prevents Unhandled Promise Rejections if callback throws
       res
         .then(() => process.nextTick(() => callback(null)))
         .catch(err => process.nextTick(() => callback(err)));
       return;
     }
     return res;
@@ -1746,7 +1745,7 @@
         if (!this.clients_.has(clientName)) {
           this.clients_.set(
             clientName,
             new (v1 as Record<string, any>)[clientName](this.options),
           );
         }
       } catch (err) {
@@ -1779,6 +1778,7 @@
         });
         this.projectIdReplaced_ = true;
       }
+      config.headers = extend(true, {}, config.headers);
       config.headers[CLOUD_RESOURCE_HEADER] = replaceProjectIdToken(
         config.headers[CLOUD_RESOURCE_HEADER],
         projectId!,
@@ -1793,8 +1793,11 @@
         // Attach the x-goog-spanner-request-id to the currently active span.
         attributeXGoogSpannerRequestIdToActiveSpan(config);
       }
       const interceptors: any[] = [];
-      if (this._metricsEnabled) {
+      if (
+        this._metricsEnabled &&
+        (config.client === 'SpannerClient' || config.metricsTracer)
+      ) {
         interceptors.push(MetricInterceptor);
       }
       const requestFn = gaxClient[config.method].bind(
@@ -1806,6 +1809,7 @@
             headers: config.headers,
             options: {
               interceptors: interceptors,
+              metricsTracer: config.metricsTracer,
             },
           },
         }),
@@ -1853,7 +1857,7 @@
         // If the promise is cancellable (e.g. google-gax CancellablePromise), preserve
         // its .cancel() method so callers can cancel the underlying operation.
         if (res && typeof (res as PromiseLike<unknown>).then === 'function') {
           const chained = (res as PromiseLike<unknown>).then(null, err => {
             injectRequestIDIntoError(errorConfig, err as Error);
             throw err;
           });
@@ -1881,6 +1885,26 @@
     });
   }
 
+  private _getResourceName(reqOpts?: {
+    database?: string | object;
+    session?: string | object;
+    name?: string;
+  }): string {
+    if (!reqOpts) {
+      return '';
+    }
+    if (typeof reqOpts.database === 'string') {
+      return reqOpts.database;
+    }
+    if (typeof reqOpts.session === 'string') {
+      return reqOpts.session;
+    }
+    if (typeof reqOpts.name === 'string') {
+      return reqOpts.name;
+    }
+    return '';
+  }
+
   /**
    * Funnel all API requests through this method to be sure we have a project
    * ID.
@@ -1904,22 +1928,48 @@
       metricsTracer =
         MetricsTracerFactory?.getInstance(this.projectId_)?.createMetricsTracer(
           config.method,
-          config.reqOpts.database ?? config.reqOpts.session,
-          config.headers['x-goog-spanner-request-id'],
+          this._getResourceName(config.reqOpts),
+          config.headers?.['x-goog-spanner-request-id'],
         ) ?? null;
     }
     metricsTracer?.recordOperationStart();
+    config.metricsTracer = metricsTracer ?? undefined;
     if (typeof callback === 'function') {
       this.prepareGapicRequest_(config, (err, requestFn) => {
         if (err) {
           callback(err);
           metricsTracer?.recordOperationCompletion();
         } else {
-          const wrappedCallback = (...args) => {
+          let callbackInvoked = false;
+          let callbackThrew = false;
+          let callbackError: unknown;
+          const wrappedCallback = (...args: unknown[]) => {
+            if (callbackInvoked) {
+              return;
+            }
+            callbackInvoked = true;
             metricsTracer?.recordOperationCompletion();
-            callback(...args);
+            try {
+              callback(...args);
+            } catch (error) {
+              callbackThrew = true;
+              callbackError = error;
+              throw error;
+            }
           };
-          requestFn(wrappedCallback);
+          try {
+            requestFn(wrappedCallback);
+          } catch (error) {
+            if (callbackThrew) {
+              throw callbackError;
+            }
+            if (callbackInvoked) {
+              return;
+            }
+            callbackInvoked = true;
+            metricsTracer?.recordOperationCompletion();
+            callback(error);
+          }
         }
       });
     } else {
@@ -1929,21 +1979,27 @@
             metricsTracer?.recordOperationCompletion();
             reject(err);
           } else {
-            const result = requestFn();
-            if (result && typeof result.then === 'function') {
-              result
-                .then(val => {
-                  metricsTracer?.recordOperationCompletion();
-                  resolve(val);
-                  return val;
-                })
-                .catch(error => {
-                  metricsTracer?.recordOperationCompletion();
-                  reject(error);
-                });
-            } else {
+            try {
+              const result = requestFn();
+              if (result && typeof result.then === 'function') {
+                result
+                  .then(val => {
+                    metricsTracer?.recordOperationCompletion();
+                    resolve(val);
+                    return val;
+                  })
+                  .catch(error => {
+                    metricsTracer?.recordOperationCompletion();
+                    reject(error);
+                    return null;
+                  });
+              } else {
+                metricsTracer?.recordOperationCompletion();
+                resolve(result);
+              }
+            } catch (error) {
               metricsTracer?.recordOperationCompletion();
-              resolve(result);
+              reject(error);
             }
           }
         });
@@ -1974,30 +2030,76 @@
       metricsTracer =
         MetricsTracerFactory?.getInstance(this.projectId_)?.createMetricsTracer(
           config.method,
-          config.reqOpts.session ?? config.reqOpts.database,
-          config.headers['x-goog-spanner-request-id'],
+          this._getResourceName(config.reqOpts),
+          config.headers?.['x-goog-spanner-request-id'],
         ) ?? null;
     }
     metricsTracer?.recordOperationStart();
+    config.metricsTracer = metricsTracer ?? undefined;
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    let callStream: any = null;
+    let cleanedUp = false;
+    const cleanup = () => {
+      if (cleanedUp) {
+        return;
+      }
+      cleanedUp = true;
+      if (
+        callStream &&
+        typeof callStream.destroy === 'function' &&
+        !callStream.destroyed
+      ) {
+        callStream.destroy();
+      }
+      metricsTracer?.recordOperationCompletion();
+    };
+
     const stream = streamEvents(through.obj());
+    const origDestroy = stream._destroy;
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    stream._destroy = function (err: any, cb: any) {
+      cleanup();
+      if (typeof origDestroy === 'function') {
+        origDestroy.call(stream, err, cb);
+      } else if (typeof cb === 'function') {
+        cb(err);
+      }
+    };
     stream.once('reading', () => {
       this.prepareGapicRequest_(config, (err, requestFn) => {
+        if (stream.destroyed) {
+          cleanup();
+          return;
+        }
         if (err) {
           stream.destroy(err);
           return;
         }
-        requestFn()
-          .on('error', err => {
-            stream.destroy(err);
-          })
-          .pipe(stream);
+        try {
+          callStream = requestFn();
+          if (stream.destroyed) {
+            cleanup();
+            return;
+          }
+          if (callStream) {
+            callStream
+              .on('error', err => {
+                stream.destroy(err);
+              })
+              .pipe(stream);
+          } else {
+            stream.destroy(new Error('Failed to initialize request stream.'));
+          }
+        } catch (error) {
+          stream.destroy(error as Error);
+        }
       });
     });
     stream.on('finish', () => {
       stream.destroy();
     });
     stream.on('close', () => {
-      metricsTracer?.recordOperationCompletion();
+      cleanup();
     });
     return stream;
   }

handwritten/spanner/src/metrics/constants.ts

@@ -16,8 +16,13 @@ import {AggregationType, ViewOptions} from '@opentelemetry/sdk-metrics';
 export const SPANNER_METER_NAME = 'spanner-nodejs';
 export const CLIENT_METRICS_PREFIX = 'spanner.googleapis.com/internal/client';
 export const SPANNER_RESOURCE_TYPE = 'spanner_instance_client';
-// Maximum time to keep MetricsTracers before considering them stale, and stop tracking them.
+/**
+ * @deprecated No longer used after eliminating the background tracer cleanup timer.
+ */
 export const TRACER_CLEANUP_THRESHOLD_MS = 60 * 60 * 1000; // 60 minutes
+/**
+ * @deprecated No longer used after eliminating the background tracer cleanup timer.
+ */
 export const TRACER_CLEANUP_INTERVAL_MS = 30 * 60 * 1000; // 30 Minutes
 // OTel semantic conventions
 // See https://github.com/open-telemetry/opentelemetry-js/blob/main/semantic-conventions/README.md#unstable-semconv

handwritten/spanner/src/metrics/interceptor.ts

@@ -14,7 +14,6 @@
 
 import {grpc} from 'google-gax';
 import {InterceptingListener, Metadata, StatusObject} from '@grpc/grpc-js';
-import {MetricsTracerFactory} from './metrics-tracer-factory';
 import {isAFEServerTimingEnabled} from '../common';
 
 /**
@@ -31,18 +30,9 @@ import {isAFEServerTimingEnabled} from '../common';
 export const MetricInterceptor = (options, nextCall) => {
   return new grpc.InterceptingCall(nextCall(options), {
     start: function (metadata, listener, next) {
-      // Record attempt metric on request start
-      const resourcePrefix = metadata.get(
-        'google-cloud-resource-prefix',
-      )[0] as string;
-      const match = resourcePrefix?.match(/^projects\/([^/]+)\//);
-      const projectId = match ? match[1] : undefined;
-      let factory;
-      if (projectId) {
-        factory = MetricsTracerFactory.getInstance(projectId);
-      }
-      const requestId = metadata.get('x-goog-spanner-request-id')[0] as string;
-      const metricsTracer = factory?.getCurrentTracer(requestId);
+      // Record attempt metric on request start.
+      // The tracer is carried directly on the call options.
+      const metricsTracer = options?.metricsTracer ?? null;
       metricsTracer?.recordAttemptStart();
       const afeServerTimingEnabled = isAFEServerTimingEnabled();
 
@@ -73,26 +63,32 @@ export const MetricInterceptor = (options, nextCall) => {
           listener.onReceiveMessage(message);
         },
         onReceiveStatus: function (status: StatusObject) {
-          listener.onReceiveStatus(status);
-
-          if (!metricsTracer) {
-            return;
-          }
-
-          // Record attempt metric completion
-          metricsTracer.recordAttemptCompletion(status.code);
-          if (typeof metricsTracer.gfeLatency === 'number') {
-            metricsTracer.recordGfeLatency(status.code);
-          } else {
-            metricsTracer.recordGfeConnectivityErrorCount(status.code);
-          }
-          if (afeServerTimingEnabled) {
-            if (typeof metricsTracer.afeLatency === 'number') {
-              metricsTracer.recordAfeLatency(status.code);
+          if (metricsTracer) {
+            // Record attempt metric completion before notifying downstream listener
+            metricsTracer.recordAttemptCompletion(status?.code);
+            if (
+              typeof metricsTracer.gfeLatency === 'number' &&
+              Number.isFinite(metricsTracer.gfeLatency) &&
+              metricsTracer.gfeLatency >= 0
+            ) {
+              metricsTracer.recordGfeLatency(status?.code);
             } else {
-              metricsTracer.recordAfeConnectivityErrorCount(status.code);
+              metricsTracer.recordGfeConnectivityErrorCount(status?.code);
+            }
+            if (afeServerTimingEnabled) {
+              if (
+                typeof metricsTracer.afeLatency === 'number' &&
+                Number.isFinite(metricsTracer.afeLatency) &&
+                metricsTracer.afeLatency >= 0
+              ) {
+                metricsTracer.recordAfeLatency(status?.code);
+              } else {
+                metricsTracer.recordAfeConnectivityErrorCount(status?.code);
+              }
             }
           }
+
+          listener.onReceiveStatus(status);
         },
       };