wip: add OpenTelemetry metrics instrumentation

packages/client/index.ts

@@ -35,3 +35,4 @@ export { GEO_REPLY_WITH, GeoReplyWith } from './lib/commands/GEOSEARCH_WITH';
 export { SetOptions, CLIENT_KILL_FILTERS, FAILOVER_MODES, CLUSTER_SLOT_STATES, COMMAND_LIST_FILTER_BY, REDIS_FLUSH_MODES } from './lib/commands'
 
 export { BasicClientSideCache, BasicPooledClientSideCache } from './lib/client/cache';
+export { OpenTelemetry } from './lib/opentelemetry';

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/enterprise-maintenance-manager.ts

@@ -7,6 +7,7 @@ import assert from "node:assert";
 import { setTimeout } from "node:timers/promises";
 import RedisSocket, { RedisTcpSocketOptions } from "./socket";
 import diagnostics_channel from "node:diagnostics_channel";
+import { METRIC_ERROR_TYPE, OTelClientAttributes, OTelMetrics } from "../opentelemetry";
 
 export const MAINTENANCE_EVENTS = {
   PAUSE_WRITING: "pause-writing",
@@ -51,6 +52,7 @@ interface Client {
   _pause: () => void;
   _unpause: () => void;
   _maintenanceUpdate: (update: MaintenanceUpdate) => void;
+  _getClientOTelAttributes: () => OTelClientAttributes;
   duplicate: () => Client;
   connect: () => Promise<Client>;
   destroy: () => void;
@@ -109,6 +111,12 @@ export default class EnterpriseMaintenanceManager {
         if (options.maintNotifications === "enabled") {
           throw error;
         }
+
+        OTelMetrics.instance.recordClientErrorsHandled(METRIC_ERROR_TYPE.HANDSHAKE_FAILED, {
+          host,
+          // TODO add port
+          // port: options?.socket?.port,
+        });
       },
     };
   }
@@ -134,6 +142,8 @@ export default class EnterpriseMaintenanceManager {
 
     const type = String(push[0]);
 
+    OTelMetrics.instance.recordMaintenanceNotifications(this.#client._getClientOTelAttributes()); 
+
     emitDiagnostics({
           type,
           timestamp: Date.now(),
@@ -267,6 +277,7 @@ export default class EnterpriseMaintenanceManager {
     dbgMaintenance("Resume writing");
     this.#client._unpause();
     this.#onMigrated();
+    OTelMetrics.instance.recordConnectionHandoff(this.#client._getClientOTelAttributes());
   };
 
   #onMigrating = () => {

packages/client/lib/client/index.ts

@@ -21,6 +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 } from '../opentelemetry';
 
 export interface RedisClientOptions<
   M extends RedisModules = RedisModules,
@@ -651,7 +652,7 @@ export default class RedisClient<
           .addCommand(cmd, {
             chainId,
             asap
-          })
+          }, this._getClientOTelAttributes())
           .catch(errorHandler)
       );
     }
@@ -1060,23 +1061,39 @@ export default class RedisClient<
       reply;
   }
 
+  /**
+   * @internal
+  */
+  _getClientOTelAttributes(): OTelClientAttributes { // TODO maybe rename this to something more generic
+    return {
+      host: this._self.#socket.host,
+      port: this._self.#socket.port,
+      db: this._self.#selectedDB,
+    };
+  }
+
   sendCommand<T = ReplyUnion>(
     args: ReadonlyArray<RedisArgument>,
     options?: CommandOptions
   ): Promise<T> {
     if (!this._self.#socket.isOpen) {
       return Promise.reject(new ClientClosedError());
-    } else if (!this._self.#socket.isReady && this._self.#options.disableOfflineQueue) {
+    } else if (
+      !this._self.#socket.isReady &&
+      this._self.#options.disableOfflineQueue
+    ) {
       return Promise.reject(new ClientOfflineError());
     }
 
     // Merge global options with provided options
     const opts = {
       ...this._self._commandOptions,
-      ...options
-    }
+      ...options,
+    };
+
+    const promise = this._self.#queue.addCommand<T>(args, opts, this._self._getClientOTelAttributes());
+
 
-    const promise = this._self.#queue.addCommand<T>(args, opts);
     this._self.#scheduleWrite();
     return promise;
   }
@@ -1314,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();
@@ -1503,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/client/socket.ts

@@ -5,6 +5,7 @@ import { ConnectionTimeoutError, ClientClosedError, SocketClosedUnexpectedlyErro
 import { setTimeout } from 'node:timers/promises';
 import { RedisArgument } from '../RESP/types';
 import { dbgMaintenance } from './enterprise-maintenance-manager';
+import { OTelMetrics } from '../opentelemetry';
 
 type NetOptions = {
   tls?: false;
@@ -85,6 +86,14 @@ export default class RedisSocket extends EventEmitter {
     return this.#socketEpoch;
   }
 
+  get host() {
+    return this.#socket?.remoteAddress;
+  }
+
+  get port() {
+    return this.#socket?.remotePort;
+  }
+
   constructor(initiator: RedisSocketInitiator, options?: RedisSocketOptions) {
     super();
 
@@ -215,6 +224,7 @@ export default class RedisSocket extends EventEmitter {
     let retries = 0;
     do {
       try {
+        const started = performance.now();
         this.#socket = await this.#createSocket();
         this.emit('connect');
 
@@ -228,6 +238,14 @@ export default class RedisSocket extends EventEmitter {
         this.#isReady = true;
         this.#socketEpoch++;
         this.emit('ready');
+        OTelMetrics.instance.recordConnectionCount(1, {
+          host: this.host,
+          port: this.port,
+        });
+        OTelMetrics.instance.recordConnectionCreateTime(performance.now() - started, {
+          host: this.host,
+          port: this.port,
+        });
       } catch (err) {
         const retryIn = this.#shouldReconnect(retries++, err as Error);
         if (typeof retryIn !== 'number') {
@@ -252,8 +270,16 @@ export default class RedisSocket extends EventEmitter {
 
     if(ms !== undefined) {
       this.#socket?.setTimeout(ms);
+      OTelMetrics.instance.recordConnectionRelaxedTimeout(1, {
+        host: this.host,
+        port: this.port,
+      });
     } else {
       this.#socket?.setTimeout(this.#socketTimeout ?? 0);
+      OTelMetrics.instance.recordConnectionRelaxedTimeout(-1, {
+        host: this.host,
+        port: this.port,
+      });
     }
   }
 
@@ -304,6 +330,13 @@ export default class RedisSocket extends EventEmitter {
     this.#isReady = false;
     this.emit('error', err);
 
+    if (wasReady) {
+      OTelMetrics.instance.recordConnectionCount(-1, {
+        host: this.host,
+        port: this.port,
+      });
+    }
+
     if (!wasReady || !this.#isOpen || typeof this.#shouldReconnect(0, err) !== 'number') return;
 
     this.emit('reconnecting');
@@ -362,6 +395,10 @@ export default class RedisSocket extends EventEmitter {
       this.#socket = undefined;
     }
 
+    OTelMetrics.instance.recordConnectionCount(-1, {
+      host: this.host,
+      port: this.port,
+    });
     this.emit('end');
   }