使用 Spring Webflux 实现 trace-id

Question

我想为每个请求生成唯一的 traceId 并将其传递给所有服务。在 Spring MVC 中，使用 MDC 上下文并将 traceId 放在 header 中相当容易，但在响应式堆栈中，由于 ThreadLocal，它根本不起作用。

一般来说，我想用单个 traceId 记录我拥有的每个服务的每个请求和响应，它可以识别整个系统中的特定操作。

我尝试根据文章创建自定义过滤器：https://azizulhaq-ananto.medium.com/how-to-handle-logs-and-tracing-in-spring-webflux-and-microservices-a0b45adc4610，但它似乎不起作用。 我目前的解决方案只记录响应和 traceId 在发出请求后丢失，所以没有响应。 让我们尝试假设有两个服务：service1 和 service2。下面我试图勾勒出它应该如何工作。

它应该如何工作

1. client -> service1 - service1 应该生成 traceId 并记录请求
2. service1 -> service2 - service2 应该从请求中获取 traceId，然后记录请求
3. service1 service2 - 经过一些计算，service2 应该记录响应并将响应返回给 service1
4. client service1 - 最后 service1 应该记录响应（仍然具有相同的 traceId）并将响应返回给客户端

实际工作原理

1. client -> service1 - 日志中没有任何内容
2. service1 -> service2 - 日志中没有任何内容
3. service1 service2 - service2 正在正确记录并向 service1 返回响应
4. client service1 - service1 正在记录响应（但没有 traceId）

这是我的方法

@Component
public class TraceIdFilter implements WebFilter {

    private static final Logger log = LoggerFactory.getLogger(TraceIdFilter.class);

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        Map<String, String> headers = exchange.getRequest().getHeaders().toSingleValueMap();
        return Mono.fromCallable(() ->  {
            final long startTime = System.currentTimeMillis();

            return new ServerWebExchangeDecorator(exchange) {
                @Override
                public ServerHttpRequest getRequest() {
                    return new RequestLoggingInterceptor(super.getRequest(), false);
                }

                @Override
                public ServerHttpResponse getResponse() {
                    return new ResponseLoggingInterceptor(super.getResponse(), startTime, false);
                }
            };
        }).contextWrite(context -> {
            var traceId = "";
            if (headers.containsKey("X-B3-TRACEID")) {
                traceId = headers.get("X-B3-TRACEID");
                MDC.put("X-B3-TraceId", traceId);
            } else if (!exchange.getRequest().getURI().getPath().contains("/actuator")) {
                traceId = UUID.randomUUID().toString();
                MDC.put("X-B3-TraceId", traceId);
            }

            Context contextTmp = context.put("X-B3-TraceId", traceId);
            exchange.getAttributes().put("X-B3-TraceId", traceId);


            return contextTmp;
        }).flatMap(chain::filter);

    }


}

Github：https://github.com/Faelivrinx/kotlin-spring-boot

有任何现有的解决方案吗？

Answer 1

在 Spring Webflux 中，您不再有 ThreadLocal，但您对每个链请求都有一个唯一的上下文。您可以将 traceId 附加到此上下文，如下所示：

@Component
public class TraceIdFilter implements WebFilter {

    private static final Logger log = LoggerFactory.getLogger(TraceIdFilter.class);

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {

        return chain.filter(exchange)
                .subscriberContext(
                        ctx -> {
                            .....
                            var traceId = UUID.randomUUID().toString();
                            return   ctx.put("X-B3-TraceId", traceId);
                            .....
                        }
                );

    }


}

现在您的服务中的链将在上下文中具有此属性。您可以使用静态方法 Mono.subscriberContext() 从您的服务中检索它。比如可以通过这种方式获取traceId

Mono.subscriberContext()
    .flaMap(ctx -> {
       .....
       var traceId = ctx.getOrDefault("traceId", null);
       .....
    )

Answer 2

Sleuth 3.0 为 WebFlux 提供自动检测，这意味着，如果您什么都不做，您将始终获得订阅 Mono 或 Flux 的线程的当前跨度。如果您希望覆盖它，例如因为您正在批处理大量请求并希望为每笔交易提供唯一的跟踪，那么您所要做的就是操纵您的运营商链的上下文。

private Mono<Data> performRequest() {
    var span = tracer.spanBuilder().setNoParent().start(); // generate a completely new trace
                                                           // note: you can also just generate a new span if you want
    return Mono.defer(() -> callRealService())
        .contextWrite(Context.of(TraceContext.class, span.context());
}

当采用这种方法时，请确保导入 org.springframework.cloud.sleuth.Tracer 而不是勇敢的，因为它们使用不同的类型，如果它们没有正确对齐，Reactor 会丢弃你的 Mono 并显示一个丑陋的错误消息（不幸的是，因为上下文只是一个普通的旧Map<Object, Object>，不会出现编译器错误）。