Context propagation в NestJS: AsyncLocalStorage, middleware и очереди

← назад к разделу

Представь: в логах видно сотни записей от разных запросов, всё перемешано. Как понять, какие из них относятся к конкретному запросу, упавшему с ошибкой? Для этого каждый запрос должен нести свой идентификатор — requestId — и он должен автоматически попадать в каждую log-запись, порождённую этим запросом.

В Java для этого используют MDC (Map Diagnostic Context) — thread-local хранилище, где каждый поток хранит свои данные. В Node.js потоков нет: event loop один, и через него проходят все запросы. AsyncLocalStorage — это аналог MDC для Node.js: контейнер, который хранит данные отдельно для каждого «async-дерева» запроса.

Что такое AsyncLocalStorage

AsyncLocalStorage — встроенный в Node.js механизм (модуль async_hooks). Его суть: запускаешь функцию через als.run(store, fn), и всё, что вызывается внутри fn — включая все await и вложенные асинхронные вызовы — видит store. Другие запросы получают свой store и не пересекаются.

Это значит, что requestId можно положить в ALS один раз при старте запроса, и дальше он будет доступен в любом сервисе, репозитории или логгере — без явной передачи через аргументы.

nestjs-pino делает именно это: при каждом входящем HTTP-запросе создаёт ALS-контекст с уникальным requestId и per-request логгер, который автоматически добавляет requestId в каждую запись.

Настройка nestjs-pino

Настройка через LoggerModule.forRoot в корневом модуле:

// 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 — функция, которая определяет requestId для каждого запроса. Если клиент прислал X-Request-Id в заголовке — используем его (это удобно для трассировки между сервисами). Иначе генерируем UUID. nestjs-pino также возвращает requestId в ответе, чтобы клиент мог использовать его при разборе проблемы.

Использовать логгер в сервисе:

// 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 не знают об ALS — записи через них не будут содержать requestId. Везде используй @InjectPinoLogger().

trace_id и span_id из OpenTelemetry

Кроме requestId в логах нужны trace_id и span_id — идентификаторы текущего трейса. Они нужны, чтобы в Grafana/Tempo можно было перейти от конкретной log-записи к полному дереву вызовов.

@opentelemetry/instrumentation-pino добавляет trace_id и span_id автоматически в каждую pino-запись — ничего дополнительно писать не нужно. Главное условие: tracing.ts должен быть импортирован первой строкой main.ts, до всего остального.

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

const sdk = new NodeSDK({
  serviceName: process.env.OTEL_SERVICE_NAME ?? 'order-service',
  traceExporter: new OTLPTraceExporter(),
  instrumentations: [
    getNodeAutoInstrumentations(),
    new PinoInstrumentation(),  // добавляет trace_id/span_id в каждый лог
  ],
});

sdk.start();

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

Если tracing.ts импортирован позже — модули уже загружены без патча, и интеграция не заработает. Не добавляй trace_id в логи руками через logger.assign({ traceId: ... }): при смене активного span поле устареет.

userId в guard после JWT-валидации

requestId устанавливается на старте запроса — до аутентификации. userId доступен только после JWT-валидации. Поэтому userId нужно добавлять в ALS-контекст в guard-е, а не в middleware.

// 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-контекст текущего запроса — все последующие log-записи в рамках этого запроса будут содержать userId. Guard выполняется после nestjs-pino middleware, значит ALS уже инициализирован.

Не добавляй userId в сервисном методе: там логика очистки неочевидна, и контекст будет загрязнён для других вызовов.

BullMQ: куда девается контекст

Вот где есть настоящая ловушка. AsyncLocalStorage работает только в рамках одного async-дерева. Когда ты ставишь задачу в очередь BullMQ — задача уходит в отдельный worker_thread или отдельный процесс. ALS туда не переходит: сериализованный payload пересекает границу потока, а вместе с ним — никакого контекста.

Решение простое: передать requestId и traceparent явно в payload задачи при постановке, и восстановить контекст в обработчике.

Постановка задачи:

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

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

  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.cls.get<string>('requestId'),  // явно берём из контекста
      traceparent: carrier['traceparent'],
      tracestate: carrier['tracestate'],
    });

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

Обработчик задачи — восстанавливаем контекст:

// 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. Это связывает span обработчика с исходным трейсом — в 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);

В сервисах — только чтение, не запись:

const requestId = this.cls.get<string>('requestId');
this.logger.info({ orderId, requestId }, 'product reserved');

Порядок слоёв в NestJS

Важно понимать, в каком порядке работают слои — это влияет на то, когда ALS уже готов, а когда ещё нет:

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

LoggerModule регистрируется через app.use(logger) в main.ts до app.listen(). OTel патчит http-модуль ещё на уровне tracing.ts — до старта приложения.

Частые ошибки

Общий мутируемый стор вместо per-request ALS. Если хранить requestId в обычном объекте или модульной переменной — значение одного запроса перезапишет значение другого. Утечка userId соседнего запроса в логи — серьёзная проблема. Используй nestjs-pino per-request ALS или ClsModule.

logger.assign({ userId }) в сервисном методе. Логика очистки неочевидна, и последующие вызовы могут получить устаревший userId. Обогащать контекст только в middleware, guard или interceptor.

BullMQ job без requestId и traceparent в payload. Без явной передачи нет способа связать log-записи обработчика с исходным запросом.

logger.assign({ traceId: span.traceId }) вручную. При смене активного span поле устареет. Используй @opentelemetry/instrumentation-pino.

tracing.ts импортируется после NestFactory. Модули уже загружены без OTel-патча, интеграция не работает. import './tracing' — первая строка main.ts.

Коротко

• AsyncLocalStorage — аналог Java MDC для Node.js: хранит данные отдельно для каждого async-дерева запроса, без явной передачи через аргументы.
• nestjs-pino создаёт per-request ALS-контекст через genReqId — requestId автоматически в каждой записи.
• trace_id и span_id добавляет @opentelemetry/instrumentation-pino — не нужно прописывать руками.
• tracing.ts импортировать первой строкой main.ts — до NestFactory.
• userId добавлять в ALS только в guard после JWT-валидации, не в middleware и не в сервисах.
• BullMQ разрывает ALS — requestId и traceparent передавать явно в payload задачи и восстанавливать через propagation.extract в обработчике.
• Обогащение контекста (logger.assign, cls.set) — только в middleware, guard, interceptor. В сервисах — только чтение.

Что почитать дальше

• Logging в NestJS — структура log-записи, redact для персональных данных, обработка ошибок.
• Tracing в NestJS — NodeSDK, активные span-ы, sampling, ограничения по данным.
• Metrics в NestJS — RED-гистограммы, collectDefaultMetrics, low-cardinality метки.
• Health checks в NestJS — liveness/readiness через @nestjs/terminus.