wip: add OpenTelemetry metrics instrumentation

Author: PavelPashov

package-lock.json

@@ -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/opentelemetry';

packages/client/index.ts

@@ -1,5 +1,5 @@
 import COMMANDS from '../commands';
-import RedisSocket, { RedisSocketOptions } from './socket';
+import RedisSocket, { RedisSocketOptions, RedisTcpSocketOptions } from './socket';
 import { BasicAuth, CredentialsError, CredentialsProvider, StreamingCredentialsProvider, UnableToObtainNewCredentialsError, Disposable } from '../authx';
 import RedisCommandsQueue, { CommandOptions } from './commands-queue';
 import { EventEmitter } from 'node:events';
@@ -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 { OTelMetrics } from '../opentelemetry/metrics';
 
 export interface RedisClientOptions<
   M extends RedisModules = RedisModules,
@@ -1064,21 +1065,47 @@ export default class RedisClient<
     args: ReadonlyArray<RedisArgument>,
     options?: CommandOptions
   ): Promise<T> {
+    const recordOperation = OTelMetrics.instance.createRecordOperationDuration(args, {
+      host: (this._self.#options.socket as RedisTcpSocketOptions)?.host || "",
+      port:
+        (
+          this._self.#options.socket as RedisTcpSocketOptions
+        )?.port?.toString() || "",
+      db: this._self.#selectedDB.toString(),
+    });
+
     if (!this._self.#socket.isOpen) {
+      recordOperation(new ClientClosedError());
       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
+    ) {
+      recordOperation(new ClientOfflineError());
       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);
+    OTelMetrics.instance.recordPendingRequests(1);
+
+    const trackedPromise = promise.then((reply) => {
+      recordOperation();
+      return reply;
+    }).catch((err) => {
+      recordOperation(err);
+      throw err;
+    }).finally(() => {
+      OTelMetrics.instance.recordPendingRequests(-1);
+    });
+
     this._self.#scheduleWrite();
-    return promise;
+    return trackedPromise;
   }
 
   async SELECT(db: number): Promise<void> {

packages/client/lib/client/index.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/metrics';
 
 type NetOptions = {
   tls?: false;
@@ -215,6 +216,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 +230,8 @@ export default class RedisSocket extends EventEmitter {
         this.#isReady = true;
         this.#socketEpoch++;
         this.emit('ready');
+        OTelMetrics.instance.recordConnectionCount(1);
+        OTelMetrics.instance.recordConnectionCreateTime(performance.now() - started);
       } catch (err) {
         const retryIn = this.#shouldReconnect(retries++, err as Error);
         if (typeof retryIn !== 'number') {
@@ -304,6 +308,10 @@ export default class RedisSocket extends EventEmitter {
     this.#isReady = false;
     this.emit('error', err);
 
+    if (wasReady) {
+      OTelMetrics.instance.recordConnectionCount(-1);
+    }
+
     if (!wasReady || !this.#isOpen || typeof this.#shouldReconnect(0, err) !== 'number') return;
 
     this.emit('reconnecting');
@@ -362,6 +370,7 @@ export default class RedisSocket extends EventEmitter {
       this.#socket = undefined;
     }
 
+    OTelMetrics.instance.recordConnectionCount(-1);
     this.emit('end');
   }