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

@@ -169,7 +169,7 @@
   >;
   observabilityOptions?: ObservabilityOptions;
   disableBuiltInMetrics?: boolean;
   interceptors?: any[];
   sessionLabels?: {[key: string]: string};
   /**
    * The Trusted Cloud Domain (TPC) DNS of the service used to make requests.
@@ -534,7 +534,7 @@
     if (!this.clients_.has(clientName)) {
       this.clients_.set(
         clientName,
-        new v1[clientName](this.options as ClientOptions),
+        new v1.InstanceAdminClient(this.options as ClientOptions),
       );
     }
     return this.clients_.get(clientName)! as v1.InstanceAdminClient;
@@ -558,7 +558,7 @@
     if (!this.clients_.has(clientName)) {
       this.clients_.set(
         clientName,
-        new v1[clientName](this.options as ClientOptions),
+        new v1.DatabaseAdminClient(this.options as ClientOptions),
       );
     }
     return this.clients_.get(clientName)! as v1.DatabaseAdminClient;
@@ -615,9 +615,10 @@
 
     if (callback) {
       // process.nextTick prevents Unhandled Promise Rejections if callback throws
+      // eslint-disable-next-line promise/catch-or-return
       res.then(
         () => process.nextTick(() => callback(null)),
         err => process.nextTick(() => callback(err)),
       );
     } else {
       return res;
@@ -1727,6 +1728,7 @@
       const clientName = config.client;
       try {
         if (!this.clients_.has(clientName)) {
+          // eslint-disable-next-line import/namespace
           this.clients_.set(clientName, new v1[clientName](this.options));
         }
       } catch (err) {
@@ -1759,6 +1761,7 @@
         });
         this.projectIdReplaced_ = true;
       }
+      config.headers = extend(true, {}, config.headers);
       config.headers[CLOUD_RESOURCE_HEADER] = replaceProjectIdToken(
         config.headers[CLOUD_RESOURCE_HEADER],
         projectId!,
@@ -1773,8 +1776,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(
@@ -1786,6 +1792,7 @@
             headers: config.headers,
             options: {
               interceptors: interceptors,
+              metricsTracer: config.metricsTracer,
             },
           },
         }),
@@ -1828,7 +1835,7 @@
             }
 
             return new Promise((resolve, reject) => {
               requestFn(...args)
                 .then(resolve)
                 .catch(err => {
                   injectRequestIDIntoError(config, err as Error);
@@ -1843,6 +1850,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.
@@ -1865,22 +1892,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 {
@@ -1890,20 +1943,27 @@
             metricsTracer?.recordOperationCompletion();
             reject(err);
           } else {
-            const result = requestFn();
-            if (result && typeof result.then === 'function') {
-              result
-                .then(val => {
-                  metricsTracer?.recordOperationCompletion();
-                  resolve(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);
             }
           }
         });
@@ -1933,30 +1993,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

@@ -19,8 +19,13 @@ import {
 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