NestJS 分布式追踪：AsyncLocalStorage + Zipkin/Jaeger 实战指南

NestJS 分布式追踪：AsyncLocalStorage + Zipkin/Jaeger 实战指南

你好！在微服务架构中，一个请求往往会跨越多个服务，这使得问题排查和性能分析变得异常困难。分布式追踪技术应运而生，它能够帮助我们清晰地了解请求在各个服务中的流转情况，从而快速定位问题和瓶颈。今天，我们就来聊聊如何在 NestJS 中利用 AsyncLocalStorage 实现分布式追踪，并集成 Zipkin 和 Jaeger 这两个流行的追踪系统。

为什么需要分布式追踪？

在单体应用中，我们可以通过查看日志或使用调试器来跟踪请求的执行过程。但在微服务架构中，一个请求可能涉及多个服务，这些服务可能部署在不同的机器上，甚至使用不同的编程语言。传统的调试方法在这种情况下就显得力不从心了。

分布式追踪系统通过为每个请求分配一个唯一的 Trace ID，并在请求跨越服务边界时传递这个 ID，从而将整个请求链路串联起来。我们可以通过可视化工具（如 Zipkin 或 Jaeger）查看请求经过了哪些服务、每个服务的耗时、以及服务之间的调用关系，极大地提高了问题排查和性能优化的效率。

AsyncLocalStorage：Node.js 中的追踪利器

AsyncLocalStorage 是 Node.js 提供的一个核心模块（从 v13.10.0 开始稳定），它允许我们在异步操作之间共享数据，而无需显式地传递上下文。这对于实现分布式追踪至关重要，因为我们可以在请求入口处生成 Trace ID，并将其存储在 AsyncLocalStorage 中，然后在整个请求处理过程中随时访问这个 ID，而无需将其作为参数在各个函数之间传递。

AsyncLocalStorage 的基本用法

import { AsyncLocalStorage } from 'async_hooks';
 
// 创建一个 AsyncLocalStorage 实例
const asyncLocalStorage = new AsyncLocalStorage<Map<string, any>>();
 
// 在请求入口处运行一个函数，并传入一个初始的 store
asyncLocalStorage.run(new Map(), () => {
  // 在 store 中设置数据
  asyncLocalStorage.getStore().set('key', 'value');
 
  // 异步操作
  setTimeout(() => {
    // 在异步操作中访问 store 中的数据
    console.log(asyncLocalStorage.getStore().get('key')); // 输出: value
  }, 100);
});

AsyncLocalStorage.run() 方法接受一个初始的 store（通常是一个 Map 对象）和一个回调函数。在回调函数中，我们可以通过 asyncLocalStorage.getStore() 方法获取当前的 store，并对其进行读写操作。即使在异步操作中，我们也可以访问到正确的 store。

NestJS 中的分布式追踪实践

现在，让我们看看如何在 NestJS 项目中利用 AsyncLocalStorage 和 Zipkin/Jaeger 实现分布式追踪。

1. 安装依赖

npm install --save @nestjs/core @nestjs/common @nestjs/platform-express @opentelemetry/api @opentelemetry/sdk-node @opentelemetry/sdk-trace-base @opentelemetry/exporter-zipkin @opentelemetry/exporter-jaeger @opentelemetry/instrumentation-http @opentelemetry/instrumentation-express @opentelemetry/resources @opentelemetry/semantic-conventions

我们安装了 NestJS 的核心模块，以及 OpenTelemetry 相关的包。OpenTelemetry 是一个开源的可观测性框架，它提供了一套统一的 API 和工具，用于收集、处理和导出遥测数据（包括 Traces、Metrics 和 Logs）。我们将使用 OpenTelemetry SDK 来创建和管理 Trace，并将其导出到 Zipkin 或 Jaeger。

2. 创建 Trace ID 生成器

// src/trace/trace-id.generator.ts
import { Injectable } from '@nestjs/common';
import { v4 as uuidv4 } from 'uuid';
 
@Injectable()
export class TraceIdGenerator {
  generate(): string {
    return uuidv4();
  }
}

我们创建了一个简单的服务 TraceIdGenerator，用于生成唯一的 Trace ID。这里使用了 uuid 库来生成 UUID v4 格式的 ID。

3. 创建 AsyncLocalStorage 提供者

// src/trace/trace.module.ts
import { Module, Global } from '@nestjs/common';
import { AsyncLocalStorage } from 'async_hooks';
import { TraceIdGenerator } from './trace-id.generator';
 
@Global()
@Module({
  providers: [
    TraceIdGenerator,
    {
      provide: AsyncLocalStorage,
      useValue: new AsyncLocalStorage<Map<string, any>>(),
    },
  ],
  exports: [TraceIdGenerator, AsyncLocalStorage],
})
export class TraceModule {}

我们创建了一个全局模块 TraceModule，并提供了 AsyncLocalStorage 实例。@Global() 装饰器使得这个模块在整个应用中都可以使用，而无需在每个模块中单独导入。

4. 创建拦截器

// src/trace/trace.interceptor.ts
import { Injectable, NestInterceptor, ExecutionContext, CallHandler, Inject } from '@nestjs/common';
import { Observable } from 'rxjs';
import { tap } from 'rxjs/operators';
import { AsyncLocalStorage } from 'async_hooks';
import { TraceIdGenerator } from './trace-id.generator';
 
@Injectable()
export class TraceInterceptor implements NestInterceptor {
  constructor(
    private readonly traceIdGenerator: TraceIdGenerator,
    @Inject(AsyncLocalStorage) private readonly asyncLocalStorage: AsyncLocalStorage<Map<string, any>>,
  ) {}
 
  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    const request = context.switchToHttp().getRequest();
    const traceId = this.traceIdGenerator.generate();
    const store = new Map();
    store.set('traceId', traceId);
    //将traceId 设置到 header 中
    request.headers['x-trace-id'] = traceId;
 
    return this.asyncLocalStorage.run(store, () => {
      return next.handle().pipe(
        tap(() => {
          // 在响应返回后记录 Trace ID
          const response = context.switchToHttp().getResponse();
          response.setHeader('x-trace-id', traceId);
        }),
      );
    });
  }
}

我们创建了一个拦截器 TraceInterceptor，它会在每个请求进入时生成 Trace ID，并将其存储在 AsyncLocalStorage 中。同时，我们将 Trace ID 添加到请求头和响应头中（x-trace-id），以便在服务之间传递。

5. 配置 OpenTelemetry

// src/opentelemetry.ts
import { NodeSDK } from '@opentelemetry/sdk-node';
import { HttpInstrumentation } from '@opentelemetry/instrumentation-http';
import { ExpressInstrumentation } from '@opentelemetry/instrumentation-express';
import { Resource } from '@opentelemetry/resources';
import { SemanticResourceAttributes } from '@opentelemetry/semantic-conventions';
import { ZipkinExporter } from '@opentelemetry/exporter-zipkin';
//import { JaegerExporter } from '@opentelemetry/exporter-jaeger'; //如果需要使用Jaeger,打开注释
import { BasicTracerProvider, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base';
 
const sdk = new NodeSDK({
  resource: new Resource({
    [SemanticResourceAttributes.SERVICE_NAME]: 'my-nestjs-app', // 你的服务名称
  }),
  instrumentations: [new HttpInstrumentation(), new ExpressInstrumentation()],
  traceExporter: new ZipkinExporter({ // 或者 new JaegerExporter()
     url: 'http://localhost:9411/api/v2/spans', // Zipkin 或 Jaeger 的地址
  }),
});
 
sdk.start();
 
process.on('SIGTERM', () => {
  sdk.shutdown()
    .then(() => console.log('Tracing terminated'))
    .catch((error) => console.error('Error terminating tracing', error))
    .finally(() => process.exit(0));
});

我们创建了一个 opentelemetry.ts 文件，用于配置 OpenTelemetry SDK。这里我们启用了 HttpInstrumentation 和 ExpressInstrumentation，它们会自动收集 HTTP 请求和 Express 框架相关的 Trace 信息。我们还配置了 ZipkinExporter（或 JaegerExporter），用于将 Trace 数据导出到 Zipkin 或 Jaeger。你需要根据你的实际情况修改 SERVICE_NAME 和 url。

6. 在 main.ts 中引入配置

// src/main.ts
import './opentelemetry'; // 引入 OpenTelemetry 配置
import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { TraceInterceptor } from './trace/trace.interceptor';
 
async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  app.useGlobalInterceptors(new TraceInterceptor()); // 使用全局拦截器
  await app.listen(3000);
}
bootstrap();

在 main.ts 中，我们引入了 opentelemetry.ts，并使用 app.useGlobalInterceptors() 方法注册了 TraceInterceptor。

7. 在服务中使用 AsyncLocalStorage

// src/app.service.ts
import { Injectable, Inject } from '@nestjs/common';
import { AsyncLocalStorage } from 'async_hooks';
 
@Injectable()
export class AppService {
  constructor(@Inject(AsyncLocalStorage) private readonly asyncLocalStorage: AsyncLocalStorage<Map<string, any>>) {}
 
  getHello(): string {
    const traceId = this.asyncLocalStorage.getStore().get('traceId');
    console.log(`Processing request with trace ID: ${traceId}`);
    return 'Hello World!';
  }
}

在你的服务中，你可以通过注入 AsyncLocalStorage 实例来访问 Trace ID。例如，在 AppService 中，我们打印了当前的 Trace ID。

8. 启动 Zipkin 或 Jaeger

你可以使用 Docker 来快速启动 Zipkin 或 Jaeger：

Zipkin:

docker run -d -p 9411:9411 openzipkin/zipkin

Jaeger:

docker run -d --name jaeger \
  -e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \
  -p 5775:5775/udp \
  -p 6831:6831/udp \
  -p 6832:6832/udp \
  -p 5778:5778 \
  -p 16686:16686 \
  -p 14268:14268 \
  -p 14250:14250 \
  -p 9411:9411 \
  jaegertracing/all-in-one:latest

9. 测试

启动你的 NestJS 应用，并发送几个请求。然后打开 Zipkin 或 Jaeger 的 UI（Zipkin: http://localhost:9411, Jaeger: http://localhost:16686），你应该能看到请求的 Trace 信息。

常见问题解答

• Q: 如何在服务之间传递 Trace ID？

  A: 我们已经在 TraceInterceptor 中将 Trace ID 添加到了请求头（x-trace-id）中。如果你的服务之间通过 HTTP 通信，这个头部会自动传递。如果使用其他通信方式（如消息队列），你需要手动将 Trace ID 添加到消息中。

• Q: 如何记录自定义的 Span？

  A: 你可以使用 OpenTelemetry API 来创建自定义的 Span。例如：

  import { trace, context } from '@opentelemetry/api';
   
  const tracer = trace.getTracer('my-tracer');
   
  const span = tracer.startSpan('my-span');
  // ... 执行一些操作 ...
   const currentStore = this.asyncLocalStorage.getStore();
  if (currentStore) {
      const traceId = currentStore.get('traceId');
      if (traceId) {
          span.setAttribute('traceId', traceId);
      }
  }
  span.end();

  将 AsyncLocalStorage 中的traceId 注入到span中

  const currentStore = this.asyncLocalStorage.getStore();
      if (currentStore) {
          const traceId = currentStore.get('traceId');
          if (traceId) {
              span.setAttribute('traceId', traceId);
          }
      }

• Q: 如何处理异步操作？

  A: AsyncLocalStorage 已经处理了异步操作。只要你在 asyncLocalStorage.run() 的回调函数中执行代码，AsyncLocalStorage 就能保证在异步操作之间正确地传递上下文。

总结

通过本文，相信你已经掌握了在 NestJS 中使用 AsyncLocalStorage 实现分布式追踪，并集成 Zipkin 和 Jaeger 的方法。这只是分布式追踪的入门，OpenTelemetry 还提供了许多高级功能，如自定义 Span、采样、传播上下文等。希望你在实际项目中能够灵活运用这些技术，构建出更健壮、更易于维护的微服务应用！

如果你在实践过程中遇到任何问题，欢迎随时提问，我会尽力帮助你。 祝你在 NestJS 的世界里玩得开心！