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..9a5292048712 --- /dev/null +++ b/packages/server-utils/src/redis/index.ts @@ -0,0 +1,35 @@ +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). + if (!dc.tracingChannel) { + return; + } + + 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). + */ +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(); - }); - }); });