fix(otel): optimize metrics tracking by eliminating promise chaining overhead

Author: PavelPashov

packages/client/lib/client/commands-queue.ts

@@ -6,6 +6,7 @@ import { ChannelListeners, PubSub, PubSubCommand, PubSubListener, PubSubType, Pu
 import { AbortError, ErrorReply, CommandTimeoutDuringMaintenanceError, TimeoutError } from '../errors';
 import { MonitorCallback } from '.';
 import { dbgMaintenance } from './enterprise-maintenance-manager';
+import { OTelClientAttributes, OTelMetrics } from '../opentelemetry';
 
 export interface CommandOptions<T = TypeMapping> {
   chainId?: symbol;
@@ -199,23 +200,35 @@ export default class RedisCommandsQueue {
 
   addCommand<T>(
     args: ReadonlyArray<RedisArgument>,
-    options?: CommandOptions
+    options?: CommandOptions,
+    otelAttributes?: OTelClientAttributes
   ): Promise<T> {
     if (this.#maxLength && this.#toWrite.length + this.#waitingForReply.length >= this.#maxLength) {
       return Promise.reject(new Error('The queue is full'));
     } else if (options?.abortSignal?.aborted) {
       return Promise.reject(new AbortError());
     }
 
+    const recordOperation = OTelMetrics.instance.createRecordOperationDuration(args, otelAttributes);
+    OTelMetrics.instance.recordPendingRequests(1, otelAttributes);
+
     return new Promise((resolve, reject) => {
       let node: DoublyLinkedNode<CommandToWrite>;
       const value: CommandToWrite = {
         args,
         chainId: options?.chainId,
         abort: undefined,
         timeout: undefined,
-        resolve,
-        reject,
+        resolve: reply => {
+          recordOperation()
+          OTelMetrics.instance.recordPendingRequests(-1, otelAttributes);
+          resolve(reply as T);
+        },
+        reject: (err) => {
+          recordOperation(err as Error);
+          OTelMetrics.instance.recordPendingRequests(-1, otelAttributes);
+          reject(err);
+        },
         channelsCounter: undefined,
         typeMapping: options?.typeMapping
       };

packages/client/lib/client/index.ts

@@ -21,7 +21,7 @@ import { BasicCommandParser, CommandParser } from './parser';
 import SingleEntryCache from '../single-entry-cache';
 import { version } from '../../package.json'
 import EnterpriseMaintenanceManager, { MaintenanceUpdate, MovingEndpointType } from './enterprise-maintenance-manager';
-import { OTelClientAttributes, OTelMetrics } from '../opentelemetry';
+import { OTelClientAttributes } from '../opentelemetry';
 
 export interface RedisClientOptions<
   M extends RedisModules = RedisModules,
@@ -652,7 +652,7 @@ export default class RedisClient<
           .addCommand(cmd, {
             chainId,
             asap
-          })
+          }, this._getClientOTelAttributes())
           .catch(errorHandler)
       );
     }
@@ -1076,17 +1076,12 @@ export default class RedisClient<
     args: ReadonlyArray<RedisArgument>,
     options?: CommandOptions
   ): Promise<T> {
-    const clientAttributes = this._self._getClientOTelAttributes();
-    const recordOperation = OTelMetrics.instance.createRecordOperationDuration(args, clientAttributes);
-
     if (!this._self.#socket.isOpen) {
-      recordOperation(new ClientClosedError());
       return Promise.reject(new ClientClosedError());
     } else if (
       !this._self.#socket.isReady &&
       this._self.#options.disableOfflineQueue
     ) {
-      recordOperation(new ClientOfflineError());
       return Promise.reject(new ClientOfflineError());
     }
 
@@ -1096,29 +1091,11 @@ export default class RedisClient<
       ...options,
     };
 
-    const promise = this._self.#queue.addCommand<T>(args, opts);
-
-    if (OTelMetrics.isInitialized()) {
-      OTelMetrics.instance.recordPendingRequests(1, clientAttributes);
+    const promise = this._self.#queue.addCommand<T>(args, opts, this._self._getClientOTelAttributes());
 
-      const trackedPromise = promise
-        .then((reply) => {
-          recordOperation();
-          OTelMetrics.instance.recordPendingRequests(-1, clientAttributes);
-          return reply;
-        })
-        .catch((err) => {
-          recordOperation(err);
-          OTelMetrics.instance.recordPendingRequests(-1, clientAttributes);
-          throw err;
-        });
 
-      this._self.#scheduleWrite();
-      return trackedPromise;
-    } else {
-      this._self.#scheduleWrite();
-      return promise;
-    }
+    this._self.#scheduleWrite();
+    return promise;
   }
 
   async SELECT(db: number): Promise<void> {
@@ -1354,20 +1331,20 @@ export default class RedisClient<
     const typeMapping = this._commandOptions?.typeMapping;
     const chainId = Symbol('MULTI Chain');
     const promises = [
-      this._self.#queue.addCommand(['MULTI'], { chainId }),
+      this._self.#queue.addCommand(['MULTI'], { chainId }, this._self._getClientOTelAttributes()),
     ];
 
     for (const { args } of commands) {
       promises.push(
         this._self.#queue.addCommand(args, {
           chainId,
           typeMapping
-        })
+        }, this._self._getClientOTelAttributes())
       );
     }
 
     promises.push(
-      this._self.#queue.addCommand(['EXEC'], { chainId })
+      this._self.#queue.addCommand(['EXEC'], { chainId }, this._self._getClientOTelAttributes())
     );
 
     this._self.#scheduleWrite();
@@ -1543,7 +1520,7 @@ export default class RedisClient<
     this._self.#credentialsSubscription = null;
     return this._self.#socket.quit(async () => {
       clearTimeout(this._self.#pingTimer);
-      const quitPromise = this._self.#queue.addCommand<string>(['QUIT']);
+      const quitPromise = this._self.#queue.addCommand<string>(['QUIT'], undefined, this._self._getClientOTelAttributes());
       this._self.#scheduleWrite();
       return quitPromise;
     });

packages/client/lib/opentelemetry/metrics.spec.ts

@@ -12,7 +12,7 @@ import {
 import { spy } from "sinon";
 
 import { OTelMetrics } from "./metrics";
-import { METRIC_NAMES, ObservabilityConfig } from "./types";
+import { METRIC_NAMES, ObservabilityConfig, OTEL_ATTRIBUTES } from "./types";
 import { NOOP_UP_DOWN_COUNTER_METRIC } from "./noop-meter";
 import { noopFunction, waitForMetrics } from "./utils";
 import testUtils, { GLOBAL } from "../test-utils";
@@ -333,7 +333,13 @@ describe("OTel Metrics E2E", function () {
 
       const dataPoints = metric.dataPoints as DataPoint<Histogram>[];
 
-      assert.strictEqual(dataPoints[0].value.count, 3);
+      const setDataPoint = dataPoints.find(
+        (dp) => dp.attributes[OTEL_ATTRIBUTES.dbOperationName] === "SET"
+      );
+
+      assert.ok(setDataPoint, "expected SET data point to be present");
+
+      assert.strictEqual(setDataPoint.value.count, 3);
     },
     {
       client: GLOBAL.SERVERS.OPEN,