Context propagation — AsyncLocalStorage, middleware и BullMQ в NestJS

Опирается на правила: R-OBS-CTX-1 … R-OBS-CTX-4 и R-OBS-CTX-X1 … R-OBS-CTX-X3 из Observability Style Guide → раздел 6. Context propagation.

Важно знать

• nestjs-pino создаёт per-request logger через AsyncLocalStorage автоматически — requestId из X-Request-Id header или UUID в каждой записи без передачи через аргументы.
• AsyncLocalStorage нативно проходит через await, Promise-цепочки и таймеры — аналог MDC без явного проброса.
• trace_id/span_id автоматически через @opentelemetry/instrumentation-pino. Не добавлять руками через logger.assign.
• userId populates в guard или interceptor после JWT-валидации, не в middleware до аутентификации.
• worker_threads и BullMQ разрывают AsyncLocalStorage — передавать requestId и traceparent явно в payload джобы и восстанавливать в processor-е.
• Обогащение контекста (logger.assign, cls.set) — только в middleware/guard/interceptor, не в handler или service.
• Утечка userId соседнего запроса в логи — compliance-инцидент; возникает при общем мутируемом сторе вместо per-request ALS.

В Java Spring MDC — thread-local, управляемый явным MDC.clear() в finally. В NestJS/Node нет thread pool — event loop однопоточен, а async-операции обслуживает та же нить через микротаски. AsyncLocalStorage (ALS) из async_hooks работает как контекстный ключ для каждого «async-дерева»: значение, заложенное в als.run(store, fn), видно всем await-ям внутри fn и их потомкам — без явного передавания аргументом.

genReqId и per-request logger

R-OBS-CTX-1: nestjs-pino настраивается через LoggerModule.forRoot с genReqId — каждый incoming HTTP request получает собственный ALS-контекст с requestId.

// src/app.module.ts
import { LoggerModule } from 'nestjs-pino';
import { randomUUID } from 'node:crypto';

@Module({
  imports: [
    LoggerModule.forRoot({
      pinoHttp: {
        genReqId: (req) => req.headers['x-request-id'] ?? randomUUID(),
        customSuccessMessage: () => 'request completed',
        customErrorMessage: (_req, res) => `request failed with status ${res.statusCode}`,
        redact: ['req.headers.authorization', '*.password', '*.email', '*.phone'],
        transport: process.env.NODE_ENV !== 'production'
          ? { target: 'pino-pretty', options: { colorize: true } }
          : undefined,
      },
    }),
  ],
})
export class AppModule {}

genReqId читает X-Request-Id из входящего header, а если его нет — генерирует UUID. nestjs-pino возвращает requestId в response (X-Request-Id), клиент может предоставить его при разборе инцидента.

PinoLogger инжектируется через DI:

// src/order/order.service.ts
import { PinoLogger, InjectPinoLogger } from 'nestjs-pino';

@Injectable()
export class OrderService {
  constructor(
    @InjectPinoLogger(OrderService.name)
    private readonly logger: PinoLogger,
  ) {}

  async confirmOrder(orderId: string, customerId: string): Promise<void> {
    this.logger.info({ orderId, customerId }, 'confirming order');
    // requestId и trace_id уже в ALS — pino возьмёт их автоматически
  }
}

console.log и new Logger() из @nestjs/common — не используются (R-OBS-LOG-X2): они мимо ALS-пайплайна и не содержат requestId.

trace_id/span_id автоматически

R-OBS-CTX-2: @opentelemetry/instrumentation-pino подключается в tracing.ts до NestFactory.create и автоматически добавляет trace_id/span_id из активного OTel-span в каждую pino-запись.

// src/tracing.ts
import { NodeSDK } from '@opentelemetry/sdk-node';
import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';

const sdk = new NodeSDK({
  serviceName: process.env.OTEL_SERVICE_NAME ?? 'order-service',
  traceExporter: new OTLPTraceExporter(),
  instrumentations: [getNodeAutoInstrumentations()],  // включает instrumentation-pino
});

sdk.start();

// src/main.ts
import './tracing';  // первым — до всего остального
import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';

Если tracing.ts импортируется после NestFactory — модули уже загружены без патча, pino-интеграция не работает. trace_id добавлять через logger.assign({ traceId: ... }) вручную — нарушение R-OBS-CTX-2; при смене активного span поле устаревает.

userId в guard после JWT

R-OBS-CTX-4: userId populates после аутентификации, не до неё.

// src/common/guards/jwt-auth.guard.ts
import { Injectable, CanActivate, ExecutionContext } from '@nestjs/common';
import { PinoLogger, InjectPinoLogger } from 'nestjs-pino';

@Injectable()
export class JwtAuthGuard implements CanActivate {
  constructor(
    private readonly jwtService: JwtService,
    @InjectPinoLogger(JwtAuthGuard.name)
    private readonly logger: PinoLogger,
  ) {}

  canActivate(context: ExecutionContext): boolean {
    const request = context.switchToHttp().getRequest();
    const token = request.headers.authorization?.replace('Bearer ', '');

    const payload = this.jwtService.verify(token);
    request.user = payload;

    this.logger.assign({ userId: payload.sub });  // обогащение ALS — только здесь
    return true;
  }
}

logger.assign у nestjs-pino обогащает ALS-контекст текущего request — все последующие log-записи в этом request будут содержать userId. Guard выполняется после nestjs-pino middleware, значит ALS уже инициализирован.

Писать userId в service-методе — нарушение R-OBS-CTX-X2: clear-логика неочевидна, ALS будет загрязнён для следующих вызовов на той же async-цепочке.

BullMQ: явная передача контекста

R-OBS-CTX-3: AsyncLocalStorage рвётся при передаче задачи в worker_threads или отдельный BullMQ-процесс. Jobdata пересекает границу через сериализованный payload — ALS не переходит.

Паттерн — кладём requestId и traceparent в payload явно при постановке задачи:

// src/order/order.service.ts
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';
import { context, propagation } from '@opentelemetry/api';

@Injectable()
export class OrderService {
  constructor(
    @InjectQueue('notifications') private readonly notificationsQueue: Queue,
    @InjectPinoLogger(OrderService.name) private readonly logger: PinoLogger,
  ) {}

  async placeOrder(customerId: string, items: OrderItem[]): Promise<string> {
    const order = await this.ordersRepository.create(customerId, items);

    const carrier: Record<string, string> = {};
    propagation.inject(context.active(), carrier);  // traceparent + tracestate

    await this.notificationsQueue.add('order-placed', {
      orderId: order.id,
      customerId,
      requestId: this.logger.logger.bindings()['requestId'],  // из ALS
      traceparent: carrier['traceparent'],
      tracestate: carrier['tracestate'],
    });

    this.logger.info({ orderId: order.id }, 'order placed, notification queued');
    return order.id;
  }
}

В processor-е восстанавливаем контекст до начала обработки:

// src/notifications/processors/order-placed.processor.ts
import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';
import { propagation, context, trace } from '@opentelemetry/api';
import { PinoLogger, InjectPinoLogger } from 'nestjs-pino';

@Processor('notifications')
export class OrderPlacedProcessor extends WorkerHost {
  constructor(
    @InjectPinoLogger(OrderPlacedProcessor.name)
    private readonly logger: PinoLogger,
    private readonly notificationService: NotificationService,
  ) {
    super();
  }

  async process(job: Job): Promise<void> {
    const { orderId, customerId, requestId, traceparent, tracestate } = job.data;

    const carrier = { traceparent, tracestate };
    const parentContext = propagation.extract(context.active(), carrier);

    await context.with(parentContext, async () => {
      this.logger.assign({ requestId, orderId, customerId });
      this.logger.info({ orderId }, 'processing order-placed notification');

      await this.notificationService.sendOrderConfirmation(customerId, orderId);
    });
  }
}

context.with(parentContext, fn) — устанавливает OTel-контекст для всего async-дерева внутри fn, связывая processor-span с originating trace. Без этого Tempo покажет два несвязанных дерева — дебаггинг инцидента превращается в угадывание.

nestjs-cls как альтернатива

Если проект использует nestjs-cls (ClsModule) вместо прямого ALS — паттерн аналогичен, но через ClsService:

// src/common/interceptors/request-context.interceptor.ts
import { Injectable, NestInterceptor, ExecutionContext, CallHandler } from '@nestjs/common';
import { ClsService } from 'nestjs-cls';
import { Observable } from 'rxjs';
import { randomUUID } from 'node:crypto';

@Injectable()
export class RequestContextInterceptor implements NestInterceptor {
  constructor(private readonly cls: ClsService) {}

  intercept(context: ExecutionContext, next: CallHandler): Observable<unknown> {
    const request = context.switchToHttp().getRequest();
    const requestId = request.headers['x-request-id'] ?? randomUUID();
    this.cls.set('requestId', requestId);
    return next.handle();
  }
}

// в guard после JWT:
this.cls.set('userId', payload.sub);

// в service — только чтение:
const requestId = this.cls.get<string>('requestId');
this.logger.info({ orderId, requestId }, 'product reserved');

ClsService.set — только в middleware/guard/interceptor (R-OBS-CTX-X2). В BullMQ processor — cls.run(store, fn) в обёртке аналогично context.with.

Порядок middleware и guard в NestJS

Порядок critical: OTel должен видеть ALS с requestId до создания span.

HTTP Request
  └─ nestjs-pino (LoggerModule) ← создаёт ALS, genReqId → X-Request-Id
       └─ OTel auto-instrumentation ← читает traceparent, создаёт span
            └─ Guards (JwtAuthGuard) ← logger.assign({ userId })
                 └─ Interceptors
                      └─ Controller handler

LoggerModule регистрируется через app.use(logger) в main.ts до app.listen() — это NestJS middleware, работает до Guards. OTel-инструментация патчит http-модуль на уровне tracing.ts до старта — никакого дополнительного порядка не требует.

Что запрещено

Куда дальше

• Конфигурация — management-порт на :9090, NODE_ENV-aware pino transport.
• Health checks — @nestjs/terminus liveness/readiness с TTL-кешем.
• Logging — merge-объект вместо конкатенации, redact для PII, { err } для ошибок.
• Metrics — RED-histogram по шаблону роута, collectDefaultMetrics, low-cardinality labels.
• SLO и алерты — SLO recording rules и multi-window burn-rate alerts.
• Tracing — NodeSDK, startActiveSpan, sampling 1–10%, нет PII в атрибутах.