diff --git a/packages/backend/src/core/telemetry/adapters/OpenTelemetryAdapter.ts b/packages/backend/src/core/telemetry/adapters/OpenTelemetryAdapter.ts index c3d5064a6b..0cf292314b 100644 --- a/packages/backend/src/core/telemetry/adapters/OpenTelemetryAdapter.ts +++ b/packages/backend/src/core/telemetry/adapters/OpenTelemetryAdapter.ts @@ -50,6 +50,8 @@ type CreateResourceDeps = { serviceVersion: string; }; +type InstrumentationInstaller = () => (() => void) | Promise<() => void>; + export class OpenTelemetryAdapter implements TelemetryAdapter { public constructor( private readonly deps: OpenTelemetryAdapterDeps, @@ -111,39 +113,46 @@ export class OpenTelemetryAdapter implements TelemetryAdapter { // provider操作をdepsに閉じ込め、span wrapper本体をユニットテストしやすくする。 const tracer = provider.getTracer('misskey-backend'); - - return new OpenTelemetryAdapter({ - tracer, - provider, - getActiveSpan: () => trace.getActiveSpan(), - spanStatusCodeError: SpanStatusCode.ERROR, - shutdownTimeout: DEFAULT_SHUTDOWN_TIMEOUT, - shutdownHttpClientInstrumentation: installHttpClientInstrumentation({ + return installInstrumentationsWithCleanup([ + () => installHttpClientInstrumentation({ tracer, spanKindClient: SpanKind.CLIENT, spanStatusCodeError: SpanStatusCode.ERROR, }), // pg のrequire hookとioredis diagnostics channelは、Nest moduleの動的importより前に有効化する。 - shutdownDatabaseInstrumentation: await installDatabaseInstrumentation(provider, { + () => installDatabaseInstrumentation(provider, { capturePgSpans: config.capturePgSpans === true, capturePgStatement: config.capturePgStatement === true, capturePgConnectionSpans: config.capturePgConnectionSpans === true, }), - shutdownRedisInstrumentation: installRedisInstrumentation(tracer, SpanKind.CLIENT, SpanStatusCode.ERROR, { + () => installRedisInstrumentation(tracer, SpanKind.CLIENT, SpanStatusCode.ERROR, { captureConnectionSpans: config.captureRedisConnectionSpans === true, captureCommandSpans: config.captureRedisCommandSpans === true, requireParentSpan: config.captureRedisRootSpans !== true, }), - queueTraceContext: { + ], ([ + shutdownHttpClientInstrumentation, + shutdownDatabaseInstrumentation, + shutdownRedisInstrumentation, + ]) => new OpenTelemetryAdapter({ tracer, - propagation, - trace, - getActiveContext: () => context.active(), - rootContext: ROOT_CONTEXT, - mode: getQueueTraceContextMode(config.jobTraceContextMode), + provider, + getActiveSpan: () => trace.getActiveSpan(), spanStatusCodeError: SpanStatusCode.ERROR, - }, - }); + shutdownTimeout: DEFAULT_SHUTDOWN_TIMEOUT, + shutdownHttpClientInstrumentation, + shutdownDatabaseInstrumentation, + shutdownRedisInstrumentation, + queueTraceContext: { + tracer, + propagation, + trace, + getActiveContext: () => context.active(), + rootContext: ROOT_CONTEXT, + mode: getQueueTraceContextMode(config.jobTraceContextMode), + spanStatusCodeError: SpanStatusCode.ERROR, + }, + })); } public captureMessage(message: string, _opts: TelemetryCaptureMessageOptions): void { @@ -192,9 +201,11 @@ export class OpenTelemetryAdapter implements TelemetryAdapter { } public async shutdown(): Promise { - this.deps.shutdownHttpClientInstrumentation?.(); - this.deps.shutdownDatabaseInstrumentation?.(); - this.deps.shutdownRedisInstrumentation?.(); + shutdownInstrumentations([ + this.deps.shutdownHttpClientInstrumentation, + this.deps.shutdownDatabaseInstrumentation, + this.deps.shutdownRedisInstrumentation, + ]); // BatchSpanProcessorのflushが詰まってもプロセス終了を妨げないよう、上限時間を設ける。 // タイムアウト側のtimerは、flushが先に終わった場合にイベントループを無駄に引き留めないようclearする。 let timer: NodeJS.Timeout | undefined; @@ -209,6 +220,32 @@ export class OpenTelemetryAdapter implements TelemetryAdapter { } } +export async function installInstrumentationsWithCleanup( + installers: readonly InstrumentationInstaller[], + create: (shutdowns: ReadonlyArray<() => void>) => T, +): Promise { + const shutdowns: Array<() => void> = []; + try { + for (const install of installers) { + shutdowns.push(await install()); + } + return create(shutdowns); + } catch (error) { + shutdownInstrumentations(shutdowns.reverse()); + throw error; + } +} + +function shutdownInstrumentations(shutdowns: ReadonlyArray<(() => void) | undefined>): void { + for (const shutdown of shutdowns) { + try { + shutdown?.(); + } catch { + // instrumentation の停止失敗で、残りの停止処理や provider の flush を妨げない。 + } + } +} + export function createResource(config: OtelBackendRuntimeConfig, deps: CreateResourceDeps): Resource { // resourceを明示指定するとSDKのdefaultResource()は自動付与されなくなる(マージではなく上書き)ため、 // telemetry.sdk.*等の標準属性を失わないよう明示的にmergeする。 diff --git a/packages/backend/test/unit/core/telemetry/adapters/OpenTelemetryAdapter.ts b/packages/backend/test/unit/core/telemetry/adapters/OpenTelemetryAdapter.ts index ad6a559250..545686a24c 100644 --- a/packages/backend/test/unit/core/telemetry/adapters/OpenTelemetryAdapter.ts +++ b/packages/backend/test/unit/core/telemetry/adapters/OpenTelemetryAdapter.ts @@ -3,13 +3,13 @@ * SPDX-License-Identifier: AGPL-3.0-only */ -import { beforeEach, describe, expect, test, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; import { SpanStatusCode } from '@opentelemetry/api'; import { defaultResource, detectResources, envDetector, resourceFromAttributes } from '@opentelemetry/resources'; import { ParentBasedSampler, TraceIdRatioBasedSampler } from '@opentelemetry/sdk-trace-base'; import { ATTR_SERVICE_INSTANCE_ID, ATTR_SERVICE_NAME, ATTR_SERVICE_VERSION } from '@opentelemetry/semantic-conventions'; import type { Context, SpanContext } from '@opentelemetry/api'; -import { OpenTelemetryAdapter, createResource, createSampler, getMisskeyProcessRole } from '@/core/telemetry/adapters/OpenTelemetryAdapter.js'; +import { OpenTelemetryAdapter, createResource, createSampler, getMisskeyProcessRole, installInstrumentationsWithCleanup } from '@/core/telemetry/adapters/OpenTelemetryAdapter.js'; const mocks = vi.hoisted(() => { return { @@ -40,6 +40,10 @@ const samplerDeps = { }; describe('OpenTelemetryAdapter', () => { + afterEach(() => { + vi.useRealTimers(); + }); + test('wraps async work in an active span and ends it after success', async () => { const span = { end: vi.fn(), @@ -216,7 +220,6 @@ describe('OpenTelemetryAdapter', () => { await vi.advanceTimersByTimeAsync(50); await expect(shutdown).resolves.toBeUndefined(); - vi.useRealTimers(); }); test('clears the shutdown timeout timer once provider.shutdown() resolves first', async () => { @@ -234,7 +237,30 @@ describe('OpenTelemetryAdapter', () => { expect(clearTimeoutSpy).toHaveBeenCalled(); clearTimeoutSpy.mockRestore(); - vi.useRealTimers(); + }); + + test('continues shutdown and flushes the provider when instrumentation cleanup fails', async () => { + const shutdownHttpClientInstrumentation = vi.fn(() => { throw new Error('failed'); }); + const shutdownDatabaseInstrumentation = vi.fn(); + const shutdownRedisInstrumentation = vi.fn(); + const provider = { shutdown: vi.fn().mockResolvedValue(undefined) }; + const adapter = new OpenTelemetryAdapter({ + tracer: { startActiveSpan: vi.fn() }, + provider, + getActiveSpan: () => undefined, + spanStatusCodeError: SpanStatusCode.ERROR, + shutdownTimeout: 10, + shutdownHttpClientInstrumentation, + shutdownDatabaseInstrumentation, + shutdownRedisInstrumentation, + }); + + await expect(adapter.shutdown()).resolves.toBeUndefined(); + + expect(shutdownHttpClientInstrumentation).toHaveBeenCalledOnce(); + expect(shutdownDatabaseInstrumentation).toHaveBeenCalledOnce(); + expect(shutdownRedisInstrumentation).toHaveBeenCalledOnce(); + expect(provider.shutdown).toHaveBeenCalledOnce(); }); test('captureMessage starts a standalone span to report the error when there is no active span', () => { @@ -271,6 +297,38 @@ describe('OpenTelemetryAdapter', () => { }); }); +describe('installInstrumentationsWithCleanup', () => { + test('cleans up acquired instrumentations in reverse order and preserves the installation error', async () => { + const installationError = new Error('failed to install'); + const shutdownHttpClientInstrumentation = vi.fn(); + const shutdownDatabaseInstrumentation = vi.fn(() => { throw new Error('failed to clean up'); }); + + await expect(installInstrumentationsWithCleanup([ + () => shutdownHttpClientInstrumentation, + () => shutdownDatabaseInstrumentation, + () => { throw installationError; }, + ], () => { throw new Error('should not create'); })).rejects.toBe(installationError); + + expect(shutdownDatabaseInstrumentation).toHaveBeenCalledOnce(); + expect(shutdownHttpClientInstrumentation).toHaveBeenCalledOnce(); + expect(shutdownDatabaseInstrumentation.mock.invocationCallOrder[0]).toBeLessThan(shutdownHttpClientInstrumentation.mock.invocationCallOrder[0]); + }); + + test('cleans up all instrumentations when adapter construction fails', async () => { + const constructionError = new Error('failed to create adapter'); + const shutdownHttpClientInstrumentation = vi.fn(); + const shutdownDatabaseInstrumentation = vi.fn(); + + await expect(installInstrumentationsWithCleanup([ + () => shutdownHttpClientInstrumentation, + () => shutdownDatabaseInstrumentation, + ], () => { throw constructionError; })).rejects.toBe(constructionError); + + expect(shutdownDatabaseInstrumentation).toHaveBeenCalledOnce(); + expect(shutdownHttpClientInstrumentation).toHaveBeenCalledOnce(); + }); +}); + describe('createSampler', () => { test('accepts sample rates within [0, 1]', () => { expect(() => createSampler(0, samplerDeps)).not.toThrow(); diff --git a/packages/backend/test/unit/core/telemetry/queue-trace-context.ts b/packages/backend/test/unit/core/telemetry/queue-trace-context.ts index 680cafc87d..b128a89e06 100644 --- a/packages/backend/test/unit/core/telemetry/queue-trace-context.ts +++ b/packages/backend/test/unit/core/telemetry/queue-trace-context.ts @@ -4,8 +4,9 @@ */ import { describe, expect, test, vi } from 'vitest'; +import { SpanStatusCode } from '@opentelemetry/api'; import type { Context, SpanContext } from '@opentelemetry/api'; -import { getQueueSpanContext, getQueueTraceContextMode, injectActiveTraceContext, injectQueueTraceContext } from '@/core/telemetry/queue-trace-context.js'; +import { executeSpan, getQueueSpanContext, getQueueTraceContextMode, injectActiveTraceContext, injectQueueTraceContext, recordSpanError, startSpanWithQueueTraceContext } from '@/core/telemetry/queue-trace-context.js'; const rootContext = { kind: 'root' } as unknown as Context; const extractedContext = { kind: 'extracted' } as unknown as Context; @@ -132,4 +133,88 @@ describe('queue-trace-context', () => { expect(getQueueTraceContextMode('parent')).toBe('parent'); expect(() => getQueueTraceContextMode('children')).toThrow('otelForBackend.jobTraceContextMode'); }); + + test('executes synchronous work and ends the span once', () => { + const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn() }; + + expect(executeSpan(span as any, () => 'ok', SpanStatusCode.ERROR)).toBe('ok'); + expect(span.end).toHaveBeenCalledOnce(); + expect(span.recordException).not.toHaveBeenCalled(); + }); + + test('waits for resolved Promise work before ending the span', async () => { + const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn() }; + + await expect(executeSpan(span as any, () => Promise.resolve('ok'), SpanStatusCode.ERROR)).resolves.toBe('ok'); + expect(span.end).toHaveBeenCalledOnce(); + expect(span.recordException).not.toHaveBeenCalled(); + }); + + test('records and propagates synchronous failures', () => { + const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn() }; + const error = new Error('boom'); + + expect(() => executeSpan(span as any, () => { throw error; }, SpanStatusCode.ERROR)).toThrow(error); + expect(span.recordException).toHaveBeenCalledWith(error); + expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR, message: error.message }); + expect(span.end).toHaveBeenCalledOnce(); + }); + + test('records and propagates rejected Promise work', async () => { + const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn() }; + const error = new Error('boom'); + + await expect(executeSpan(span as any, () => Promise.reject(error), SpanStatusCode.ERROR)).rejects.toBe(error); + expect(span.recordException).toHaveBeenCalledWith(error); + expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR, message: error.message }); + expect(span.end).toHaveBeenCalledOnce(); + }); + + test('normalizes non-Error failures before recording them', () => { + const span = { recordException: vi.fn(), setStatus: vi.fn() }; + + recordSpanError(span as any, 'boom', SpanStatusCode.ERROR); + + expect(span.recordException).toHaveBeenCalledWith(expect.objectContaining({ message: 'boom' })); + expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR, message: 'boom' }); + }); + + test('starts a span with the extracted queue context', () => { + const span = { end: vi.fn(), recordException: vi.fn(), setStatus: vi.fn() }; + const startActiveSpan = vi.fn((_name: string, _options: unknown, _context: unknown, fn: (spanArg: typeof span) => string) => fn(span)); + const deps = { + tracer: { startActiveSpan } as any, + rootContext, + propagation: { inject: vi.fn(), extract: () => extractedContext } as any, + trace: { getSpanContext: () => sourceSpanContext }, + getActiveContext: () => rootContext, + mode: 'link' as const, + spanStatusCodeError: SpanStatusCode.ERROR, + }; + + expect(startSpanWithQueueTraceContext(deps, 'Queue: Deliver', jobData(), () => 'ok', () => 'fallback')).toBe('ok'); + expect(startActiveSpan).toHaveBeenCalledWith('Queue: Deliver', { + root: true, + links: [{ context: sourceSpanContext }], + }, rootContext, expect.any(Function)); + expect(span.end).toHaveBeenCalledOnce(); + }); + + test('uses the fallback without starting a span when queue context is missing', () => { + const startActiveSpan = vi.fn(); + const fallback = vi.fn(() => 'fallback'); + const deps = { + tracer: { startActiveSpan } as any, + rootContext, + propagation: { inject: vi.fn(), extract: vi.fn() } as any, + trace: { getSpanContext: vi.fn() }, + getActiveContext: () => rootContext, + mode: 'link' as const, + spanStatusCodeError: SpanStatusCode.ERROR, + }; + + expect(startSpanWithQueueTraceContext(deps, 'Queue: Deliver', {}, () => 'ok', fallback)).toBe('fallback'); + expect(fallback).toHaveBeenCalledOnce(); + expect(startActiveSpan).not.toHaveBeenCalled(); + }); }); diff --git a/packages/backend/test/unit/telemetry-database-instrumentation.ts b/packages/backend/test/unit/telemetry-database-instrumentation.ts index 5ac31fd0e6..20b3789a26 100644 --- a/packages/backend/test/unit/telemetry-database-instrumentation.ts +++ b/packages/backend/test/unit/telemetry-database-instrumentation.ts @@ -95,7 +95,7 @@ describe('database-instrumentation', () => { })); }); - test('cleans up both instrumentations when initialization fails', () => { + test('cleans up the pg instrumentation when initialization fails', () => { const pg = { setTracerProvider: vi.fn(), enable: vi.fn(), disable: vi.fn() }; pg.enable.mockImplementation(() => { throw new Error('failed'); });