diff --git a/.config/example.yml b/.config/example.yml index ce4e332c0c..78f14854e5 100644 --- a/.config/example.yml +++ b/.config/example.yml @@ -357,6 +357,10 @@ id: 'aidx' # #resourceAttributes: # # deployment.environment: 'production' # #propagateTraceToRemote: false +# # Queue worker spans are independent traces linked to their enqueuer by default. +# # Set to 'parent' to make them child spans instead. This can create very large +# # traces for high-fan-out queues such as federation delivery. +# #jobTraceContextMode: 'link' #sentryForFrontend: # vueIntegration: diff --git a/CHANGELOG.md b/CHANGELOG.md index 9be4160b5d..8492476c66 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,7 @@ - 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 リクエストを自動計装するように +- Enhance: OpenTelemetry 利用時に BullMQ ジョブを enqueue 元の trace と Span Link で関連付けられるように - Enhance: Sentry バックエンドの自動計装を `sentryForBackend.disabledIntegrations` で個別に無効化できるように - Enhance: Node.js 22.23.0以降、24.17.0以降、26.4.0以降をサポートするように - Enhance: Docker Image の Node.js を 26.4.0 に、Debian を trixie (v13) に更新 diff --git a/packages/backend/src/config.ts b/packages/backend/src/config.ts index 8ab82ef4c7..ff0df65f93 100644 --- a/packages/backend/src/config.ts +++ b/packages/backend/src/config.ts @@ -32,6 +32,7 @@ type OtelBackendConfig = { sampleRate?: number; resourceAttributes?: Record; propagateTraceToRemote?: boolean; + jobTraceContextMode?: 'link' | 'parent'; }; /** diff --git a/packages/backend/src/core/QueueModule.ts b/packages/backend/src/core/QueueModule.ts index ecd96261e0..ad25631152 100644 --- a/packages/backend/src/core/QueueModule.ts +++ b/packages/backend/src/core/QueueModule.ts @@ -9,6 +9,7 @@ import { DI } from '@/di-symbols.js'; import type { Config } from '@/config.js'; import { baseQueueOptions, QUEUE } from '@/queue/const.js'; import { allSettled } from '@/misc/promise-tracker.js'; +import { instrumentQueue } from '@/core/telemetry/queue-instrumentation.js'; import { DeliverJobData, EndedPollNotificationJobData, @@ -31,63 +32,72 @@ export type ObjectStorageQueue = Bull.Queue; export type UserWebhookDeliverQueue = Bull.Queue; export type SystemWebhookDeliverQueue = Bull.Queue; +function createQueue(queueName: string, config: Config): Bull.Queue { + const queue = new Bull.Queue(queueName, baseQueueOptions(config, queueName)); + // Queue のラップは、enqueue 時に OTel context をジョブデータへ埋め込むためのもの。 + // Sentry 単独ではジョブ間の context 伝播を使わないので、OTel 未設定時は元の Queue を返す。 + if (config.otelForBackend == null) return queue; + + return instrumentQueue(queue); +} + const $system: Provider = { provide: 'queue:system', - useFactory: (config: Config) => new Bull.Queue(QUEUE.SYSTEM, baseQueueOptions(config, QUEUE.SYSTEM)), + useFactory: (config: Config) => createQueue>(QUEUE.SYSTEM, config), inject: [DI.config], }; const $endedPollNotification: Provider = { provide: 'queue:endedPollNotification', - useFactory: (config: Config) => new Bull.Queue(QUEUE.ENDED_POLL_NOTIFICATION, baseQueueOptions(config, QUEUE.ENDED_POLL_NOTIFICATION)), + useFactory: (config: Config) => createQueue(QUEUE.ENDED_POLL_NOTIFICATION, config), inject: [DI.config], }; const $postScheduledNote: Provider = { provide: 'queue:postScheduledNote', - useFactory: (config: Config) => new Bull.Queue(QUEUE.POST_SCHEDULED_NOTE, baseQueueOptions(config, QUEUE.POST_SCHEDULED_NOTE)), + useFactory: (config: Config) => createQueue(QUEUE.POST_SCHEDULED_NOTE, config), inject: [DI.config], }; const $deliver: Provider = { provide: 'queue:deliver', - useFactory: (config: Config) => new Bull.Queue(QUEUE.DELIVER, baseQueueOptions(config, QUEUE.DELIVER)), + useFactory: (config: Config) => createQueue(QUEUE.DELIVER, config), inject: [DI.config], }; const $inbox: Provider = { provide: 'queue:inbox', - useFactory: (config: Config) => new Bull.Queue(QUEUE.INBOX, baseQueueOptions(config, QUEUE.INBOX)), + useFactory: (config: Config) => createQueue(QUEUE.INBOX, config), inject: [DI.config], }; const $db: Provider = { provide: 'queue:db', - useFactory: (config: Config) => new Bull.Queue(QUEUE.DB, baseQueueOptions(config, QUEUE.DB)), + useFactory: (config: Config) => createQueue>(QUEUE.DB, config), inject: [DI.config], }; const $relationship: Provider = { provide: 'queue:relationship', - useFactory: (config: Config) => new Bull.Queue(QUEUE.RELATIONSHIP, baseQueueOptions(config, QUEUE.RELATIONSHIP)), + useFactory: (config: Config) => createQueue(QUEUE.RELATIONSHIP, config), inject: [DI.config], }; const $objectStorage: Provider = { provide: 'queue:objectStorage', - useFactory: (config: Config) => new Bull.Queue(QUEUE.OBJECT_STORAGE, baseQueueOptions(config, QUEUE.OBJECT_STORAGE)), + useFactory: (config: Config) => createQueue>(QUEUE.OBJECT_STORAGE, config), inject: [DI.config], }; const $userWebhookDeliver: Provider = { provide: 'queue:userWebhookDeliver', - useFactory: (config: Config) => new Bull.Queue(QUEUE.USER_WEBHOOK_DELIVER, baseQueueOptions(config, QUEUE.USER_WEBHOOK_DELIVER)), + useFactory: (config: Config) => createQueue(QUEUE.USER_WEBHOOK_DELIVER, config), inject: [DI.config], }; const $systemWebhookDeliver: Provider = { provide: 'queue:systemWebhookDeliver', - useFactory: (config: Config) => new Bull.Queue(QUEUE.SYSTEM_WEBHOOK_DELIVER, baseQueueOptions(config, QUEUE.SYSTEM_WEBHOOK_DELIVER)), + useFactory: (config: Config) => createQueue(QUEUE.SYSTEM_WEBHOOK_DELIVER, config), inject: [DI.config], }; diff --git a/packages/backend/src/core/telemetry/TelemetryService.ts b/packages/backend/src/core/telemetry/TelemetryService.ts index e73beb4f8a..8a62a521df 100644 --- a/packages/backend/src/core/telemetry/TelemetryService.ts +++ b/packages/backend/src/core/telemetry/TelemetryService.ts @@ -5,7 +5,7 @@ import { Injectable } from '@nestjs/common'; import { bindThis } from '@/decorators.js'; -import { captureMessage, shutdownTelemetry, startSpan } from './telemetry-registry.js'; +import { captureMessage, shutdownTelemetry, startSpan, startSpanWithTraceContext } from './telemetry-registry.js'; import type { OnApplicationShutdown } from '@nestjs/common'; import type { TelemetryCaptureMessageOptions } from './adapters/TelemetryAdapter.js'; @@ -21,6 +21,13 @@ export class TelemetryService implements OnApplicationShutdown { return startSpan(name, fn); } + @bindThis + public startSpanWithTraceContext(name: string, jobData: object, fn: () => T): T { + // jobData に enqueue 元の context があれば worker span へ復元する。 + // context の無い既存ジョブは、通常の startSpan と同じ扱いになる。 + return startSpanWithTraceContext(name, jobData, fn); + } + @bindThis public async onApplicationShutdown(_signal?: string): Promise { await shutdownTelemetry(); diff --git a/packages/backend/src/core/telemetry/adapters/OpenTelemetryAdapter.ts b/packages/backend/src/core/telemetry/adapters/OpenTelemetryAdapter.ts index 4fc7f2ef4f..b8aa827588 100644 --- a/packages/backend/src/core/telemetry/adapters/OpenTelemetryAdapter.ts +++ b/packages/backend/src/core/telemetry/adapters/OpenTelemetryAdapter.ts @@ -8,10 +8,12 @@ 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 { 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'; import type { ParentBasedSampler, Sampler } from '@opentelemetry/sdk-trace-base'; import type { OtelBackendRuntimeConfig, TelemetryAdapter, TelemetryCaptureMessageOptions } from './TelemetryAdapter.js'; +import type { QueueTraceContextCarrier, QueueTraceContextDeps } from '../queue-trace-context.js'; const DEFAULT_SHUTDOWN_TIMEOUT = 5000; @@ -24,6 +26,7 @@ type OpenTelemetryAdapterDeps = { spanStatusCodeError: SpanStatusCode; shutdownTimeout: number; shutdownHttpClientInstrumentation?: () => void; + queueTraceContext?: QueueTraceContextDeps; }; type CreateSamplerDeps = { @@ -50,7 +53,7 @@ export class OpenTelemetryAdapter implements TelemetryAdapter { public static async create(config: OtelBackendRuntimeConfig): Promise { const [ - { diag, DiagLogLevel, SpanKind, SpanStatusCode, trace }, + { context, diag, DiagLogLevel, propagation, ROOT_CONTEXT, SpanKind, SpanStatusCode, trace }, { W3CTraceContextPropagator }, { OTLPTraceExporter }, { defaultResource, detectResources, envDetector, resourceFromAttributes }, @@ -115,6 +118,15 @@ export class OpenTelemetryAdapter implements TelemetryAdapter { spanKindClient: SpanKind.CLIENT, spanStatusCodeError: SpanStatusCode.ERROR, }), + queueTraceContext: { + tracer, + propagation, + trace, + getActiveContext: () => context.active(), + rootContext: ROOT_CONTEXT, + mode: getQueueTraceContextMode(config.jobTraceContextMode), + spanStatusCodeError: SpanStatusCode.ERROR, + }, }); } @@ -124,45 +136,34 @@ export class OpenTelemetryAdapter implements TelemetryAdapter { // 通知を握り潰さないよう、報告専用の短命spanを作ってそこに記録する。 const span = this.deps.getActiveSpan(); if (span != null) { - recordError(span, new Error(message), this.deps.spanStatusCodeError); + recordSpanError(span, new Error(message), this.deps.spanStatusCodeError); return; } this.deps.tracer.startActiveSpan('captureMessage', reportSpan => { - recordError(reportSpan, new Error(message), this.deps.spanStatusCodeError); + recordSpanError(reportSpan, new Error(message), this.deps.spanStatusCodeError); reportSpan.end(); }); } public startSpan(name: string, fn: () => T): T { // 既存のTelemetryAdapter契約に合わせ、同期/非同期どちらでも同じspan lifetimeを保証する。 - return this.deps.tracer.startActiveSpan(name, span => { - try { - const result = fn(); - if (isPromiseLike(result)) { - // Promiseの場合は完了/失敗までspanを開いたままにし、失敗時はerror statusへ変換する。 - return result.then( - value => { - span.end(); - return value; - }, - error => { - recordError(span, error, this.deps.spanStatusCodeError); - span.end(); - throw error; - }, - ) as T; - } + return this.deps.tracer.startActiveSpan(name, span => executeSpan(span, fn, this.deps.spanStatusCodeError)); + } - // 同期成功はここでspanを閉じる。例外はcatch側で記録してから再throwする。 - span.end(); - return result; - } catch (error) { - recordError(span, error, this.deps.spanStatusCodeError); - span.end(); - throw error; - } - }); + public injectTraceContext(carrier: QueueTraceContextCarrier): void { + const queueTraceContext = this.deps.queueTraceContext; + // Queue context 用の依存は任意なので、無い場合はジョブデータを変更しない。 + if (queueTraceContext == null) return; + injectActiveTraceContext(queueTraceContext, carrier); + } + + public startSpanWithTraceContext(name: string, jobData: object, fn: () => T): T { + const queueTraceContext = this.deps.queueTraceContext; + // Queue context 用の依存が無い場合は、従来の span 作成経路と同じ動作を保つ。 + if (queueTraceContext == null) return this.startSpan(name, fn); + + return startSpanWithQueueTraceContext(queueTraceContext, name, jobData, fn, () => this.startSpan(name, fn)); } public async shutdown(): Promise { @@ -211,20 +212,6 @@ export function createSampler(sampleRate: number, deps: CreateSamplerDeps): Pare }); } -function recordError(span: Span, error: unknown, spanStatusCodeError: SpanStatusCode): void { - // throw値がError以外でもOTel exporterへ渡せる例外表現に正規化する。 - const exception = error instanceof Error ? error : new Error(String(error)); - span.recordException(exception); - span.setStatus({ - code: spanStatusCodeError, - message: exception.message, - }); -} - -function isPromiseLike(value: T): value is T & PromiseLike> { - return value != null && typeof (value as { then?: unknown }).then === 'function'; -} - export function getMisskeyProcessRole(): string { // Trace backend上でserver/queue/workerを見分けられるよう、Misskey固有の役割をresourceに載せる。 if (envOption.disableClustering) { diff --git a/packages/backend/src/core/telemetry/adapters/SentryTelemetryAdapter.ts b/packages/backend/src/core/telemetry/adapters/SentryTelemetryAdapter.ts index 3b6ab6d9be..8b9674af6d 100644 --- a/packages/backend/src/core/telemetry/adapters/SentryTelemetryAdapter.ts +++ b/packages/backend/src/core/telemetry/adapters/SentryTelemetryAdapter.ts @@ -5,9 +5,11 @@ import Logger from '@/logger.js'; import { registerDiagLogger } from '@/core/telemetry/telemetry-diag.js'; +import { getQueueTraceContextMode, injectActiveTraceContext, startSpanWithQueueTraceContext } from '@/core/telemetry/queue-trace-context.js'; import type * as SentryNode from '@sentry/node'; import type { NodeOptions } from '@sentry/node'; import type { OtelBackendRuntimeConfig, SentryBackendConfig, TelemetryAdapter, TelemetryCaptureMessageOptions } from './TelemetryAdapter.js'; +import type { QueueTraceContextCarrier, QueueTraceContextDeps } from '../queue-trace-context.js'; // OpenTelemetryAdapterのDEFAULT_SHUTDOWN_TIMEOUTと揃え、Sentryのtransportが詰まってもプロセス終了を妨げないようにする。 const DEFAULT_SHUTDOWN_TIMEOUT = 5000; @@ -113,6 +115,7 @@ export function buildSentryOtlpInitOptions(options: BuildSentryOtlpInitOptions): export class SentryTelemetryAdapter implements TelemetryAdapter { private constructor( private readonly Sentry: typeof SentryNode, + private readonly queueTraceContext?: QueueTraceContextDeps, ) { } @@ -131,7 +134,7 @@ export class SentryTelemetryAdapter implements TelemetryAdapter { ): Promise { const Sentry = await import('@sentry/node'); const { nodeProfilingIntegration } = await import('@sentry/profiling-node'); - const { diag, DiagLogLevel } = await import('@opentelemetry/api'); + const { context, diag, DiagLogLevel, propagation, ROOT_CONTEXT, SpanStatusCode, trace } = await import('@opentelemetry/api'); const { BatchSpanProcessor } = await import('@opentelemetry/sdk-trace-base'); const { OTLPTraceExporter } = await import('@opentelemetry/exporter-trace-otlp-proto'); @@ -151,7 +154,17 @@ export class SentryTelemetryAdapter implements TelemetryAdapter { nodeProfilingIntegration, })); - return new SentryTelemetryAdapter(Sentry); + // Sentry が初期化した同じ OTel provider から tracer/context API を受け取り、 + // Queue を跨ぐ context 伝播も Sentry と OTLP の両方へ同一 span として出力する。 + return new SentryTelemetryAdapter(Sentry, { + tracer: trace.getTracer('misskey-backend'), + propagation, + trace, + getActiveContext: () => context.active(), + rootContext: ROOT_CONTEXT, + mode: getQueueTraceContextMode(otelConfig.jobTraceContextMode), + spanStatusCodeError: SpanStatusCode.ERROR, + }); } public captureMessage(message: string, opts: TelemetryCaptureMessageOptions): void { @@ -166,6 +179,19 @@ export class SentryTelemetryAdapter implements TelemetryAdapter { return this.Sentry.startSpan({ name }, fn); } + public injectTraceContext(carrier: QueueTraceContextCarrier): void { + // Sentry 単体構成では queueTraceContext を持たず、従来どおりジョブデータを変更しない。 + if (this.queueTraceContext == null) return; + injectActiveTraceContext(this.queueTraceContext, carrier); + } + + public startSpanWithTraceContext(name: string, jobData: object, fn: () => T): T { + // Sentry 単体構成では Sentry 既存の span 作成経路を使う。 + if (this.queueTraceContext == null) return this.startSpan(name, fn); + + return startSpanWithQueueTraceContext(this.queueTraceContext, name, jobData, fn, () => this.startSpan(name, fn)); + } + public async shutdown(): Promise { // timeout未指定だとtransportのflushが詰まった際にプロセス終了を妨げるため、上限時間を設ける。 await this.Sentry.close(DEFAULT_SHUTDOWN_TIMEOUT); diff --git a/packages/backend/src/core/telemetry/adapters/TelemetryAdapter.ts b/packages/backend/src/core/telemetry/adapters/TelemetryAdapter.ts index 2152e901db..f9564e7e7e 100644 --- a/packages/backend/src/core/telemetry/adapters/TelemetryAdapter.ts +++ b/packages/backend/src/core/telemetry/adapters/TelemetryAdapter.ts @@ -4,6 +4,7 @@ */ import type { Config } from '@/config.js'; +import type { QueueTraceContextCarrier } from '../queue-trace-context.js'; export type SentryBackendConfig = NonNullable; export type OtelBackendConfig = NonNullable; @@ -40,6 +41,18 @@ export interface TelemetryAdapter { */ startSpan(name: string, fn: () => T): T; + /** + * BullMQ のジョブデータへ保存する carrier に、active trace context を注入する。 + * OTel を使わない adapter は実装しない。 + */ + injectTraceContext?(carrier: QueueTraceContextCarrier): void; + + /** + * ジョブに保存された enqueue 元の context を、worker span の Link または parent として復元する。 + * context を持たないジョブの互換性は adapter 側で保つ。 + */ + startSpanWithTraceContext?(name: string, jobData: object, fn: () => T): T; + /** * プロセス終了時にtelemetry backendへ残りのデータをflushする。 * 実装側ではtransport停止に引きずられないよう、待機時間に上限を設ける。 diff --git a/packages/backend/src/core/telemetry/queue-instrumentation.ts b/packages/backend/src/core/telemetry/queue-instrumentation.ts new file mode 100644 index 0000000000..90eb710efa --- /dev/null +++ b/packages/backend/src/core/telemetry/queue-instrumentation.ts @@ -0,0 +1,33 @@ +/* + * SPDX-FileCopyrightText: syuilo and misskey-project + * SPDX-License-Identifier: AGPL-3.0-only + */ + +import { injectTraceContext } from './telemetry-registry.js'; +import { injectQueueTraceContext } from './queue-trace-context.js'; +import type * as Bull from 'bullmq'; + +/** + * Queue の add/addBulk をラップし、全ての BullMQ enqueue 経路を一箇所で捕捉する。 + * QueueService を通さず直接 add/addBulk する呼び出し元もあるため、それぞれで注入すると漏れやすい。 + */ +export function instrumentQueue(queue: Bull.Queue): Bull.Queue { + // BullMQ のメソッドは Queue インスタンスを this として使うため、差し替え前に bind して保持する。 + const add = queue.add.bind(queue); + queue.add = ((name, data, opts) => { + // BullMQ が data を Redis 用にシリアライズする前に、enqueue 元の context を内部フィールドへ追加する。 + injectQueueTraceContext(data, injectTraceContext); + return add(name, data, opts); + }) as typeof queue.add; + + // addBulk は複数ジョブを一度にシリアライズするので、各 data へ同じ context を注入する。 + const addBulk = queue.addBulk.bind(queue); + queue.addBulk = ((jobs) => { + for (const job of jobs) { + injectQueueTraceContext(job.data, injectTraceContext); + } + return addBulk(jobs); + }) as typeof queue.addBulk; + + return queue; +} diff --git a/packages/backend/src/core/telemetry/queue-trace-context.ts b/packages/backend/src/core/telemetry/queue-trace-context.ts new file mode 100644 index 0000000000..d9f7f7462d --- /dev/null +++ b/packages/backend/src/core/telemetry/queue-trace-context.ts @@ -0,0 +1,182 @@ +/* + * SPDX-FileCopyrightText: syuilo and misskey-project + * SPDX-License-Identifier: AGPL-3.0-only + */ + +import type { Context, PropagationAPI, Span, SpanContext, SpanOptions, SpanStatusCode, Tracer } from '@opentelemetry/api'; + +/* + * enqueue (push) から worker による取得・処理 (pop) までをテレメトリ上で関連付け、 + * Queue を挟む非同期処理を一連の流れとして追跡できるようにする。 + * そのため、trace context を次の流れで伝播する: + * + * 1. producer 側で active context を W3C Trace Context 形式の carrier へ注入する。 + * 2. carrier をジョブデータの内部フィールドに保存し、BullMQ/Redis 経由で worker へ渡す。 + * 3. worker 側で carrier を context へ戻し、設定に応じて Link または parent に使う。 + * + * OTel API への依存を引数にしているのは、OTel 単体構成と Sentry + OTLP 構成で + * 同じ伝播処理を使うため。 + */ +/** Redis 上のジョブデータにだけ保存する、OpenTelemetry propagator 用の carrier。 */ +export type QueueTraceContextCarrier = Record; +export type QueueTraceContextMode = 'link' | 'parent'; + +// 通常のジョブプロセッサが参照しない Misskey 内部フィールドとして、ユーザー定義の data と区別する。 +const QUEUE_TRACE_CONTEXT_KEY = '__misskeyTraceContext'; + +export type QueueTraceContextDeps = { + tracer: Pick; + propagation: Pick; + trace: { + getSpanContext(context: Context): SpanContext | undefined; + }; + getActiveContext: () => Context; + rootContext: Context; + mode: QueueTraceContextMode; + spanStatusCodeError: SpanStatusCode; +}; + +export type QueueSpanContext = { + options: SpanOptions; + parentContext: Context; +}; + +type QueueSpanContextDeps = Pick; + +/** + * enqueue 元の active context を、ジョブ本来のデータを壊さない内部フィールドとして保持する。 + * propagator が何も注入しなかった場合は、Redis に不要な空オブジェクトを残さない。 + */ +export function injectQueueTraceContext(data: unknown, inject: (carrier: QueueTraceContextCarrier) => void): void { + // BullMQ の型定義外から渡される値も考慮し、書き込めない値は無視する。 + if (data == null || typeof data !== 'object') return; + + const carrier: QueueTraceContextCarrier = {}; + inject(carrier); + + // active span が無い場合、propagator は何も注入しない。空の内部フィールドは Redis に保存しない。 + if (Object.keys(carrier).length === 0) return; + Object.assign(data, { [QUEUE_TRACE_CONTEXT_KEY]: carrier }); +} + +/** 現在実行中の span を propagator の標準形式で carrier へ書き出す。 */ +export function injectActiveTraceContext(deps: QueueTraceContextDeps, carrier: QueueTraceContextCarrier): void { + deps.propagation.inject(deps.getActiveContext(), carrier); +} + +/** + * ジョブに保存された carrier から、worker span の開始に必要な context と options を組み立てる。 + * + * - parent: enqueue span の子として同じ trace を継続する。 + * - link: worker span を別の root trace にし、enqueue span への関連だけを Link に残す。 + * + * link モードで trace を分けることで、worker 側の sampling 判定を enqueue 側から独立させられる。 + */ +export function getQueueSpanContext(data: unknown, deps: QueueSpanContextDeps): QueueSpanContext | undefined { + const carrier = getQueueTraceContextCarrier(data); + if (carrier == null) return undefined; + + // Redis から復元した carrier は別プロセス由来なので、現在の context ではなく ROOT_CONTEXT から展開する。 + const extractedContext = deps.propagation.extract(deps.rootContext, carrier); + if (deps.mode === 'parent') { + return { + options: {}, + parentContext: extractedContext, + }; + } + + // Link が受け取るのは Context ではなく SpanContext なので、extract 後に取り出す。 + const spanContext = deps.trace.getSpanContext(extractedContext); + return { + options: { + root: true, + ...(spanContext != null ? { links: [{ context: spanContext }] } : {}), + }, + parentContext: deps.rootContext, + }; +} + +/** + * context を持つジョブは Link/parent の規則で span を開始する。 + * デプロイ前に enqueue されたジョブなど、context を持たない場合は既存の adapter 固有実装へ委ねる。 + */ +export function startSpanWithQueueTraceContext( + deps: QueueTraceContextDeps, + name: string, + jobData: object, + fn: () => T, + fallback: () => T, +): T { + const spanContext = getQueueSpanContext(jobData, deps); + if (spanContext == null) return fallback(); + + return deps.tracer.startActiveSpan(name, spanContext.options, spanContext.parentContext, span => executeSpan(span, fn, deps.spanStatusCodeError)); +} + +/** + * 既存の TelemetryAdapter 契約に合わせ、同期値と Promise のどちらを返す処理も span で包む。 + * 成功・失敗のどちらでも処理完了まで span を開き、必ず一度だけ閉じる。 + */ +export function executeSpan(span: Span, fn: () => T, spanStatusCodeError: SpanStatusCode): T { + try { + const result = fn(); + if (isPromiseLike(result)) { + // fn() の戻り値は T のまま保ちつつ、Promise の settle 時に span を閉じる。 + return result.then( + value => { + span.end(); + return value; + }, + error => { + recordSpanError(span, error, spanStatusCodeError); + span.end(); + throw error; + }, + ) as T; + } + + span.end(); + return result; + } catch (error) { + // fn() が同期的に throw した場合も、Promise の reject と同じ形で記録して呼び出し元へ戻す。 + recordSpanError(span, error, spanStatusCodeError); + span.end(); + throw error; + } +} + +/** Error 以外の throw 値も OTel exporter が扱える例外に正規化し、span をエラー状態にする。 */ +export function recordSpanError(span: Span, error: unknown, spanStatusCodeError: SpanStatusCode): void { + const exception = error instanceof Error ? error : new Error(String(error)); + span.recordException(exception); + span.setStatus({ + code: spanStatusCodeError, + message: exception.message, + }); +} + +/** + * 未設定時は、enqueue と worker の sampling を分離できる link を使う。 + * 設定ミスを無言でフォールバックさせないため、未知の値は起動時にエラーにする。 + */ +export function getQueueTraceContextMode(mode: unknown): QueueTraceContextMode { + if (mode == null || mode === 'link') return 'link'; + if (mode === 'parent') return 'parent'; + throw new Error('otelForBackend.jobTraceContextMode must be either \'link\' or \'parent\'.'); +} + +function getQueueTraceContextCarrier(data: unknown): QueueTraceContextCarrier | undefined { + if (data == null || typeof data !== 'object') return undefined; + const carrier = (data as Record)[QUEUE_TRACE_CONTEXT_KEY]; + if (carrier == null || typeof carrier !== 'object' || Array.isArray(carrier)) return undefined; + + // Redis 上の旧いジョブや壊れたデータで worker を落とさないよう、propagator に渡せる string map だけを受け入れる。 + const entries = Object.entries(carrier); + if (entries.length === 0 || entries.some(([, value]) => typeof value !== 'string')) return undefined; + return Object.fromEntries(entries) as QueueTraceContextCarrier; +} + +function isPromiseLike(value: T): value is T & PromiseLike> { + // native Promise に限らず thenable も span の完了を待てるよう、instanceof ではなく then の有無で判定する。 + return value != null && typeof (value as { then?: unknown }).then === 'function'; +} diff --git a/packages/backend/src/core/telemetry/telemetry-registry.ts b/packages/backend/src/core/telemetry/telemetry-registry.ts index 840def3a14..b08301816b 100644 --- a/packages/backend/src/core/telemetry/telemetry-registry.ts +++ b/packages/backend/src/core/telemetry/telemetry-registry.ts @@ -7,6 +7,7 @@ import type { Config } from '@/config.js'; import { OpenTelemetryAdapter } from './adapters/OpenTelemetryAdapter.js'; import { SentryTelemetryAdapter } from './adapters/SentryTelemetryAdapter.js'; import type { OtelBackendRuntimeConfig, TelemetryAdapter, TelemetryCaptureMessageOptions } from './adapters/TelemetryAdapter.js'; +import type { QueueTraceContextCarrier } from './queue-trace-context.js'; /** * NestのDIコンテナが構築される前(boot処理内)で初期化する必要があるため、 @@ -54,6 +55,19 @@ export function startSpan(name: string, fn: () => T): T { return wrapped(); } +export function injectTraceContext(carrier: QueueTraceContextCarrier): void { + // Queue の carrier は共有データなので、通知と異なり全 adapter にブロードキャストしない。 + // OTel provider は現在 1 つだけなので、同じ header を上書きしないよう最初の対応 adapter だけを使う。 + adapters.find(adapter => adapter.injectTraceContext != null)?.injectTraceContext?.(carrier); +} + +export function startSpanWithTraceContext(name: string, jobData: object, fn: () => T): T { + // Queue context を解釈できる adapter に span 作成を任せる。 + // 対応 adapter が無い構成では通常の startSpan へフォールバックする。 + const adapter = adapters.find(adapter => adapter.startSpanWithTraceContext != null); + return adapter?.startSpanWithTraceContext?.(name, jobData, fn) ?? startSpan(name, fn); +} + export async function shutdownTelemetry(): Promise { // 終了時は登録済みadapterを並列にflush/shutdownする。 await Promise.allSettled(adapters.map(adapter => adapter.shutdown())); diff --git a/packages/backend/src/queue/QueueProcessorService.ts b/packages/backend/src/queue/QueueProcessorService.ts index cd1e752fd5..c20c39f15a 100644 --- a/packages/backend/src/queue/QueueProcessorService.ts +++ b/packages/backend/src/queue/QueueProcessorService.ts @@ -158,6 +158,8 @@ export class QueueProcessorService implements OnApplicationShutdown { }; } + // 以下の各 Worker は job.data に保存された enqueue 元の trace context を復元し、 + // ジョブの実処理全体を Link または parent の worker span で囲む。 //#region system { const processer = (job: Bull.Job) => { @@ -176,7 +178,7 @@ export class QueueProcessorService implements OnApplicationShutdown { }; this.systemQueueWorker = new Bull.Worker(QUEUE.SYSTEM, (job) => { - return this.telemetryService.startSpan('Queue: System: ' + job.name, () => processer(job)); + return this.telemetryService.startSpanWithTraceContext('Queue: System: ' + job.name, job.data, () => processer(job)); }, { ...baseWorkerOptions(this.config, QUEUE.SYSTEM), autorun: false, @@ -227,7 +229,7 @@ export class QueueProcessorService implements OnApplicationShutdown { }; this.dbQueueWorker = new Bull.Worker(QUEUE.DB, (job) => { - return this.telemetryService.startSpan('Queue: DB: ' + job.name, () => processer(job)); + return this.telemetryService.startSpanWithTraceContext('Queue: DB: ' + job.name, job.data, () => processer(job)); }, { ...baseWorkerOptions(this.config, QUEUE.DB), autorun: false, @@ -253,7 +255,7 @@ export class QueueProcessorService implements OnApplicationShutdown { //#region deliver { this.deliverQueueWorker = new Bull.Worker(QUEUE.DELIVER, (job) => { - return this.telemetryService.startSpan('Queue: Deliver', () => this.deliverProcessorService.process(job)); + return this.telemetryService.startSpanWithTraceContext('Queue: Deliver', job.data, () => this.deliverProcessorService.process(job)); }, { ...baseWorkerOptions(this.config, QUEUE.DELIVER), autorun: false, @@ -287,7 +289,7 @@ export class QueueProcessorService implements OnApplicationShutdown { //#region inbox { this.inboxQueueWorker = new Bull.Worker(QUEUE.INBOX, (job) => { - return this.telemetryService.startSpan('Queue: Inbox', () => this.inboxProcessorService.process(job)); + return this.telemetryService.startSpanWithTraceContext('Queue: Inbox', job.data, () => this.inboxProcessorService.process(job)); }, { ...baseWorkerOptions(this.config, QUEUE.INBOX), autorun: false, @@ -321,7 +323,7 @@ export class QueueProcessorService implements OnApplicationShutdown { //#region user-webhook deliver { this.userWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.USER_WEBHOOK_DELIVER, (job) => { - return this.telemetryService.startSpan('Queue: UserWebhookDeliver', () => this.userWebhookDeliverProcessorService.process(job)); + return this.telemetryService.startSpanWithTraceContext('Queue: UserWebhookDeliver', job.data, () => this.userWebhookDeliverProcessorService.process(job)); }, { ...baseWorkerOptions(this.config, QUEUE.USER_WEBHOOK_DELIVER), autorun: false, @@ -355,7 +357,7 @@ export class QueueProcessorService implements OnApplicationShutdown { //#region system-webhook deliver { this.systemWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.SYSTEM_WEBHOOK_DELIVER, (job) => { - return this.telemetryService.startSpan('Queue: SystemWebhookDeliver', () => this.systemWebhookDeliverProcessorService.process(job)); + return this.telemetryService.startSpanWithTraceContext('Queue: SystemWebhookDeliver', job.data, () => this.systemWebhookDeliverProcessorService.process(job)); }, { ...baseWorkerOptions(this.config, QUEUE.SYSTEM_WEBHOOK_DELIVER), autorun: false, @@ -399,7 +401,7 @@ export class QueueProcessorService implements OnApplicationShutdown { }; this.relationshipQueueWorker = new Bull.Worker(QUEUE.RELATIONSHIP, (job) => { - return this.telemetryService.startSpan('Queue: Relationship: ' + job.name, () => processer(job)); + return this.telemetryService.startSpanWithTraceContext('Queue: Relationship: ' + job.name, job.data, () => processer(job)); }, { ...baseWorkerOptions(this.config, QUEUE.RELATIONSHIP), autorun: false, @@ -438,7 +440,7 @@ export class QueueProcessorService implements OnApplicationShutdown { }; this.objectStorageQueueWorker = new Bull.Worker(QUEUE.OBJECT_STORAGE, (job) => { - return this.telemetryService.startSpan('Queue: ObjectStorage: ' + job.name, () => processer(job)); + return this.telemetryService.startSpanWithTraceContext('Queue: ObjectStorage: ' + job.name, job.data, () => processer(job)); }, { ...baseWorkerOptions(this.config, QUEUE.OBJECT_STORAGE), autorun: false, @@ -465,7 +467,7 @@ export class QueueProcessorService implements OnApplicationShutdown { //#region ended poll notification { this.endedPollNotificationQueueWorker = new Bull.Worker(QUEUE.ENDED_POLL_NOTIFICATION, (job) => { - return this.telemetryService.startSpan('Queue: EndedPollNotification', () => this.endedPollNotificationProcessorService.process(job)); + return this.telemetryService.startSpanWithTraceContext('Queue: EndedPollNotification', job.data, () => this.endedPollNotificationProcessorService.process(job)); }, { ...baseWorkerOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION), autorun: false, @@ -476,7 +478,7 @@ export class QueueProcessorService implements OnApplicationShutdown { //#region post scheduled note { this.postScheduledNoteQueueWorker = new Bull.Worker(QUEUE.POST_SCHEDULED_NOTE, async (job) => { - return this.telemetryService.startSpan('Queue: PostScheduledNote', () => this.postScheduledNoteProcessorService.process(job)); + return this.telemetryService.startSpanWithTraceContext('Queue: PostScheduledNote', job.data, () => this.postScheduledNoteProcessorService.process(job)); }, { ...baseWorkerOptions(this.config, QUEUE.POST_SCHEDULED_NOTE), autorun: false, diff --git a/packages/backend/test/unit/core/telemetry/adapters/OpenTelemetryAdapter.ts b/packages/backend/test/unit/core/telemetry/adapters/OpenTelemetryAdapter.ts index 12224bdb51..4b615179f4 100644 --- a/packages/backend/test/unit/core/telemetry/adapters/OpenTelemetryAdapter.ts +++ b/packages/backend/test/unit/core/telemetry/adapters/OpenTelemetryAdapter.ts @@ -8,6 +8,7 @@ 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'; const mocks = vi.hoisted(() => { @@ -97,6 +98,61 @@ describe('OpenTelemetryAdapter', () => { expect(span.end).toHaveBeenCalledTimes(1); }); + test('creates a root worker span linked to the enqueue span by default', () => { + const span = { + end: vi.fn(), + recordException: vi.fn(), + setStatus: vi.fn(), + }; + const rootContext = {} as Context; + const extractedContext = {} as Context; + const sourceSpanContext = { + traceId: '0123456789abcdef0123456789abcdef', + spanId: '0123456789abcdef', + traceFlags: 1, + isRemote: true, + } as SpanContext; + const propagation = { + isPropagationApi: true, + inject: vi.fn(), + extract(this: { isPropagationApi: boolean }) { + if (!this.isPropagationApi) throw new Error('lost propagation API receiver'); + return extractedContext; + }, + }; + const tracer = { + startActiveSpan: vi.fn((_name: string, _options: unknown, _context: unknown, fn: (spanArg: typeof span) => string) => fn(span)), + } as any; + const adapter = new OpenTelemetryAdapter({ + tracer, + provider: { shutdown: vi.fn() }, + getActiveSpan: () => undefined, + spanStatusCodeError: SpanStatusCode.ERROR, + shutdownTimeout: 10, + queueTraceContext: { + tracer, + propagation: propagation as any, + trace: { getSpanContext: () => sourceSpanContext }, + getActiveContext: () => rootContext, + rootContext, + mode: 'link', + spanStatusCodeError: SpanStatusCode.ERROR, + }, + }); + + expect(adapter.startSpanWithTraceContext('Queue: Deliver', { + __misskeyTraceContext: { + traceparent: '00-0123456789abcdef0123456789abcdef-0123456789abcdef-01', + }, + }, () => 'ok')).toBe('ok'); + + expect(tracer.startActiveSpan).toHaveBeenCalledWith('Queue: Deliver', { + root: true, + links: [{ context: sourceSpanContext }], + }, rootContext, expect.any(Function)); + expect(span.end).toHaveBeenCalledTimes(1); + }); + test('bridges captureMessage to the active span when one exists', () => { const activeSpan = { recordException: vi.fn(), diff --git a/packages/backend/test/unit/core/telemetry/adapters/SentryTelemetryAdapter.ts b/packages/backend/test/unit/core/telemetry/adapters/SentryTelemetryAdapter.ts index 00caaaa660..c304731987 100644 --- a/packages/backend/test/unit/core/telemetry/adapters/SentryTelemetryAdapter.ts +++ b/packages/backend/test/unit/core/telemetry/adapters/SentryTelemetryAdapter.ts @@ -208,8 +208,13 @@ describe('SentryTelemetryAdapter.createWithOtlpExport', () => { nodeProfilingIntegration, })); vi.doMock('@opentelemetry/api', () => ({ + context: { active: vi.fn() }, diag: { setLogger }, DiagLogLevel: { WARN: 50 }, + propagation: { inject: vi.fn(), extract: vi.fn() }, + ROOT_CONTEXT: {}, + SpanStatusCode: { ERROR: 2 }, + trace: { getTracer: vi.fn(), getSpanContext: vi.fn() }, })); vi.doMock('@opentelemetry/sdk-trace-base', () => ({ BatchSpanProcessor, diff --git a/packages/backend/test/unit/core/telemetry/queue-instrumentation.ts b/packages/backend/test/unit/core/telemetry/queue-instrumentation.ts new file mode 100644 index 0000000000..ac73df3c13 --- /dev/null +++ b/packages/backend/test/unit/core/telemetry/queue-instrumentation.ts @@ -0,0 +1,57 @@ +/* + * SPDX-FileCopyrightText: syuilo and misskey-project + * SPDX-License-Identifier: AGPL-3.0-only + */ + +import { beforeEach, describe, expect, test, vi } from 'vitest'; +import type * as Bull from 'bullmq'; +import { instrumentQueue } from '@/core/telemetry/queue-instrumentation.js'; + +const mocks = vi.hoisted(() => ({ + injectTraceContext: vi.fn((carrier: Record) => { + carrier['traceparent'] = '00-0123456789abcdef0123456789abcdef-0123456789abcdef-01'; + }), +})); + +vi.mock('@/core/telemetry/telemetry-registry.js', () => ({ + injectTraceContext: mocks.injectTraceContext, +})); + +describe('queue-instrumentation', () => { + beforeEach(() => { + mocks.injectTraceContext.mockClear(); + }); + + test('injects the active trace context for add()', () => { + const add = vi.fn(); + const queue = instrumentQueue({ add, addBulk: vi.fn() } as unknown as Bull.Queue<{ noteId: string }>); + const data = { noteId: '9d6b9a65-46c9-4e1b-a640-9589693893c9' }; + + queue.add('endedPollNotification', data); + + expect(mocks.injectTraceContext).toHaveBeenCalledTimes(1); + expect(data).toMatchObject({ + __misskeyTraceContext: { + traceparent: '00-0123456789abcdef0123456789abcdef-0123456789abcdef-01', + }, + }); + expect(add).toHaveBeenCalledWith('endedPollNotification', data, undefined); + }); + + test('injects every job passed to addBulk()', () => { + const addBulk = vi.fn(); + const queue = instrumentQueue({ add: vi.fn(), addBulk } as unknown as Bull.Queue<{ to: string }>); + const jobs = [ + { name: 'deliver', data: { to: 'https://remote.example/inbox' } }, + { name: 'deliver', data: { to: 'https://remote2.example/inbox' } }, + ]; + + queue.addBulk(jobs); + + expect(mocks.injectTraceContext).toHaveBeenCalledTimes(2); + expect(jobs).toEqual(expect.arrayContaining([ + expect.objectContaining({ data: expect.objectContaining({ __misskeyTraceContext: expect.any(Object) }) }), + ])); + expect(addBulk).toHaveBeenCalledWith(jobs); + }); +}); diff --git a/packages/backend/test/unit/core/telemetry/queue-trace-context.ts b/packages/backend/test/unit/core/telemetry/queue-trace-context.ts new file mode 100644 index 0000000000..3ecf4fe270 --- /dev/null +++ b/packages/backend/test/unit/core/telemetry/queue-trace-context.ts @@ -0,0 +1,132 @@ +/* + * SPDX-FileCopyrightText: syuilo and misskey-project + * SPDX-License-Identifier: AGPL-3.0-only + */ + +import { describe, expect, test, vi } from 'vitest'; +import type { Context, SpanContext } from '@opentelemetry/api'; +import { getQueueSpanContext, getQueueTraceContextMode, injectActiveTraceContext, injectQueueTraceContext } from '@/core/telemetry/queue-trace-context.js'; + +const rootContext = {} as Context; +const extractedContext = {} as Context; +const sourceSpanContext: SpanContext = { + traceId: '0123456789abcdef0123456789abcdef', + spanId: '0123456789abcdef', + traceFlags: 1, + isRemote: true, +}; + +function jobData() { + return { + name: 'deliver', + __misskeyTraceContext: { + traceparent: '00-0123456789abcdef0123456789abcdef-0123456789abcdef-01', + }, + }; +} + +describe('queue-trace-context', () => { + test('stores only a non-empty carrier in the job data', () => { + const data = { noteId: '9d6b9a65-46c9-4e1b-a640-9589693893c9' }; + + injectQueueTraceContext(data, carrier => { + carrier['traceparent'] = '00-0123456789abcdef0123456789abcdef-0123456789abcdef-01'; + }); + + expect(data).toMatchObject({ + __misskeyTraceContext: { + traceparent: '00-0123456789abcdef0123456789abcdef-0123456789abcdef-01', + }, + }); + }); + + test('does not store an empty carrier when no active trace exists', () => { + const data = { noteId: '9d6b9a65-46c9-4e1b-a640-9589693893c9' }; + + injectQueueTraceContext(data, () => {}); + + expect(data).not.toHaveProperty('__misskeyTraceContext'); + }); + + test('ignores non-object job data', () => { + const inject = vi.fn(); + + injectQueueTraceContext(null, inject); + injectQueueTraceContext('not a job object', inject); + + expect(inject).not.toHaveBeenCalled(); + }); + + test('injects the active context with the configured propagator', () => { + const activeContext = {} as Context; + const carrier = {}; + const inject = vi.fn(); + + injectActiveTraceContext({ + tracer: { startActiveSpan: vi.fn() } as any, + propagation: { inject, extract: vi.fn() } as any, + trace: { getSpanContext: vi.fn() }, + getActiveContext: () => activeContext, + rootContext, + mode: 'link', + spanStatusCodeError: 2 as any, + }, carrier); + + expect(inject).toHaveBeenCalledWith(activeContext, carrier); + }); + + test('starts a new root trace with a link by default', () => { + const extract = vi.fn(() => extractedContext); + const getSpanContext = vi.fn(() => sourceSpanContext); + + const result = getQueueSpanContext(jobData(), { + rootContext, + propagation: { inject: vi.fn(), extract }, + trace: { getSpanContext }, + mode: 'link', + }); + + expect(extract).toHaveBeenCalledWith(rootContext, jobData().__misskeyTraceContext); + expect(result).toEqual({ + options: { + root: true, + links: [{ context: sourceSpanContext }], + }, + parentContext: rootContext, + }); + }); + + test('uses the extracted context as the parent when parent mode is selected', () => { + const result = getQueueSpanContext(jobData(), { + rootContext, + propagation: { inject: vi.fn(), extract: () => extractedContext }, + trace: { getSpanContext: () => sourceSpanContext }, + mode: 'parent', + }); + + expect(result).toEqual({ + options: {}, + parentContext: extractedContext, + }); + }); + + test('ignores malformed or missing carriers', () => { + const extract = vi.fn(() => extractedContext); + const deps = { + rootContext, + propagation: { inject: vi.fn(), extract }, + trace: { getSpanContext: () => sourceSpanContext }, + mode: 'link' as const, + }; + + expect(getQueueSpanContext({}, deps)).toBeUndefined(); + expect(getQueueSpanContext({ __misskeyTraceContext: { traceparent: 1 } }, deps)).toBeUndefined(); + expect(extract).not.toHaveBeenCalled(); + }); + + test('defaults to link mode and rejects invalid configuration', () => { + expect(getQueueTraceContextMode(undefined)).toBe('link'); + expect(getQueueTraceContextMode('parent')).toBe('parent'); + expect(() => getQueueTraceContextMode('children')).toThrow('otelForBackend.jobTraceContextMode'); + }); +});