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';

packages/client/index.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/enterprise-maintenance-manager.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, OTelMetrics } from '../opentelemetry';
 
 export interface RedisClientOptions<
   M extends RedisModules = RedisModules,
@@ -1060,25 +1061,55 @@ 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> {
+    const recordOperation = OTelMetrics.instance.createRecordOperationDuration(args, this._self._getClientOTelAttributes());
+
     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, this._self._getClientOTelAttributes());
+
+    const trackedPromise = promise.then((reply) => {
+      recordOperation();
+      return reply;
+    }).catch((err) => {
+      recordOperation(err);
+      throw err;
+    }).finally(() => {
+      OTelMetrics.instance.recordPendingRequests(-1, this._self._getClientOTelAttributes());
+    });
+
     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';
 
 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');
   }
 

packages/client/lib/client/socket.ts

@@ -14,6 +14,7 @@ import { BasicCommandParser } from '../client/parser';
 import { ASKING_CMD } from '../commands/ASKING';
 import SingleEntryCache from '../single-entry-cache'
 import { WithCommands, WithFunctions, WithModules, WithScripts } from '../client';
+import { METRIC_ERROR_TYPE, OTelMetrics } from '../opentelemetry';
 
 interface ClusterCommander<
   M extends RedisModules,
@@ -433,6 +434,7 @@ export default class RedisCluster<
         }
 
         if (err.message.startsWith('ASK')) {
+          OTelMetrics.instance.recordClientErrorsHandled(METRIC_ERROR_TYPE.ASK, client._getClientOTelAttributes());
           const address = err.message.substring(err.message.lastIndexOf(' ') + 1);
           let redirectTo = await this._slots.getMasterByAddress(address);
           if (!redirectTo) {
@@ -450,6 +452,7 @@ export default class RedisCluster<
         }
 
         if (err.message.startsWith('MOVED')) {
+          OTelMetrics.instance.recordClientErrorsHandled(METRIC_ERROR_TYPE.MOVED, client._getClientOTelAttributes());
           await this._slots.rediscover(client);
           client = await this._slots.getClient(firstKey, isReadonly);
           continue;

packages/client/lib/cluster/index.ts

@@ -0,0 +1,30 @@
+import { OTelMetrics } from "./metrics";
+import { ObservabilityConfig } from "./types";
+
+export class OpenTelemetry {
+  private static _instance: OpenTelemetry | null = null;
+
+  // Make sure it's a singleton
+  private constructor() {}
+
+  public static init(config?: ObservabilityConfig) {
+    if (OpenTelemetry._instance) {
+      throw new Error("OpenTelemetry already initialized");
+    }
+
+    let api: typeof import("@opentelemetry/api") | undefined;
+
+    try {
+      api = require("@opentelemetry/api");
+    } catch (err: unknown) {
+      // TODO add custom errors
+      throw new Error("OpenTelemetry not found");
+    }
+
+    OpenTelemetry._instance = new OpenTelemetry();
+    OTelMetrics.init({ api, config });
+  }
+}
+
+export { METRIC_ERROR_TYPE, OTelClientAttributes } from "./types";
+export { OTelMetrics } from "./metrics";