feat: Otel pgとredis自動計装 (#17712)

.config/example.yml

@@ -354,6 +354,30 @@ id: 'aidx'
 #  #headers:
 #  #  authorization: 'Bearer xxxxx'
 #  #sampleRate: 1.0
+#  # Record PostgreSQL query spans. Disabled by default to avoid instrumentation
+#  # overhead when database-level diagnostics are not needed.
+#  #capturePgSpans: false
+#  # Include raw SQL text in PostgreSQL spans. Requires capturePgSpans and is
+#  # disabled by default because non-parameterized queries can expose values to
+#  # the OTLP Collector.
+#  #capturePgStatement: false
+#  # Include PostgreSQL connection-pool spans with an existing parent span.
+#  # Requires capturePgSpans and is disabled by default to avoid noisy spans
+#  # from connection churn.
+#  #capturePgConnectionSpans: false
+#  # Record Redis command spans. Disabled by default because subscribing to the
+#  # ioredis diagnostics channel adds work to every Redis command, including
+#  # commands without a parent span. Enable for development diagnostics or when
+#  # the added overhead is acceptable in production.
+#  #captureRedisCommandSpans: false
+#  # Include Redis startup/reconnect spans. Disabled by default to avoid noisy
+#  # connection churn; these spans are recorded even without an HTTP or job
+#  # parent span.
+#  #captureRedisConnectionSpans: false
+#  # Record Redis commands without an HTTP or job parent span. Disabled by
+#  # default because queue polling and background work can produce many root
+#  # traces; only applies when captureRedisCommandSpans is enabled.
+#  #captureRedisRootSpans: false
 #  #resourceAttributes:
 #  #  deployment.environment: 'production'
 #  #propagateTraceToRemote: false

CHANGELOG.md

@@ -38,6 +38,7 @@
 
 ### Server
 - Enhance: OpenTelemetryで全受信HTTPリクエストを自動計装
+- Enhance: OpenTelemetry を単独で利用する際、PostgreSQL query と Redis command の計装を opt-in で利用できるように
 - Enhance: センシティブメディアの判定を外部サービス ([sensitive-detector](https://github.com/misskey-dev/sensitive-detector)) に分離し、`nsfwjs` / `@tensorflow/tfjs(-node)` の同梱と NSFW 判定モデルを廃止 (#16804)
 - Enhance: バックエンドの `otelForBackend` 設定で OpenTelemetry Traces を OTLP Collector に送信できるように
 - Enhance: OpenTelemetry を単独で利用する際、外向き HTTP リクエストを自動計装するように

packages/backend/package.json

@@ -71,6 +71,7 @@
 		"@opentelemetry/api": "1.9.1",
 		"@opentelemetry/core": "2.8.0",
 		"@opentelemetry/exporter-trace-otlp-proto": "0.214.0",
+		"@opentelemetry/instrumentation-pg": "0.72.0",
 		"@opentelemetry/resources": "2.7.1",
 		"@opentelemetry/sdk-trace-base": "2.7.1",
 		"@opentelemetry/sdk-trace-node": "2.7.1",

packages/backend/src/config.ts

@@ -30,6 +30,12 @@ type OtelBackendConfig = {
 	endpoint?: string;
 	headers?: Record<string, string>;
 	sampleRate?: number;
+	capturePgSpans?: boolean;
+	capturePgStatement?: boolean;
+	capturePgConnectionSpans?: boolean;
+	captureRedisCommandSpans?: boolean;
+	captureRedisConnectionSpans?: boolean;
+	captureRedisRootSpans?: boolean;
 	resourceAttributes?: Record<string, string>;
 	propagateTraceToRemote?: boolean;
 	jobTraceContextMode?: 'link' | 'parent';

packages/backend/src/core/telemetry/adapters/OpenTelemetryAdapter.ts

@@ -8,6 +8,8 @@ import cluster from 'node:cluster';
 import { envOption } from '@/env.js';
 import { registerDiagLogger } from '@/core/telemetry/telemetry-diag.js';
 import { installHttpClientInstrumentation } from '@/core/telemetry/http-client-instrumentation.js';
+import { installDatabaseInstrumentation } from '@/core/telemetry/database-instrumentation.js';
+import { installRedisInstrumentation } from '@/core/telemetry/redis-instrumentation.js';
 import { executeSpan, getQueueTraceContextMode, injectActiveTraceContext, recordSpanError, startSpanWithQueueTraceContext } from '@/core/telemetry/queue-trace-context.js';
 import type { Span, SpanStatusCode, Tracer } from '@opentelemetry/api';
 import type { Resource, ResourceDetector } from '@opentelemetry/resources';
@@ -26,6 +28,8 @@ type OpenTelemetryAdapterDeps = {
 	spanStatusCodeError: SpanStatusCode;
 	shutdownTimeout: number;
 	shutdownHttpClientInstrumentation?: () => void;
+	shutdownDatabaseInstrumentation?: () => void;
+	shutdownRedisInstrumentation?: () => void;
 	queueTraceContext?: QueueTraceContextDeps;
 };
 
@@ -118,6 +122,17 @@ export class OpenTelemetryAdapter implements TelemetryAdapter {
 				spanKindClient: SpanKind.CLIENT,
 				spanStatusCodeError: SpanStatusCode.ERROR,
 			}),
+			// pg のrequire hookとioredis diagnostics channelは、Nest moduleの動的importより前に有効化する。
+			shutdownDatabaseInstrumentation: await installDatabaseInstrumentation(provider, {
+				capturePgSpans: config.capturePgSpans === true,
+				capturePgStatement: config.capturePgStatement === true,
+				capturePgConnectionSpans: config.capturePgConnectionSpans === true,
+			}),
+			shutdownRedisInstrumentation: installRedisInstrumentation(tracer, SpanKind.CLIENT, SpanStatusCode.ERROR, {
+				captureConnectionSpans: config.captureRedisConnectionSpans === true,
+				captureCommandSpans: config.captureRedisCommandSpans === true,
+				requireParentSpan: config.captureRedisRootSpans !== true,
+			}),
 			queueTraceContext: {
 				tracer,
 				propagation,
@@ -168,6 +183,8 @@ export class OpenTelemetryAdapter implements TelemetryAdapter {
 
 	public async shutdown(): Promise<void> {
 		this.deps.shutdownHttpClientInstrumentation?.();
+		this.deps.shutdownDatabaseInstrumentation?.();
+		this.deps.shutdownRedisInstrumentation?.();
 		// BatchSpanProcessorのflushが詰まってもプロセス終了を妨げないよう、上限時間を設ける。
 		// タイムアウト側のtimerは、flushが先に終わった場合にイベントループを無駄に引き留めないようclearする。
 		let timer: NodeJS.Timeout | undefined;

packages/backend/src/core/telemetry/database-instrumentation.ts

@@ -0,0 +1,86 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import type { Span, TracerProvider } from '@opentelemetry/api';
+
+type Instrumentation = {
+	setTracerProvider(provider: TracerProvider): void;
+	enable(): void;
+	disable(): void;
+};
+
+type PgInstrumentationConfig = {
+	enhancedDatabaseReporting: boolean;
+	requireParentSpan: boolean;
+	ignoreConnectSpans: boolean;
+	requestHook?(span: Span): void;
+};
+
+type InstrumentationConstructor = new (config: PgInstrumentationConfig) => Instrumentation;
+
+type InstrumentationDeps = {
+	PgInstrumentation: InstrumentationConstructor;
+};
+
+type DatabaseInstrumentationOptions = {
+	capturePgStatement: boolean;
+	capturePgConnectionSpans: boolean;
+};
+
+type InstallDatabaseInstrumentationOptions = DatabaseInstrumentationOptions & {
+	capturePgSpans: boolean;
+};
+
+/**
+ * pg はアプリケーションが import する前に有効化しないと、require hook 型の
+ * 自動計装がモジュールを patch できない。そのため、
+ * telemetry provider の登録直後にこの関数を呼び出す。
+ */
+export async function installDatabaseInstrumentation(provider: TracerProvider, options: InstallDatabaseInstrumentationOptions): Promise<() => void> {
+	if (!options.capturePgSpans) return () => {};
+
+	const { PgInstrumentation } = await import('@opentelemetry/instrumentation-pg');
+
+	return installInstrumentation(provider, { PgInstrumentation }, options);
+}
+
+export function installInstrumentation(provider: TracerProvider, deps: InstrumentationDeps, options: DatabaseInstrumentationOptions = {
+	capturePgStatement: false,
+	capturePgConnectionSpans: false,
+}): () => void {
+	const instrumentations = [
+		new deps.PgInstrumentation({
+			// SQLパラメータには投稿内容・認証情報などが含まれ得るため、常に記録しない。
+			enhancedDatabaseReporting: false,
+			requireParentSpan: true,
+			ignoreConnectSpans: !options.capturePgConnectionSpans,
+			// instrumentation-pgはSQL本文を無加工で属性へ追加する。明示opt-in時だけ残す。
+			...(options.capturePgStatement ? {} : {
+				requestHook: (span: Span) => {
+					span.setAttribute('db.statement', '[REDACTED]');
+					span.setAttribute('db.query.text', '[REDACTED]');
+				},
+			}),
+		}),
+	];
+
+	try {
+		for (const instrumentation of instrumentations) {
+			instrumentation.setTracerProvider(provider);
+			instrumentation.enable();
+		}
+	} catch (error) {
+		for (const instrumentation of instrumentations) {
+			instrumentation.disable();
+		}
+		throw error;
+	}
+
+	return () => {
+		for (const instrumentation of instrumentations) {
+			instrumentation.disable();
+		}
+	};
+}

packages/backend/src/core/telemetry/redis-instrumentation.ts

@@ -0,0 +1,178 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { tracingChannel } from 'node:diagnostics_channel';
+import { context, trace } from '@opentelemetry/api';
+import type { Span, SpanKind, SpanStatusCode, Tracer } from '@opentelemetry/api';
+
+type IORedisCommandContext = {
+	command: string;
+	args: string[];
+	database: number;
+	serverAddress: string;
+	serverPort: number | undefined;
+};
+
+type IORedisConnectContext = {
+	serverAddress: string;
+	serverPort: number | undefined;
+};
+
+type TracingChannelSubscribers<T extends object> = {
+	start(message: T): void;
+	end(message: T & { error?: unknown }): void;
+	asyncStart(message: T & { error?: unknown }): void;
+	asyncEnd(message: T & { error?: unknown }): void;
+	error(message: T & { error: unknown }): void;
+};
+
+type TracingChannel<T extends object> = {
+	subscribe(subscribers: TracingChannelSubscribers<T>): void;
+	unsubscribe(subscribers: TracingChannelSubscribers<T>): void;
+};
+
+type RedisInstrumentationDeps = {
+	tracingChannel<T extends object>(name: string): TracingChannel<T>;
+	tracer: Pick<Tracer, 'startSpan'>;
+	getActiveSpan(): Span | undefined;
+	spanKindClient: SpanKind;
+	spanStatusCodeError: SpanStatusCode;
+};
+
+type RedisInstrumentationOptions = {
+	captureCommandSpans?: boolean;
+	/** requireParentSpan is the official ioredis instrumentation's default. */
+	requireParentSpan?: boolean;
+	captureConnectionSpans?: boolean;
+};
+
+/**
+ * ioredis 5.11以降が公開する native diagnostics channel を購読する。
+ * バンドル後もioredis自身が発行するイベントを使うため、require hook に
+ * 依存せず、ESM import 経路でもコマンド span を取得できる。
+ */
+export function installRedisInstrumentation(
+	tracer: Pick<Tracer, 'startSpan'>,
+	spanKindClient: SpanKind,
+	spanStatusCodeError: SpanStatusCode,
+	options: RedisInstrumentationOptions = {},
+): () => void {
+	return createRedisInstrumentation({
+		tracingChannel,
+		tracer,
+		getActiveSpan: () => trace.getSpan(context.active()),
+		spanKindClient,
+		spanStatusCodeError,
+	}, options);
+}
+
+export function createRedisInstrumentation(deps: RedisInstrumentationDeps, options: RedisInstrumentationOptions = {}): () => void {
+	const requireParentSpan = options.requireParentSpan ?? true;
+	const cleanup: Array<() => void> = [];
+	if (options.captureCommandSpans === true) {
+		const commandChannel = deps.tracingChannel<IORedisCommandContext>('ioredis:command');
+		const commandSubscribers = createTracingChannelSubscribers(commandChannel, deps, requireParentSpan, message => ({
+			name: message.command,
+			attributes: {
+				'db.system.name': 'redis',
+				'db.namespace': message.database.toString(10),
+				'db.operation.name': message.command,
+				'server.address': message.serverAddress,
+				...(message.serverPort != null ? { 'server.port': message.serverPort } : {}),
+			},
+		}));
+		cleanup.push(() => commandChannel.unsubscribe(commandSubscribers));
+	}
+	if (options.captureConnectionSpans === true) {
+		const connectChannel = deps.tracingChannel<IORedisConnectContext>('ioredis:connect');
+		// Connection spans are explicitly opt-in and should include startup and reconnect attempts.
+		const connectSubscribers = createTracingChannelSubscribers(connectChannel, deps, false, message => ({
+			name: 'connect',
+			attributes: {
+				'db.system.name': 'redis',
+				'db.operation.name': 'connect',
+				'server.address': message.serverAddress,
+				...(message.serverPort != null ? { 'server.port': message.serverPort } : {}),
+			},
+		}));
+		cleanup.push(() => connectChannel.unsubscribe(connectSubscribers));
+	}
+
+	return () => {
+		for (const unsubscribe of cleanup) {
+			unsubscribe();
+		}
+	};
+}
+
+function createTracingChannelSubscribers<T extends object>(
+	channel: TracingChannel<T>,
+	deps: RedisInstrumentationDeps,
+	requireParentSpan: boolean,
+	getSpanOptions: (message: T) => { name: string; attributes: Record<string, string | number> },
+): TracingChannelSubscribers<T> {
+	const spans = new WeakMap<object, { span: Span; recordedError: boolean }>();
+
+	const recordError = (state: { span: Span; recordedError: boolean }, value: unknown): void => {
+		if (state.recordedError) return;
+		const error = toError(value);
+		state.span.recordException(error);
+		state.span.setStatus({ code: deps.spanStatusCodeError, message: error.message });
+		state.span.setAttribute('error.type', getErrorType(error));
+		const statusCode = getRedisErrorStatusCode(error.message);
+		if (statusCode != null) state.span.setAttribute('db.response.status_code', statusCode);
+		state.recordedError = true;
+	};
+
+	const finish = (message: T & { error?: unknown }): void => {
+		const state = spans.get(message);
+		if (state == null) return;
+
+		if (message.error != null) {
+			recordError(state, message.error);
+		}
+		state.span.end();
+		spans.delete(message);
+	};
+
+	const subscribers: TracingChannelSubscribers<T> = {
+		start: (message) => {
+			if (requireParentSpan && deps.getActiveSpan() == null) return;
+
+			const options = getSpanOptions(message);
+			const span = deps.tracer.startSpan(options.name, {
+				kind: deps.spanKindClient,
+				attributes: options.attributes,
+			});
+			spans.set(message, { span, recordedError: false });
+		},
+		// Promiseを返すコマンドでは完了前にもendイベントが来るため、同期例外だけここで閉じる。
+		end: (message) => {
+			if (message.error != null) finish(message);
+		},
+		asyncStart: () => {},
+		asyncEnd: finish,
+		error: (message) => {
+			const state = spans.get(message);
+			if (state == null) return;
+			recordError(state, message.error);
+		},
+	};
+
+	channel.subscribe(subscribers);
+	return subscribers;
+}
+
+function toError(value: unknown): Error {
+	return value instanceof Error ? value : new Error(String(value));
+}
+
+function getRedisErrorStatusCode(message: string): string | undefined {
+	return message.match(/^([A-Z][A-Z0-9_]*)\b/)?.[1];
+}
+
+function getErrorType(error: Error): string {
+	return (error as NodeJS.ErrnoException).code ?? error.name;
+}

packages/backend/test/e2e/telemetry-redis-instrumentation.ts

@@ -0,0 +1,88 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { randomUUID } from 'node:crypto';
+import { describe, expect, test } from 'vitest';
+import { Queue, Worker } from 'bullmq';
+import { SpanKind, SpanStatusCode } from '@opentelemetry/api';
+import { InMemorySpanExporter, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base';
+import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
+import { loadConfig } from '@/config.js';
+import { installRedisInstrumentation } from '@/core/telemetry/redis-instrumentation.js';
+
+const config = loadConfig();
+
+describe('Redis telemetry instrumentation', () => {
+	test('records Redis spans below HTTP and BullMQ worker spans without Redis arguments', async () => {
+		const exporter = new InMemorySpanExporter();
+		const provider = new NodeTracerProvider({
+			spanProcessors: [new SimpleSpanProcessor(exporter)],
+		});
+		provider.register();
+
+		const tracer = provider.getTracer('telemetry-redis-instrumentation-test');
+		const uninstall = installRedisInstrumentation(tracer, SpanKind.CLIENT, SpanStatusCode.ERROR, {
+			captureCommandSpans: true,
+		});
+		const queueName = `telemetry-${randomUUID()}`;
+		const prefix = `telemetry-${randomUUID()}`;
+		const connection = {
+			host: config.redis.host,
+			port: config.redis.port,
+			...(config.redis.password != null ? { password: config.redis.password } : {}),
+		};
+		const queue = new Queue(queueName, { connection, prefix });
+		let worker: Worker | undefined;
+		let httpSpanId: string | undefined;
+		let jobSpanId: string | undefined;
+		const secret = `secret-${randomUUID()}`;
+
+		try {
+			const processed = new Promise<void>((resolve, reject) => {
+				worker = new Worker(queueName, async job => {
+					return await tracer.startActiveSpan('Queue: telemetry test', async jobSpan => {
+						jobSpanId = jobSpan.spanContext().spanId;
+						try {
+							// updateData uses BullMQ's worker-side ioredis client.
+							await job.updateData({ secret });
+							return 'ok';
+						} finally {
+							jobSpan.end();
+						}
+					});
+				}, { connection, prefix });
+				worker.once('completed', () => resolve());
+				worker.once('failed', (_job, error) => reject(error));
+			});
+
+			await tracer.startActiveSpan('HTTP POST /telemetry-test', async httpSpan => {
+				httpSpanId = httpSpan.spanContext().spanId;
+				try {
+					// Queue#add uses BullMQ's producer-side ioredis client.
+					await queue.add('probe', { secret });
+				} finally {
+					httpSpan.end();
+				}
+			});
+			await processed;
+			await provider.forceFlush();
+
+			const redisSpans = exporter.getFinishedSpans().filter(span => span.attributes['db.system.name'] === 'redis');
+			expect(redisSpans.some(span => span.parentSpanContext?.spanId === httpSpanId)).toBe(true);
+			expect(redisSpans.some(span => span.parentSpanContext?.spanId === jobSpanId)).toBe(true);
+			for (const span of redisSpans) {
+				expect(span.attributes).not.toHaveProperty('db.statement');
+				expect(span.attributes).not.toHaveProperty('db.query.text');
+				expect(Object.values(span.attributes)).not.toContain(secret);
+			}
+		} finally {
+			await worker?.close();
+			await queue.obliterate({ force: true });
+			await queue.close();
+			uninstall();
+			await provider.shutdown();
+		}
+	}, 30000);
+});

packages/backend/test/unit/telemetry-database-instrumentation.ts

@@ -0,0 +1,108 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { describe, expect, test, vi } from 'vitest';
+import { installDatabaseInstrumentation, installInstrumentation } from '@/core/telemetry/database-instrumentation.js';
+
+describe('database-instrumentation', () => {
+	test('does not install PostgreSQL instrumentation when disabled', async () => {
+		const uninstall = await installDatabaseInstrumentation({} as any, {
+			capturePgSpans: false,
+			capturePgStatement: false,
+			capturePgConnectionSpans: false,
+		});
+
+		expect(uninstall).toBeTypeOf('function');
+		expect(() => uninstall()).not.toThrow();
+	});
+
+	test('registers pg instrumentation with the active provider', () => {
+		const provider = {};
+		const pg = { setTracerProvider: vi.fn(), enable: vi.fn(), disable: vi.fn() };
+		const config = vi.fn();
+		const PgInstrumentation = class {
+			public constructor(options: unknown) {
+				config(options);
+				return pg as any;
+			}
+		};
+
+		const uninstall = installInstrumentation(provider as any, {
+			PgInstrumentation: PgInstrumentation as any,
+		}, {
+			capturePgStatement: false,
+			capturePgConnectionSpans: false,
+		});
+
+		expect(pg.setTracerProvider).toHaveBeenCalledWith(provider);
+		expect(pg.enable).toHaveBeenCalledOnce();
+		expect(config).toHaveBeenCalledWith(expect.objectContaining({
+			enhancedDatabaseReporting: false,
+			requireParentSpan: true,
+			ignoreConnectSpans: true,
+			requestHook: expect.any(Function),
+		}));
+
+		const span = { setAttribute: vi.fn() };
+		(config.mock.calls[0][0] as { requestHook: (span: any) => void }).requestHook(span);
+		expect(span.setAttribute).toHaveBeenCalledWith('db.statement', '[REDACTED]');
+		expect(span.setAttribute).toHaveBeenCalledWith('db.query.text', '[REDACTED]');
+
+		uninstall();
+
+		expect(pg.disable).toHaveBeenCalledOnce();
+	});
+
+	test('keeps SQL statement attributes when explicitly enabled', () => {
+		const pg = { setTracerProvider: vi.fn(), enable: vi.fn(), disable: vi.fn() };
+		const config = vi.fn();
+
+		installInstrumentation({} as any, {
+			PgInstrumentation: class {
+				public constructor(options: unknown) {
+					config(options);
+					return pg as any;
+				}
+			} as any,
+		}, {
+			capturePgStatement: true,
+			capturePgConnectionSpans: false,
+		});
+
+		expect(config.mock.calls[0][0]).not.toHaveProperty('requestHook');
+	});
+
+	test('enables connection spans when explicitly configured', () => {
+		const pg = { setTracerProvider: vi.fn(), enable: vi.fn(), disable: vi.fn() };
+		const config = vi.fn();
+
+		installInstrumentation({} as any, {
+			PgInstrumentation: class {
+				public constructor(options: unknown) {
+					config(options);
+					return pg as any;
+				}
+			} as any,
+		}, {
+			capturePgStatement: false,
+			capturePgConnectionSpans: true,
+		});
+
+		expect(config).toHaveBeenCalledWith(expect.objectContaining({
+			ignoreConnectSpans: false,
+		}));
+	});
+
+	test('cleans up both instrumentations when initialization fails', () => {
+		const pg = { setTracerProvider: vi.fn(), enable: vi.fn(), disable: vi.fn() };
+		pg.enable.mockImplementation(() => { throw new Error('failed'); });
+
+		expect(() => installInstrumentation({} as any, {
+			PgInstrumentation: class { public constructor() { return pg as any; } } as any,
+		})).toThrow('failed');
+
+		expect(pg.disable).toHaveBeenCalledOnce();
+	});
+});

packages/backend/test/unit/telemetry-redis-instrumentation.ts

@@ -0,0 +1,148 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { describe, expect, test, vi } from 'vitest';
+import { SpanKind, SpanStatusCode } from '@opentelemetry/api';
+import { createRedisInstrumentation } from '@/core/telemetry/redis-instrumentation.js';
+
+describe('redis-instrumentation', () => {
+	test('creates and completes a span for an ioredis command', () => {
+		let subscribers: any;
+		const unsubscribe = vi.fn();
+		const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn(), setAttribute: vi.fn() };
+		const tracer = { startSpan: vi.fn(() => span) };
+		const uninstall = createRedisInstrumentation({
+			tracingChannel: () => ({ subscribe: (value) => { subscribers = value; }, unsubscribe }),
+			tracer: tracer as any,
+			getActiveSpan: () => ({}) as any,
+			spanKindClient: SpanKind.CLIENT,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+		}, { captureCommandSpans: true });
+		const command = { command: 'get', args: ['key'], database: 0, serverAddress: 'redis', serverPort: 6379 };
+
+		subscribers.start(command);
+		subscribers.asyncEnd(command);
+
+		expect(tracer.startSpan).toHaveBeenCalledWith('get', expect.objectContaining({
+			kind: SpanKind.CLIENT,
+			attributes: expect.objectContaining({
+				'db.system.name': 'redis',
+				'db.operation.name': 'get',
+				'server.address': 'redis',
+				'server.port': 6379,
+			}),
+		}));
+		expect(span.end).toHaveBeenCalledOnce();
+
+		uninstall();
+		expect(unsubscribe).toHaveBeenCalledWith(subscribers);
+	});
+
+	test('records rejected Redis commands as errors', () => {
+		let subscribers: any;
+		const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn(), setAttribute: vi.fn() };
+		createRedisInstrumentation({
+			tracingChannel: () => ({ subscribe: (value) => { subscribers = value; }, unsubscribe: vi.fn() }),
+			tracer: { startSpan: vi.fn(() => span) } as any,
+			getActiveSpan: () => ({}) as any,
+			spanKindClient: SpanKind.CLIENT,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+		}, { captureCommandSpans: true });
+		const command = { command: 'get', args: ['key'], database: 0, serverAddress: 'redis', serverPort: undefined };
+		const error = Object.assign(new Error('ERR Redis failed'), { code: 'ERR' });
+
+		subscribers.start(command);
+		Object.assign(command, { error });
+		subscribers.error(command);
+		subscribers.asyncEnd(command);
+
+		expect(span.recordException).toHaveBeenCalledWith(error);
+		expect(span.recordException).toHaveBeenCalledOnce();
+		expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR, message: 'ERR Redis failed' });
+		expect(span.setAttribute).toHaveBeenCalledWith('error.type', 'ERR');
+		expect(span.setAttribute).toHaveBeenCalledWith('db.response.status_code', 'ERR');
+		expect(span.end).toHaveBeenCalledOnce();
+	});
+
+	test('does not create a root span when no parent span is active', () => {
+		let subscribers: any;
+		const tracer = { startSpan: vi.fn() };
+		createRedisInstrumentation({
+			tracingChannel: () => ({ subscribe: (value) => { subscribers = value; }, unsubscribe: vi.fn() }),
+			tracer: tracer as any,
+			getActiveSpan: () => undefined,
+			spanKindClient: SpanKind.CLIENT,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+		}, { captureCommandSpans: true });
+
+		subscribers.start({ command: 'get', args: ['key'], database: 0, serverAddress: 'redis', serverPort: 6379 });
+
+		expect(tracer.startSpan).not.toHaveBeenCalled();
+	});
+
+	test('creates a root span when explicitly enabled', () => {
+		let subscribers: any;
+		const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn(), setAttribute: vi.fn() };
+		const tracer = { startSpan: vi.fn(() => span) };
+		createRedisInstrumentation({
+			tracingChannel: () => ({ subscribe: (value) => { subscribers = value; }, unsubscribe: vi.fn() }),
+			tracer: tracer as any,
+			getActiveSpan: () => undefined,
+			spanKindClient: SpanKind.CLIENT,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+		}, { captureCommandSpans: true, requireParentSpan: false });
+
+		const command = { command: 'get', args: ['key'], database: 0, serverAddress: 'redis', serverPort: 6379 };
+		subscribers.start(command);
+		subscribers.asyncEnd(command);
+
+		expect(tracer.startSpan).toHaveBeenCalledOnce();
+		expect(span.end).toHaveBeenCalledOnce();
+	});
+
+	test('records connection spans only when explicitly enabled', () => {
+		const subscribers = new Map<string, any>();
+		const unsubscribe = vi.fn();
+		const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn(), setAttribute: vi.fn() };
+		const tracer = { startSpan: vi.fn(() => span) };
+		const uninstall = createRedisInstrumentation({
+			tracingChannel: (name) => ({ subscribe: (value) => { subscribers.set(name, value); }, unsubscribe }),
+			tracer: tracer as any,
+			getActiveSpan: () => undefined,
+			spanKindClient: SpanKind.CLIENT,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+		}, { captureConnectionSpans: true });
+
+		const connection = { serverAddress: 'redis', serverPort: 6379 };
+		subscribers.get('ioredis:connect').start(connection);
+		subscribers.get('ioredis:connect').asyncEnd(connection);
+
+		expect(tracer.startSpan).toHaveBeenCalledWith('connect', expect.objectContaining({
+			kind: SpanKind.CLIENT,
+			attributes: expect.objectContaining({
+				'db.operation.name': 'connect',
+				'server.address': 'redis',
+				'server.port': 6379,
+			}),
+		}));
+		expect(span.end).toHaveBeenCalledOnce();
+
+		uninstall();
+		expect(unsubscribe).toHaveBeenCalledWith(subscribers.get('ioredis:connect'));
+	});
+
+	test('does not subscribe to Redis command diagnostics unless explicitly enabled', () => {
+		const tracingChannel = vi.fn();
+		createRedisInstrumentation({
+			tracingChannel,
+			tracer: { startSpan: vi.fn() } as any,
+			getActiveSpan: () => undefined,
+			spanKindClient: SpanKind.CLIENT,
+			spanStatusCodeError: SpanStatusCode.ERROR,
+		});
+
+		expect(tracingChannel).not.toHaveBeenCalled();
+	});
+});

pnpm-lock.yaml

@@ -6,6 +6,7 @@ settings:
 
 overrides:
   '@aiscript-dev/aiscript-languageserver': '-'
+  bullmq>ioredis: 5.11.1
   chokidar: 5.0.0
   lodash: 4.18.1
 
@@ -131,6 +132,9 @@ importers:
       '@opentelemetry/exporter-trace-otlp-proto':
         specifier: 0.214.0
         version: 0.214.0(@opentelemetry/api@1.9.1)
+      '@opentelemetry/instrumentation-pg':
+        specifier: 0.72.0
+        version: 0.72.0(@opentelemetry/api@1.9.1)
       '@opentelemetry/resources':
         specifier: 2.7.1
         version: 2.7.1(@opentelemetry/api@1.9.1)
@@ -2257,9 +2261,6 @@ packages:
   '@ioredis/commands@1.10.0':
     resolution: {integrity: sha512-UmeW7z4LfctwoQ5wkhVzgq8tXkreED2xZGpX+Bg+zA+WJFZCT6c062AfCK/Dfk81xZnnwdhJCUMkitihRaoC2Q==}
 
-  '@ioredis/commands@1.5.1':
-    resolution: {integrity: sha512-JH8ZL/ywcJyR9MmJ5BNqZllXNZQqQbnVZOqpPQqE1vHiFgAw4NHbvE0FOduNU8IX9babitBT46571OnPTT0Zcw==}
-
   '@isaacs/cliui@8.0.2':
     resolution: {integrity: sha512-O8jcjabXaleOG9DQ0+ARXWZBTfnP4WNAqzuiJK7ll44AmxGKv/J2M4TPjxjY3znBCfvBXFzucm1twdyFybFqEA==}
     engines: {node: '>=12'}
@@ -2574,6 +2575,10 @@ packages:
     resolution: {integrity: sha512-fmEWp5kXlGEc3i/lR698Hz41DfGyN4Tbe4g7L1AxSc7fF8Xeh/FQ9Quqpa9dVA413Q1Ad43QOLzU4JoXgbFPWw==}
     engines: {node: '>=8.0.0'}
 
+  '@opentelemetry/api-logs@0.220.0':
+    resolution: {integrity: sha512-CmVa4ImJ+ynfrPMNaAXHET6Bhb44SwzmfyVJFq9ni2jgXJR/l7C6gfVFddNmHP+ZOkP9cf4f9DBe68qVLTHc9w==}
+    engines: {node: '>=8.0.0'}
+
   '@opentelemetry/api@1.9.1':
     resolution: {integrity: sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==}
     engines: {node: '>=8.0.0'}
@@ -2608,6 +2613,12 @@ packages:
     peerDependencies:
       '@opentelemetry/api': ^1.3.0
 
+  '@opentelemetry/instrumentation-pg@0.72.0':
+    resolution: {integrity: sha512-p9xrFc/6R8t6Y293sTYLZ83LnzZo/qY0bBPA4xabdQt0Qjt8i1SlYFsIeGY2Jmf5WcESNUdjQB3NxWnt5Ox7zw==}
+    engines: {node: ^18.19.0 || >=20.6.0}
+    peerDependencies:
+      '@opentelemetry/api': ^1.3.0
+
   '@opentelemetry/instrumentation@0.214.0':
     resolution: {integrity: sha512-MHqEX5Dk59cqVah5LiARMACku7jXSVk9iVDWOea4x3cr7VfdByeDCURK6o1lntT1JS/Tsovw01UJrBhN3/uC5w==}
     engines: {node: ^18.19.0 || >=20.6.0}
@@ -2620,6 +2631,12 @@ packages:
     peerDependencies:
       '@opentelemetry/api': ^1.3.0
 
+  '@opentelemetry/instrumentation@0.220.0':
+    resolution: {integrity: sha512-xQx3E2WxP1mDvKzxLxX+CTCtNLa560YJZ3087qYHerl2YmiKpv7AH+dAy7vmx+eVrZ5BwhfWUAVoKOoxCNHcpw==}
+    engines: {node: ^18.19.0 || >=20.6.0}
+    peerDependencies:
+      '@opentelemetry/api': ^1.3.0
+
   '@opentelemetry/otlp-exporter-base@0.214.0':
     resolution: {integrity: sha512-u1Gdv0/E9wP+apqWf7Wv2npXmgJtxsW2XL0TEv9FZloTZRuMBKmu8cYVXwS4Hm3q/f/3FuCnPTgiwYvIqRSpRg==}
     engines: {node: ^18.19.0 || >=20.6.0}
@@ -2678,6 +2695,12 @@ packages:
     resolution: {integrity: sha512-cifvXDhcqMwwTlTK04GBNeIe7yyo28Mfby85QXFe1Yk8nmi36Ab/5UQwptOx84SsoGNRg+EVSjwzfSZMy6pmlw==}
     engines: {node: '>=14'}
 
+  '@opentelemetry/sql-common@0.42.0':
+    resolution: {integrity: sha512-nwUwUU+8O8a4bnLqk6CodWeegGMEANgC94KTAhXcpGWLrW/2/hek/0ajNbjXnSOoNuCX+nteUPs46HFHhou9Xw==}
+    engines: {node: ^18.19.0 || >=20.6.0}
+    peerDependencies:
+      '@opentelemetry/api': ^1.1.0
+
   '@oxc-parser/binding-android-arm-eabi@0.127.0':
     resolution: {integrity: sha512-0LC7ye4hvqbIKxAzThzvswgHLFu2AURKzYLeSVvLdu2TBOYWQDmHnTqPLeA597BcUCxiLqLsS4CJ5uoI5WYWCQ==}
     engines: {node: ^20.19.0 || >=22.12.0}
@@ -3994,6 +4017,12 @@ packages:
   '@types/offscreencanvas@2019.7.3':
     resolution: {integrity: sha512-ieXiYmgSRXUDeOntE1InxjWyvEelZGP63M+cGuquuRLuIKKT1osnkXjxev9B7d1nXSug5vpunx+gNlbVxMlC9A==}
 
+  '@types/pg-pool@2.0.7':
+    resolution: {integrity: sha512-U4CwmGVQcbEuqpyju8/ptOKg6gEC+Tqsvj2xS9o1g71bUh8twxnC6ZL5rZKCsGN0iyH0CwgUyc9VR5owNQF9Ng==}
+
+  '@types/pg@8.15.6':
+    resolution: {integrity: sha512-NoaMtzhxOrubeL/7UZuNTrejB4MPAJ0RpxZqXQf2qXuVlTPuG6Y8p4u9dKRaue4yjmC7ZhzVO2/Yyyn25znrPQ==}
+
   '@types/pg@8.20.0':
     resolution: {integrity: sha512-bEPFOaMAHTEP1EzpvHTbmwR8UsFyHSKsRisLIHVMXnpNefSbGA1bD6CVy+qKjGSqmZqNqBDV2azOBo8TgkcVow==}
 
@@ -4937,10 +4966,6 @@ packages:
     resolution: {integrity: sha512-rwHwUfXL40Chm1r08yrhU3qpUvdVlgkKNeyeGPOxnW8/SyVDvgRaed/Uz54AqWNaTCAThlj6QAs3TZcKI0xDEw==}
     engines: {node: '>=0.10.0'}
 
-  cluster-key-slot@1.1.2:
-    resolution: {integrity: sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==}
-    engines: {node: '>=0.10.0'}
-
   color-convert@2.0.1:
     resolution: {integrity: sha512-RRECPsj7iu/xb5oKYcsFHSppFNnsj/52OVTRKb4zP5onXwVF3zVmmToNcOfGC+CRDpfK/U584fMg38ZHCaElKQ==}
     engines: {node: '>=7.0.0'}
@@ -6087,10 +6112,6 @@ packages:
     resolution: {integrity: sha512-7m1vEcPCxXYI8HqnL8CKI6siDyD+eIWSwgB3DZA+ZTogxk9I4CDnj4wilt9x/+/QbHI4YG5YZNmC6458/e9Ktg==}
     deprecated: The Intersection Observer polyfill is no longer needed and can safely be removed. Intersection Observer has been Baseline since 2019.
 
-  ioredis@5.10.1:
-    resolution: {integrity: sha512-HuEDBTI70aYdx1v6U97SbNx9F1+svQKBDo30o0b9fw055LMepzpOOd0Ccg9Q6tbqmBSJaMuY0fB7yw9/vjBYCA==}
-    engines: {node: '>=12.22.0'}
-
   ioredis@5.11.1:
     resolution: {integrity: sha512-ehuGcf94bQXhfagULNXrJdfnWO38v070jxSx/qE87Kjzmu2fU7ro5EFAb+OPituLqgfyuQaym5DlrNydW2sJ9A==}
     engines: {node: '>=12.22.0'}
@@ -6581,12 +6602,6 @@ packages:
     resolution: {integrity: sha512-iPZK6eYjbxRu3uB4/WZ3EsEIMJFMqAoopl3R+zuq0UjcAm/MO6KCweDgPfP3elTztoKP3KtnVHxTn2NHBSDVUw==}
     engines: {node: '>=10'}
 
-  lodash.defaults@4.2.0:
-    resolution: {integrity: sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ==}
-
-  lodash.isarguments@3.1.0:
-    resolution: {integrity: sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg==}
-
   lodash.merge@4.6.2:
     resolution: {integrity: sha512-0KpjqXRVvrYyCsX1swR/XTK0va6VQkQM6MNo7PqW77ByjAhoARA8EfrP1N4+KlKj8YS0ZUCtRT/YUuhyYDujIQ==}
 
@@ -10300,8 +10315,6 @@ snapshots:
 
   '@ioredis/commands@1.10.0': {}
 
-  '@ioredis/commands@1.5.1': {}
-
   '@isaacs/cliui@8.0.2':
     dependencies:
       string-width: 5.1.2
@@ -10612,6 +10625,10 @@ snapshots:
     dependencies:
       '@opentelemetry/api': 1.9.1
 
+  '@opentelemetry/api-logs@0.220.0':
+    dependencies:
+      '@opentelemetry/api': 1.9.1
+
   '@opentelemetry/api@1.9.1': {}
 
   '@opentelemetry/context-async-hooks@2.7.1(@opentelemetry/api@1.9.1)':
@@ -10642,6 +10659,18 @@ snapshots:
       '@opentelemetry/resources': 2.6.1(@opentelemetry/api@1.9.1)
       '@opentelemetry/sdk-trace-base': 2.6.1(@opentelemetry/api@1.9.1)
 
+  '@opentelemetry/instrumentation-pg@0.72.0(@opentelemetry/api@1.9.1)':
+    dependencies:
+      '@opentelemetry/api': 1.9.1
+      '@opentelemetry/core': 2.7.1(@opentelemetry/api@1.9.1)
+      '@opentelemetry/instrumentation': 0.220.0(@opentelemetry/api@1.9.1)
+      '@opentelemetry/semantic-conventions': 1.40.0
+      '@opentelemetry/sql-common': 0.42.0(@opentelemetry/api@1.9.1)
+      '@types/pg': 8.15.6
+      '@types/pg-pool': 2.0.7
+    transitivePeerDependencies:
+      - supports-color
+
   '@opentelemetry/instrumentation@0.214.0(@opentelemetry/api@1.9.1)':
     dependencies:
       '@opentelemetry/api': 1.9.1
@@ -10660,6 +10689,15 @@ snapshots:
     transitivePeerDependencies:
       - supports-color
 
+  '@opentelemetry/instrumentation@0.220.0(@opentelemetry/api@1.9.1)':
+    dependencies:
+      '@opentelemetry/api': 1.9.1
+      '@opentelemetry/api-logs': 0.220.0
+      import-in-the-middle: 3.0.1
+      require-in-the-middle: 8.0.1
+    transitivePeerDependencies:
+      - supports-color
+
   '@opentelemetry/otlp-exporter-base@0.214.0(@opentelemetry/api@1.9.1)':
     dependencies:
       '@opentelemetry/api': 1.9.1
@@ -10726,6 +10764,11 @@ snapshots:
 
   '@opentelemetry/semantic-conventions@1.40.0': {}
 
+  '@opentelemetry/sql-common@0.42.0(@opentelemetry/api@1.9.1)':
+    dependencies:
+      '@opentelemetry/api': 1.9.1
+      '@opentelemetry/core': 2.7.1(@opentelemetry/api@1.9.1)
+
   '@oxc-parser/binding-android-arm-eabi@0.127.0':
     optional: true
 
@@ -11967,6 +12010,16 @@ snapshots:
 
   '@types/offscreencanvas@2019.7.3': {}
 
+  '@types/pg-pool@2.0.7':
+    dependencies:
+      '@types/pg': 8.20.0
+
+  '@types/pg@8.15.6':
+    dependencies:
+      '@types/node': 26.0.1
+      pg-protocol: 1.15.0
+      pg-types: 2.2.0
+
   '@types/pg@8.20.0':
     dependencies:
       '@types/node': 26.0.1
@@ -12837,7 +12890,7 @@ snapshots:
   bullmq@5.79.2:
     dependencies:
       cron-parser: 4.9.0
-      ioredis: 5.10.1
+      ioredis: 5.11.1
       msgpackr: 2.0.4
       node-abort-controller: 3.1.1
       semver: 7.8.5
@@ -13036,8 +13089,6 @@ snapshots:
 
   cluster-key-slot@1.1.1: {}
 
-  cluster-key-slot@1.1.2: {}
-
   color-convert@2.0.1:
     dependencies:
       color-name: 1.1.4
@@ -14399,20 +14450,6 @@ snapshots:
 
   intersection-observer@0.12.2: {}
 
-  ioredis@5.10.1:
-    dependencies:
-      '@ioredis/commands': 1.5.1
-      cluster-key-slot: 1.1.2
-      debug: 4.4.3
-      denque: 2.1.0
-      lodash.defaults: 4.2.0
-      lodash.isarguments: 3.1.0
-      redis-errors: 1.2.0
-      redis-parser: 3.0.0
-      standard-as-callback: 2.1.0
-    transitivePeerDependencies:
-      - supports-color
-
   ioredis@5.11.1:
     dependencies:
       '@ioredis/commands': 1.10.0
@@ -14869,10 +14906,6 @@ snapshots:
     dependencies:
       p-locate: 5.0.0
 
-  lodash.defaults@4.2.0: {}
-
-  lodash.isarguments@3.1.0: {}
-
   lodash.merge@4.6.2: {}
 
   lodash@4.18.1: {}

pnpm-workspace.yaml

@@ -57,6 +57,7 @@ minimumReleaseAgeExclude:
   - "@opentelemetry/core@2.8.0"
 overrides:
   '@aiscript-dev/aiscript-languageserver': '-'
+  'bullmq>ioredis': 5.11.1
   chokidar: 5.0.0
   lodash: 4.18.1
 engineStrict: true