From aa0e0d1e98b90cd71a0a781a837598ef1d7992a8 Mon Sep 17 00:00:00 2001 From: Abdelrahman Awad Date: Sat, 4 Jul 2026 00:44:39 -0400 Subject: [PATCH 1/2] ref(redis): expose diagnostics-channel subscription as an integration Stop exporting the raw subscribeRedisDiagnosticChannels from @sentry/server-utils and expose a redisIntegration plugin instead, mirroring the mysql2 and graphql subscribers. The node and deno integrations now extend that base plugin rather than calling subscribe directly (the raw export predated the extendIntegration API). Drop the redundant subscribed guard, the activeUnbinds bookkeeping, the module-level response hook, and the _resetRedisDiagnosticChannelsForTesting export. The subscriber has a single deduped setupOnce caller, so the guard protected a path that can't occur; the response hook is captured via closure. Tests subscribe once in beforeAll and rely on process isolation instead of unbind/reset, matching the graphql subscriber tests. --- packages/deno/src/integrations/redis.ts | 25 +++--- .../src/integrations/tracing/redis/index.ts | 23 ++--- .../tracing/redis-ioredis-gating.test.ts | 6 -- packages/server-utils/src/index.ts | 18 +--- packages/server-utils/src/redis/index.ts | 40 +++++++++ .../src/redis/redis-dc-subscriber.ts | 89 ++++++++----------- .../test/redis/redis-dc-subscriber.test.ts | 44 ++++----- 7 files changed, 113 insertions(+), 132 deletions(-) create mode 100644 packages/server-utils/src/redis/index.ts diff --git a/packages/deno/src/integrations/redis.ts b/packages/deno/src/integrations/redis.ts index 82ea2c14b1b9..98479eb857cc 100644 --- a/packages/deno/src/integrations/redis.ts +++ b/packages/deno/src/integrations/redis.ts @@ -1,11 +1,7 @@ -// * import so that loading this module doesn't error on Deno versions -// lacking `tracingChannel` (added in Deno 1.44.3). -// On older runtimes the integration becomes a no-op. -import * as dc from 'node:diagnostics_channel'; -import type { RedisDiagnosticChannelResponseHook, RedisTracingChannelFactory } from '@sentry/server-utils'; -import { subscribeRedisDiagnosticChannels } from '@sentry/server-utils'; +import type { RedisDiagnosticChannelResponseHook } from '@sentry/server-utils'; +import { redisIntegration as redisChannelIntegration } from '@sentry/server-utils'; import type { Integration, IntegrationFn } from '@sentry/core'; -import { defineIntegration } from '@sentry/core'; +import { defineIntegration, extendIntegration } from '@sentry/core'; import { setAsyncLocalStorageAsyncContextStrategy } from '../async'; const INTEGRATION_NAME = 'DenoRedis' as const; @@ -19,18 +15,17 @@ export interface DenoRedisIntegrationOptions { } const _denoRedisIntegration = ((options: DenoRedisIntegrationOptions = {}) => { - return { + // The diagnostics_channel subscription lives in server-utils so it is shared across runtimes; we + // extend it here to install Deno's AsyncLocalStorage async-context strategy, which the channel + // binding reads via `getTracingChannelBinding`. `extendIntegration` runs the base `setupOnce` + // first, but its subscribe is deferred a tick when no binding exists yet, so the strategy set + // synchronously below is in place by the time the deferred subscribe runs. + return extendIntegration(redisChannelIntegration({ responseHook: options.responseHook }), { name: INTEGRATION_NAME, setupOnce() { - if (!dc.tracingChannel) { - return; - } - // The span is bound into Deno's AsyncLocalStorage context via the async-context - // strategy's `getTracingChannelBinding`, so the native channel can be passed directly. setAsyncLocalStorageAsyncContextStrategy(); - subscribeRedisDiagnosticChannels(dc.tracingChannel as RedisTracingChannelFactory, options.responseHook); }, - }; + }); }) satisfies IntegrationFn; /** diff --git a/packages/node/src/integrations/tracing/redis/index.ts b/packages/node/src/integrations/tracing/redis/index.ts index 7595d252a642..49caf1822e93 100644 --- a/packages/node/src/integrations/tracing/redis/index.ts +++ b/packages/node/src/integrations/tracing/redis/index.ts @@ -1,7 +1,7 @@ import type { IntegrationFn } from '@sentry/core'; -import { defineIntegration, waitForTracingChannelBinding } from '@sentry/core'; +import { defineIntegration, extendIntegration } from '@sentry/core'; import * as dc from 'node:diagnostics_channel'; -import { subscribeRedisDiagnosticChannels, type RedisTracingChannelFactory } from '@sentry/server-utils'; +import { redisIntegration as redisChannelIntegration } from '@sentry/server-utils'; import { generateInstrumentOnce } from '@sentry/node-core'; import { isDiagnosticsChannelInjectionEnabled } from '../../../sdk/diagnosticsChannelInjection'; import { cacheResponseHook, type RedisOptions, setRedisOptions } from './cache'; @@ -41,17 +41,6 @@ export const instrumentRedis = Object.assign( instrumentIORedis(); } instrumentRedisModule(); - // node-redis >= 5.12.0 and ioredis >= 5.11.0 publish via diagnostics_channel. - // `bindTracingChannelToSpan` (inside the subscriber) makes the span the active - // OTel context via `bindStore`, which needs the Sentry OTel context manager to - // be registered — `initOpenTelemetry()` does that after integration `setupOnce`, - // so defer to the next tick. - // Check this here to ensure this does not fail at runtime for Node <= 18.18.0 - if (dc.tracingChannel) { - waitForTracingChannelBinding(() => { - subscribeRedisDiagnosticChannels(dc.tracingChannel as RedisTracingChannelFactory, cacheResponseHook); - }); - } // todo: implement them gradually // new LegacyRedisInstrumentation({}), @@ -60,13 +49,17 @@ export const instrumentRedis = Object.assign( ); const _redisIntegration = ((options: RedisOptions = {}) => { - return { + // The diagnostics_channel subscription (node-redis >= 5.12.0, ioredis >= 5.11.0) lives in + // server-utils so it is shared across server runtimes; we extend it here to also run the vendored + // OTel patchers for older client versions. `cacheResponseHook` reads options set in the extension's + // `setupOnce` below, but it only runs at command time, by which point those options are set. + return extendIntegration(redisChannelIntegration({ responseHook: cacheResponseHook }), { name: INTEGRATION_NAME, setupOnce() { setRedisOptions(options); instrumentRedis(); }, - }; + }); }) satisfies IntegrationFn; /** diff --git a/packages/node/test/integrations/tracing/redis-ioredis-gating.test.ts b/packages/node/test/integrations/tracing/redis-ioredis-gating.test.ts index 9f2cd3692b15..93e671293490 100644 --- a/packages/node/test/integrations/tracing/redis-ioredis-gating.test.ts +++ b/packages/node/test/integrations/tracing/redis-ioredis-gating.test.ts @@ -20,12 +20,6 @@ vi.mock('@sentry/node-core', async importOriginal => { }; }); -// The >=5.11.0 diagnostics_channel subscription is irrelevant here; keep it inert. -vi.mock('@sentry/server-utils', async importOriginal => { - const actual = (await importOriginal()) as Record; - return { ...actual, subscribeRedisDiagnosticChannels: () => undefined }; -}); - import { instrumentRedis } from '../../../src/integrations/tracing/redis'; describe('instrumentRedis ioredis gating', () => { diff --git a/packages/server-utils/src/index.ts b/packages/server-utils/src/index.ts index ead82f39a4d7..33ca16bf11a6 100644 --- a/packages/server-utils/src/index.ts +++ b/packages/server-utils/src/index.ts @@ -5,22 +5,8 @@ */ export { mongooseIntegration } from './mongoose'; -export { - IOREDIS_DC_CHANNEL_COMMAND, - IOREDIS_DC_CHANNEL_CONNECT, - REDIS_DC_CHANNEL_BATCH, - REDIS_DC_CHANNEL_COMMAND, - REDIS_DC_CHANNEL_CONNECT, - subscribeRedisDiagnosticChannels, -} from './redis/redis-dc-subscriber'; -export type { - IORedisCommandData, - RedisBatchData, - RedisCommandData, - RedisConnectData, - RedisDiagnosticChannelResponseHook, - RedisTracingChannelFactory, -} from './redis/redis-dc-subscriber'; +export { redisIntegration, type RedisDiagnosticChannelsOptions } from './redis'; +export type { RedisDiagnosticChannelResponseHook } from './redis/redis-dc-subscriber'; export { defaultDbStatementSerializer } from './redis/redis-statement-serializer'; export { bindTracingChannelToSpan } from './tracing-channel'; export type { diff --git a/packages/server-utils/src/redis/index.ts b/packages/server-utils/src/redis/index.ts new file mode 100644 index 000000000000..b39f0301d1d3 --- /dev/null +++ b/packages/server-utils/src/redis/index.ts @@ -0,0 +1,40 @@ +import { defineIntegration, type IntegrationFn, waitForTracingChannelBinding } from '@sentry/core'; +import * as dc from 'node:diagnostics_channel'; +import { type RedisDiagnosticChannelResponseHook, subscribeRedisDiagnosticChannels } from './redis-dc-subscriber'; + +/** Options controlling the redis diagnostics-channel subscription. */ +export interface RedisDiagnosticChannelsOptions { + /** + * Optional hook invoked once the redis command response arrives. Useful for attaching + * response-derived attributes (e.g. cache hit/miss, payload size). + */ + responseHook?: RedisDiagnosticChannelResponseHook; +} + +const _redisIntegration = ((options: RedisDiagnosticChannelsOptions = {}) => { + return { + name: 'Redis', + setupOnce() { + // Bail on runtimes without `tracingChannel` (Node <= 18.18.0, Deno < 1.44.3). + if (!dc.tracingChannel) { + return; + } + + // Subscribe to node-redis (>= 5.12.0) and ioredis (>= 5.11.0) native tracing channels. + // This is a no-op on versions that don't publish to the channels, so it is always safe to call. + waitForTracingChannelBinding(() => { + subscribeRedisDiagnosticChannels(dc.tracingChannel, options.responseHook); + }); + }, + }; +}) satisfies IntegrationFn; + +/** + * Auto-instrument the [redis](https://www.npmjs.com/package/redis) and + * [ioredis](https://www.npmjs.com/package/ioredis) libraries via their native + * `node:diagnostics_channel` tracing channels (node-redis >= 5.12.0, ioredis >= 5.11.0). + * + * On older client versions the channels are never published to, so this integration is inert and + * the vendored OTel instrumentation handles instrumentation instead. + */ +export const redisIntegration = defineIntegration(_redisIntegration); diff --git a/packages/server-utils/src/redis/redis-dc-subscriber.ts b/packages/server-utils/src/redis/redis-dc-subscriber.ts index 1dded05ea4a8..b1736cd74242 100644 --- a/packages/server-utils/src/redis/redis-dc-subscriber.ts +++ b/packages/server-utils/src/redis/redis-dc-subscriber.ts @@ -7,8 +7,7 @@ import { SERVER_PORT, } from '@sentry/conventions/attributes'; import type { Span } from '@sentry/core'; -import { debug, SEMANTIC_ATTRIBUTE_SENTRY_OP, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, startInactiveSpan } from '@sentry/core'; -import { DEBUG_BUILD } from '../debug-build'; +import { SEMANTIC_ATTRIBUTE_SENTRY_OP, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, startInactiveSpan } from '@sentry/core'; import { bindTracingChannelToSpan } from '../tracing-channel'; // Channel names published by node-redis >= 5.12.0 and ioredis >= 5.11.0. @@ -106,10 +105,6 @@ export type RedisDiagnosticChannelResponseHook = ( */ export type RedisTracingChannelFactory = (name: string) => TracingChannel; -let subscribed = false; -let currentResponseHook: RedisDiagnosticChannelResponseHook | undefined; -let activeUnbinds: Array<() => void> = []; - /** * Subscribe Sentry span handlers to node-redis and ioredis diagnostics-channel * events: `node-redis:command`/`:batch`/`:connect` (published by node-redis @@ -118,46 +113,37 @@ let activeUnbinds: Array<() => void> = []; * On older client versions the channels are never published to, so subscribers * are inert — there is no double-instrumentation against any IITM-based * patcher gated to those older versions. - * - * Idempotent: subsequent calls update the response hook but do not - * re-subscribe. */ export function subscribeRedisDiagnosticChannels( tracingChannel: RedisTracingChannelFactory, responseHook?: RedisDiagnosticChannelResponseHook, ): void { - currentResponseHook = responseHook; - if (subscribed) return; - subscribed = true; - - try { - // node-redis: command name appears as args[0] in the channel payload, so - // strip it before the statement and response hook see it. - activeUnbinds.push( - setupCommandChannel(tracingChannel, REDIS_DC_CHANNEL_COMMAND, data => data.args.slice(1)), - setupBatchChannel(tracingChannel, REDIS_DC_CHANNEL_BATCH, data => - data.batchMode === 'PIPELINE' ? 'PIPELINE' : 'MULTI', - ), - setupConnectChannel(tracingChannel, REDIS_DC_CHANNEL_CONNECT), - // ioredis: args already exclude the command name; no slicing needed. And - // ioredis has no separate batch channel — pipeline/MULTI metadata rides - // on the per-command payload via `batchMode`/`batchSize`. - setupCommandChannel(tracingChannel, IOREDIS_DC_CHANNEL_COMMAND, data => data.args), - setupConnectChannel(tracingChannel, IOREDIS_DC_CHANNEL_CONNECT), - ); - } catch { - // The factory may rely on `node:diagnostics_channel`, which isn't always - // available. Fail closed; the SDK simply won't emit redis spans here. - DEBUG_BUILD && debug.log('Redis node:diagnostics_channel subscription failed.'); - } + // node-redis: command name appears as args[0] in the channel payload, so + // strip it before the statement and response hook see it. + setupCommandChannel( + tracingChannel, + REDIS_DC_CHANNEL_COMMAND, + data => data.args.slice(1), + responseHook, + ); + setupBatchChannel(tracingChannel, REDIS_DC_CHANNEL_BATCH, data => + data.batchMode === 'PIPELINE' ? 'PIPELINE' : 'MULTI', + ); + setupConnectChannel(tracingChannel, REDIS_DC_CHANNEL_CONNECT); + // ioredis: args already exclude the command name; no slicing needed. And + // ioredis has no separate batch channel — pipeline/MULTI metadata rides + // on the per-command payload via `batchMode`/`batchSize`. + setupCommandChannel(tracingChannel, IOREDIS_DC_CHANNEL_COMMAND, data => data.args, responseHook); + setupConnectChannel(tracingChannel, IOREDIS_DC_CHANNEL_CONNECT); } function setupCommandChannel( tracingChannel: RedisTracingChannelFactory, channelName: string, getCommandArgs: (data: T) => string[], -): () => void { - return bindTracingChannelToSpan( + responseHook?: RedisDiagnosticChannelResponseHook, +): void { + bindTracingChannelToSpan( tracingChannel(channelName), data => { // `args` is already sanitized by the publishing library (node-redis / @@ -183,18 +169,18 @@ function setupCommandChannel( captureError: false, beforeSpanEnd(span, data) { if ('error' in data) return; - runResponseHook(span, data.command, getCommandArgs(data), data.result); + runResponseHook(responseHook, span, data.command, getCommandArgs(data), data.result); }, }, - ).unbind; + ); } function setupBatchChannel( tracingChannel: RedisTracingChannelFactory, channelName: string, getOperationName: (data: RedisBatchData) => string, -): () => void { - return bindTracingChannelToSpan( +): void { + bindTracingChannelToSpan( tracingChannel(channelName), data => { return startInactiveSpan({ @@ -212,11 +198,11 @@ function setupBatchChannel( }); }, { captureError: false }, - ).unbind; + ); } -function setupConnectChannel(tracingChannel: RedisTracingChannelFactory, channelName: string): () => void { - return bindTracingChannelToSpan( +function setupConnectChannel(tracingChannel: RedisTracingChannelFactory, channelName: string): void { + bindTracingChannelToSpan( tracingChannel(channelName), data => { return startInactiveSpan({ @@ -231,11 +217,16 @@ function setupConnectChannel(tracingChannel: RedisTracingChannelFactory, channel }); }, { captureError: false }, - ).unbind; + ); } -function runResponseHook(span: Span, command: string, args: string[], result: unknown): void { - const hook = currentResponseHook; +function runResponseHook( + hook: RedisDiagnosticChannelResponseHook | undefined, + span: Span, + command: string, + args: string[], + result: unknown, +): void { if (!hook) return; try { hook(span, command, args, result); @@ -243,11 +234,3 @@ function runResponseHook(span: Span, command: string, args: string[], result: un // never let user hooks break instrumentation } } - -/** Test-only: detach all channel bindings and reset module-local subscribe state. */ -export function _resetRedisDiagnosticChannelsForTesting(): void { - activeUnbinds.forEach(unbind => unbind()); - activeUnbinds = []; - subscribed = false; - currentResponseHook = undefined; -} diff --git a/packages/server-utils/test/redis/redis-dc-subscriber.test.ts b/packages/server-utils/test/redis/redis-dc-subscriber.test.ts index 19a64c9c60b7..cc2e25568d63 100644 --- a/packages/server-utils/test/redis/redis-dc-subscriber.test.ts +++ b/packages/server-utils/test/redis/redis-dc-subscriber.test.ts @@ -17,9 +17,8 @@ import { spanToJSON, startSpan, } from '@sentry/core'; -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; import { - _resetRedisDiagnosticChannelsForTesting, IOREDIS_DC_CHANNEL_COMMAND, IOREDIS_DC_CHANNEL_CONNECT, REDIS_DC_CHANNEL_BATCH, @@ -127,23 +126,31 @@ async function traceCommand( const factory = tracingChannel as RedisTracingChannelFactory; +// The response hook is bound into the channel handlers when we subscribe, so it must be a stable +// reference for the whole file; `vi.clearAllMocks` in `afterEach` resets its recorded calls per test. +const responseHook = vi.fn(); + describe('subscribeRedisDiagnosticChannels', () => { - let responseHook: ReturnType; let captureExceptionSpy: ReturnType; - // `node:diagnostics_channel` channels are process-global. `_reset…` calls each binding's `unbind`, - // so we can subscribe and fully detach per test without handlers leaking across tests. - beforeEach(() => { + // The subscriber binds handlers onto process-global channels with no teardown, so subscribe once + // here, mirroring production where `setupOnce` subscribes a single time. Per-test we only reset the + // client, scopes, and mock call history (in `afterEach`), so nothing leaks between tests. + beforeAll(() => { installTestAsyncContextStrategy(); + subscribeRedisDiagnosticChannels(factory, responseHook); + }); + + afterAll(() => { + setAsyncContextStrategy(undefined); + }); + + beforeEach(() => { initTestClient(); - responseHook = vi.fn(); captureExceptionSpy = vi.spyOn(SentryCore, 'captureException').mockReturnValue('event-id'); - subscribeRedisDiagnosticChannels(factory, responseHook); }); afterEach(() => { - _resetRedisDiagnosticChannelsForTesting(); - setAsyncContextStrategy(undefined); getCurrentScope().clear(); getCurrentScope().setClient(undefined); getGlobalScope().clear(); @@ -261,21 +268,4 @@ describe('subscribeRedisDiagnosticChannels', () => { expect(spanToJSON(span!).timestamp).toBeDefined(); }); }); - - describe('idempotency', () => { - it('does not re-subscribe on a second call, but updates the response hook', async () => { - const secondHook = vi.fn(); - subscribeRedisDiagnosticChannels(factory, secondHook); - - const { span } = await traceCommand( - REDIS_DC_CHANNEL_COMMAND, - { command: 'GET', args: ['GET', 'k'] }, - { result: 'v' }, - ); - - expect(secondHook).toHaveBeenCalledTimes(1); - expect(secondHook).toHaveBeenCalledWith(span, 'GET', ['k'], 'v'); - expect(responseHook).not.toHaveBeenCalled(); - }); - }); }); From 491bf179277cea2343fad13760e6d5fdd4bdca7b Mon Sep 17 00:00:00 2001 From: Abdelrahman Awad Date: Sat, 4 Jul 2026 00:54:26 -0400 Subject: [PATCH 2/2] ref(redis): trim redundant comments --- packages/server-utils/src/redis/index.ts | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/packages/server-utils/src/redis/index.ts b/packages/server-utils/src/redis/index.ts index b39f0301d1d3..9a5292048712 100644 --- a/packages/server-utils/src/redis/index.ts +++ b/packages/server-utils/src/redis/index.ts @@ -15,13 +15,11 @@ const _redisIntegration = ((options: RedisDiagnosticChannelsOptions = {}) => { return { name: 'Redis', setupOnce() { - // Bail on runtimes without `tracingChannel` (Node <= 18.18.0, Deno < 1.44.3). + // Bail on runtimes without `tracingChannel` (Node <= 18.18.0). if (!dc.tracingChannel) { return; } - // Subscribe to node-redis (>= 5.12.0) and ioredis (>= 5.11.0) native tracing channels. - // This is a no-op on versions that don't publish to the channels, so it is always safe to call. waitForTracingChannelBinding(() => { subscribeRedisDiagnosticChannels(dc.tracingChannel, options.responseHook); }); @@ -33,8 +31,5 @@ const _redisIntegration = ((options: RedisDiagnosticChannelsOptions = {}) => { * Auto-instrument the [redis](https://www.npmjs.com/package/redis) and * [ioredis](https://www.npmjs.com/package/ioredis) libraries via their native * `node:diagnostics_channel` tracing channels (node-redis >= 5.12.0, ioredis >= 5.11.0). - * - * On older client versions the channels are never published to, so this integration is inert and - * the vendored OTel instrumentation handles instrumentation instead. */ export const redisIntegration = defineIntegration(_redisIntegration);