SpringCloudGateway uses Skywalking when logging prints traceId

Environmental information

• SpringCloudGateway 3.1.3
• Skywalking Agent 8.10.0

Environment configuration

Agent

Since SpringCloudGateway is implemented based on WebFlux, you need to go to the agent directory of skywalking and copy the following two jar packages under the optional-plugins directory to the plugins directory

• apm-spring-webflux-5.x-plugin-8.10.0.jar
• apm-spring-cloud-gateway-3.x-plugin-8.10.0.jar

Maven relies on configuration

<dependency>
  <groupId>org.apache.skywalking</groupId>
    <artifactId>apm-toolkit-log4j-2.x</artifactId>
    <version>${skywalking.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.skywalking</groupId>
    <artifactId>apm-toolkit-trace</artifactId>
    <version>${skywalking.version}</version>
</dependency>           

Log pattern configuration

[%d{yyyy-MM-dd HH:mm:ss.SSS}] [%traceId] [%logger{36}] [%thread] [%-5level] %msg%n           

Startup parameters

Added startup parameters

-javaagent:D:\work\skywalking-agent\skywalking-agent.jar=agent.service_name=xxx
-Dskywalking.collector.backend_service=xxx:11800           

After starting the program, try to make an interface call through the gateway, and you can see on Skywalking-ui that the link has been stringed

Invoke Link Invoke the link

However, there is a problem that the log recorded in the log does not always show the correct TID

[2022-06-15 14:53:19.958] [TID: N/A]           

Problem handling process

See how the agent connects the links in series

Looking at the source code of Skywalking-agent, you can see that in the apm-spring-webflux-5.x-plugin-8.10.0 plugin.jar handle method is intercepted

Put < SKYWALKING_CONTEXT_SNAPSHOT - ContextSnapshot > in the interceptor to the reactor's call link

So traceId can be obtained from the reactor's context

How to get traceId in the log

public static <T> Consumer<Signal<T>> logOnNext(Consumer<T> logStatement) {
    return signal -> {
        if (!signal.isOnNext()) return; (1)
        Optional<String> toPutInMdc = signal.getContext().getOrEmpty("CONTEXT_KEY"); (2)

        toPutInMdc.ifPresentOrElse(tpim -> {
            try (MDC.MDCCloseable cMdc = MDC.putCloseable("MDC_KEY", tpim)) { (3)
                logStatement.accept(signal.get()); (4)
            }
        },
        () -> logStatement.accept(signal.get())); (5)
    };
}

@GetMapping("/byPrice")
public Flux<Restaurant> byPrice(@RequestParam Double maxPrice, @RequestHeader(required = false, name = "X-UserId") String userId) {
    String apiId = userId == null ? "" : userId; (1)

    return restaurantService.byPrice(maxPrice))
               .doOnEach(logOnNext(r -> LOG.debug("found restaurant {} for ${}", (2)
                    r.getName(), r.getPricePerPerson())))
               .contextWrite(Context.of("CONTEXT_KEY", apiId)); (3)
}
           

How to display the default value when the traceId cannot be obtained

equals{pattern}{test}{substitution}
equalsIgnoreCase{pattern}{test}{substitution}           

Full example

1. pattern改为 [%d{yyyy-MM-dd HH:mm:ss. SSS}] [TID: %equals{%X{traceId}}{}{N/A}] [%logger{36}] [%thread] [%-5level] %msg%n
2. Sign up for onEachOperator's Hooks

@Component
public class LogHooks {
    
    private static final String KEY = "logMdc";
    
    @PostConstruct
    @SuppressWarnings("unchecked")
    public void setHook() {
        reactor.core.publisher.Hooks.onEachOperator(KEY,
                Operators.lift((scannable, coreSubscriber) -> new MdcSubscriber(coreSubscriber)));
    }
    
    @PreDestroy
    public void resetHook() {
        reactor.core.publisher.Hooks.resetOnEachOperator(KEY);
    }
    
}
           

public class MdcSubscriber implements CoreSubscriber {
    
    private static final String TRACE_ID = "traceId";
    
    private static final String SKYWALKING_CTX_SNAPSHOT = "SKYWALKING_CONTEXT_SNAPSHOT";
    
    private final CoreSubscriber<Object> actual;
    
    public MdcSubscriber(CoreSubscriber<Object> actual) {
        this.actual = actual;
    }
    
    @Override
    public void onSubscribe(Subscription s) {
        actual.onSubscribe(s);
    }
    
    @Override
    public void onNext(Object o) {
        Context c = actual.currentContext();
        Optional<String> traceIdOptional = Optional.empty();
        if (!c.isEmpty() && c.hasKey(SKYWALKING_CTX_SNAPSHOT)) {
            traceIdOptional = Optional.of(c.get(SKYWALKING_CTX_SNAPSHOT)).map(BeanUtil::beanToMap)
                    .map(t -> t.get(TRACE_ID)).map(BeanUtil::beanToMap).map(t -> t.get("id")).map(Object::toString);
        }
        try (MDC.MDCCloseable cMdc = MDC.putCloseable(TRACE_ID, traceIdOptional.orElse("N/A"))) {
            actual.onNext(o);
        }
    }
    
    @Override
    public void onError(Throwable throwable) {
        actual.onError(throwable);
    }
    
    @Override
    public void onComplete() {
        actual.onComplete();
    }
    
    @Override
    public Context currentContext() {
        return actual.currentContext();
    }
}