feat(node): use diagnostics_channel for redis >= 5.12.0

logaretm · logaretm · commit 291507cfe8ed · 2026-04-28T16:25:33.000-04:00
node-redis 5.12.0 publishes command, batch, and connect events via
node:diagnostics_channel. Subscribe to those channels via
@sentry/opentelemetry/tracing-channel to produce spans without IITM-
based monkey-patching, which lets the redis integration work on
runtimes that don't support IITM (Bun, Deno, Cloudflare Workers).

The existing OTel patcher is narrowed to '&lt;5.12.0' so it does not
double-instrument when both paths are present. The DC subscription is
deferred to the next microtask so it runs after initOpenTelemetry()
sets up the Sentry context manager (required for bindStore).
diff --git a/packages/node/rollup.npm.config.mjs b/packages/node/rollup.npm.config.mjs
@@ -6,6 +6,7 @@ export default [
     makeBaseNPMConfig({
       entrypoints: ['src/index.ts', 'src/init.ts', 'src/preload.ts'],
       packageSpecificConfig: {
+        external: [/^@sentry\/opentelemetry/],
         output: {
           // set exports to 'named' or 'auto' so that rollup doesn't warn
           exports: 'named',
diff --git a/packages/node/src/integrations/tracing/redis/index.ts b/packages/node/src/integrations/tracing/redis/index.ts
@@ -23,6 +23,7 @@ import {
 import type { IORedisInstrumentationConfig } from './vendored/types';
 import { IORedisInstrumentation } from './vendored/ioredis-instrumentation';
 import { RedisInstrumentation } from './vendored/redis-instrumentation';
+import { subscribeRedisDiagnosticChannels } from './redis-dc-subscriber';
 
 interface RedisOptions {
   /**
@@ -116,6 +117,11 @@ export const instrumentRedis = Object.assign(
   (): void => {
     instrumentIORedis();
     instrumentRedisModule();
+    // node-redis >= 5.12.0 publishes via diagnostics_channel. The subscriber uses
+    // `@sentry/opentelemetry/tracing-channel`, which needs the Sentry OTel context manager
+    // to be registered before it can `bindStore`. `initOpenTelemetry()` runs after integration
+    // `setupOnce`, so defer to the next tick.
+    Promise.resolve().then(() => subscribeRedisDiagnosticChannels(cacheResponseHook));
 
     // todo: implement them gradually
     // new LegacyRedisInstrumentation({}),
diff --git a/packages/node/src/integrations/tracing/redis/redis-dc-subscriber.ts b/packages/node/src/integrations/tracing/redis/redis-dc-subscriber.ts
@@ -0,0 +1,231 @@
+import type { Span } from '@opentelemetry/api';
+import {
+  SEMANTIC_ATTRIBUTE_SENTRY_OP,
+  SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN,
+  SPAN_STATUS_ERROR,
+  startSpanManual,
+} from '@sentry/core';
+import { tracingChannel, type TracingChannelContextWithSpan } from '@sentry/opentelemetry/tracing-channel';
+import { defaultDbStatementSerializer } from './vendored/redis-common';
+import {
+  ATTR_DB_STATEMENT,
+  ATTR_DB_SYSTEM,
+  ATTR_NET_PEER_NAME,
+  ATTR_NET_PEER_PORT,
+  DB_SYSTEM_VALUE_REDIS,
+} from './vendored/semconv';
+import type { IORedisInstrumentationConfig } from './vendored/types';
+
+// Channel names as published by node-redis >= 5.12.0.
+// Hardcoded so we don't import `redis` at module-load time.
+const CHANNEL_COMMAND = 'node-redis:command';
+const CHANNEL_BATCH = 'node-redis:batch';
+const CHANNEL_CONNECT = 'node-redis:connect';
+
+const ORIGIN = 'auto.db.redis.diagnostic-channel';
+
+interface CommandData {
+  command: string;
+  args: Array<string | Buffer>;
+  database?: number;
+  serverAddress?: string;
+  serverPort?: number;
+  result?: unknown;
+  error?: Error;
+}
+
+interface BatchData {
+  batchMode?: 'MULTI' | 'PIPELINE';
+  batchSize?: number;
+  database?: number;
+  clientId?: string | number;
+  serverAddress?: string;
+  serverPort?: number;
+  result?: unknown[];
+  error?: Error;
+}
+
+interface ConnectData {
+  serverAddress?: string;
+  serverPort?: number;
+  url?: string;
+  error?: Error;
+}
+
+const NOOP = (): void => {};
+
+let subscribed = false;
+let currentResponseHook: IORedisInstrumentationConfig['responseHook'] | undefined;
+
+/**
+ * Subscribe Sentry handlers to node-redis diagnostics_channel events (>= 5.12.0).
+ *
+ * Uses `@sentry/opentelemetry/tracing-channel` so OTel AsyncLocalStorage context propagates
+ * automatically via `bindStore` — without it, spans created in `start` would not become
+ * the active context for subsequent operations.
+ *
+ * Safe on every runtime that exposes `node:diagnostics_channel` (Node, Bun, Deno, Workers).
+ * In node-redis < 5.12.0 the channels are never published to, so subscribers are inert and
+ * there is no double-instrumentation against the IITM-based patcher (gated to < 5.12.0).
+ */
+export function subscribeRedisDiagnosticChannels(
+  responseHook?: IORedisInstrumentationConfig['responseHook'],
+): void {
+  currentResponseHook = responseHook;
+  if (subscribed) return;
+
+  try {
+    setupCommandChannel();
+    setupBatchChannel();
+    setupConnectChannel();
+    subscribed = true;
+  } catch {
+    // tracingChannel from @sentry/opentelemetry requires `node:diagnostics_channel`.
+    // On runtimes where it isn't available, fail closed.
+  }
+}
+
+function setupCommandChannel(): void {
+  const channel = tracingChannel<CommandData>(CHANNEL_COMMAND, data => {
+    const statement = safeSerialize(data.command, data.args);
+    return startSpanManual(
+      {
+        name: `redis-${data.command}`,
+        attributes: {
+          [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: ORIGIN,
+          [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'db.redis',
+          [ATTR_DB_SYSTEM]: DB_SYSTEM_VALUE_REDIS,
+          ...(statement != null ? { [ATTR_DB_STATEMENT]: statement } : {}),
+          ...(data.serverAddress != null ? { [ATTR_NET_PEER_NAME]: data.serverAddress } : {}),
+          ...(data.serverPort != null ? { [ATTR_NET_PEER_PORT]: data.serverPort } : {}),
+        },
+      },
+      span => span,
+    ) as Span;
+  });
+
+  channel.subscribe({
+    start: NOOP,
+    asyncStart: NOOP,
+    end: NOOP,
+    asyncEnd: data => {
+      const span = data._sentrySpan;
+      if (!span) return;
+      runResponseHook(span, data.command, data.args, data.result);
+      span.end();
+    },
+    error: data => {
+      const span = data._sentrySpan;
+      if (!span) return;
+      if (data.error) {
+        span.setStatus({ code: SPAN_STATUS_ERROR, message: data.error.message });
+      }
+      span.end();
+    },
+  });
+}
+
+function setupBatchChannel(): void {
+  const channel = tracingChannel<BatchData>(CHANNEL_BATCH, data => {
+    const operationName = data.batchMode === 'PIPELINE' ? 'PIPELINE' : 'MULTI';
+
+    return startSpanManual(
+      {
+        name: operationName,
+        attributes: {
+          [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: ORIGIN,
+          [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'db.redis',
+          [ATTR_DB_SYSTEM]: DB_SYSTEM_VALUE_REDIS,
+          ...(data.batchSize != null ? { 'db.redis.batch_size': data.batchSize } : {}),
+          ...(data.serverAddress != null ? { [ATTR_NET_PEER_NAME]: data.serverAddress } : {}),
+          ...(data.serverPort != null ? { [ATTR_NET_PEER_PORT]: data.serverPort } : {}),
+        },
+      },
+      span => span,
+    ) as Span;
+  });
+
+  channel.subscribe({
+    start: NOOP,
+    asyncStart: NOOP,
+    end: NOOP,
+    asyncEnd: data => {
+      data._sentrySpan?.end();
+    },
+    error: data => {
+      const span = data._sentrySpan;
+      if (!span) return;
+      if (data.error) {
+        span.setStatus({ code: SPAN_STATUS_ERROR, message: data.error.message });
+      }
+      span.end();
+    },
+  });
+}
+
+function setupConnectChannel(): void {
+  const channel = tracingChannel<ConnectData>(CHANNEL_CONNECT, data => {
+    return startSpanManual(
+      {
+        name: 'redis-connect',
+        attributes: {
+          [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: ORIGIN,
+          [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'db.redis.connect',
+          [ATTR_DB_SYSTEM]: DB_SYSTEM_VALUE_REDIS,
+          ...(data.serverAddress != null ? { [ATTR_NET_PEER_NAME]: data.serverAddress } : {}),
+          ...(data.serverPort != null ? { [ATTR_NET_PEER_PORT]: data.serverPort } : {}),
+        },
+      },
+      span => span,
+    ) as Span;
+  });
+
+  channel.subscribe({
+    start: NOOP,
+    asyncStart: NOOP,
+    end: NOOP,
+    asyncEnd: data => {
+      data._sentrySpan?.end();
+    },
+    error: data => {
+      const span = data._sentrySpan;
+      if (!span) return;
+      if (data.error) {
+        span.setStatus({ code: SPAN_STATUS_ERROR, message: data.error.message });
+      }
+      span.end();
+    },
+  });
+}
+
+function runResponseHook(
+  span: Span,
+  command: string,
+  args: Array<string | Buffer>,
+  result: unknown,
+): void {
+  const hook = currentResponseHook;
+  if (!hook) return;
+  try {
+    hook(span, command, args as unknown as Parameters<typeof hook>[2], result);
+  } catch {
+    // never let user hooks break instrumentation
+  }
+}
+
+function safeSerialize(command: string, args: Array<string | Buffer>): string | undefined {
+  try {
+    return defaultDbStatementSerializer(command, args);
+  } catch {
+    return undefined;
+  }
+}
+
+// Test-only helper.
+export function _resetRedisDiagnosticChannelsForTesting(): void {
+  subscribed = false;
+  currentResponseHook = undefined;
+}
+
+// Suppress unused-import lint when only used in types.
+export type { TracingChannelContextWithSpan };
diff --git a/packages/node/src/integrations/tracing/redis/vendored/redis-instrumentation.ts b/packages/node/src/integrations/tracing/redis/vendored/redis-instrumentation.ts
@@ -368,7 +368,7 @@ class RedisInstrumentationV4_V5 extends InstrumentationBase<RedisInstrumentation
 
     const multiCommanderModule = new InstrumentationNodeModuleFile(
       `${basePackageName}/dist/lib/client/multi-command.js`,
-      ['^1.0.0', '^5.0.0'],
+      ['^1.0.0', '>=5.0.0 <5.12.0'],
       (moduleExports: any) => {
         const redisClientMultiCommandPrototype = moduleExports?.default?.prototype;
         if (isWrapped(redisClientMultiCommandPrototype?.exec)) {
@@ -398,7 +398,7 @@ class RedisInstrumentationV4_V5 extends InstrumentationBase<RedisInstrumentation
 
     const clientIndexModule = new InstrumentationNodeModuleFile(
       `${basePackageName}/dist/lib/client/index.js`,
-      ['^1.0.0', '^5.0.0'],
+      ['^1.0.0', '>=5.0.0 <5.12.0'],
       (moduleExports: any) => {
         const redisClientPrototype = moduleExports?.default?.prototype;
         if (redisClientPrototype?.multi) {
@@ -436,7 +436,7 @@ class RedisInstrumentationV4_V5 extends InstrumentationBase<RedisInstrumentation
 
     return new InstrumentationNodeModuleDefinition(
       basePackageName,
-      ['^1.0.0', '^5.0.0'],
+      ['^1.0.0', '>=5.0.0 <5.12.0'],
       (moduleExports: any) => moduleExports,
       () => {},
       [commanderModuleFile, multiCommanderModule, clientIndexModule],
