[Feature] Support Tracing for GlobalFilter and GatewayFilter in spring-cloud-gateway-plugin

[yqw570994511]

Search before asking

• I had searched in the issues and found no similar feature requirement.

Description

In Spring Cloud Gateway, GlobalFilter and GatewayFilter are crucial components for handling requests. However, in certain scenarios, we have encountered issues with the generation and propagation of traceId within these filters. This can lead to difficulties in correctly correlating and tracing requests that pass through the Gateway in a distributed tracing system.

Here is a code snippet:
`
Component
public class Filter1 implements GlobalFilter, Ordered {

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

@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
	// The traceId cannot be obtained here
	log.info("filter1 traceId: {}", TraceContext.traceId());
	return chain.filter(exchange);
}

@Override
public int getOrder() {
	return -100;
}

}
`
Desired Result:
Able to obtain SkyWalking's traceId within a custom GlobalFilter.

My implementation result:

My solution code:

package org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.define;

import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import org.apache.skywalking.apm.agent.core.plugin.match.HierarchyMatch;

import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType;

public class GatewayFilterInstrumentation extends AbstractGateway210EnhancePluginDefine {

    @Override
    protected ClassMatch enhanceClass() {
        return HierarchyMatch.byHierarchyMatch("org.springframework.cloud.gateway.filter.GatewayFilter");
    }

    @Override
    public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
        return new ConstructorInterceptPoint[0];
    }

    @Override
    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
        return new InstanceMethodsInterceptPoint[] {
            new InstanceMethodsInterceptPoint() {
                @Override
                public ElementMatcher<MethodDescription> getMethodsMatcher() {
                    return named("filter").and(
                        takesArgumentWithType(0, "org.springframework.web.server.ServerWebExchange"));
                }

                @Override
                public String getMethodsInterceptor() {
                    return "org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.GatewayFilterInterceptor";
                }

                @Override
                public boolean isOverrideArgs() {
                    return true;
                }
            }
        };
    }
}

Croe Code:

package org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x;

import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.ServerWebExchangeDecorator;
import org.springframework.web.server.adapter.DefaultServerWebExchange;

import java.lang.reflect.Method;

import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.SPRING_CLOUD_GATEWAY;

public class GatewayFilterInterceptor implements InstanceMethodsAroundInterceptor {

    private int stackDeep = 0;

    @Override
    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                             MethodInterceptResult result) throws Throwable {
        stackDeep++;
        if (isEntry()) {
            ServerWebExchange exchange = (ServerWebExchange) allArguments[0];

            EnhancedInstance enhancedInstance = getInstance(exchange);

            AbstractSpan span = ContextManager.createLocalSpan("SpringCloudGateway/GatewayFilter");
            if (enhancedInstance != null && enhancedInstance.getSkyWalkingDynamicField() != null) {
                ContextManager.continued((ContextSnapshot) enhancedInstance.getSkyWalkingDynamicField());
            }
            span.setComponent(SPRING_CLOUD_GATEWAY);
        }
    }

    public static EnhancedInstance getInstance(Object o) {
        EnhancedInstance instance = null;
        if (o instanceof DefaultServerWebExchange) {
            instance = (EnhancedInstance) o;
        } else if (o instanceof ServerWebExchangeDecorator) {
            ServerWebExchange delegate = ((ServerWebExchangeDecorator) o).getDelegate();
            return getInstance(delegate);
        }
        return instance;
    }

    @Override
    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                              Object ret) throws Throwable {
        stackDeep--;
        if (ContextManager.isActive()) {
            if (isExit()) {
                ContextManager.stopSpan();
            }
        }
        return ret;
    }

    @Override
    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
                                      Class<?>[] argumentsTypes, Throwable t) {
        ContextManager.activeSpan().log(t);
    }

    private boolean isEntry() {
        return stackDeep == 1;
    }

    private boolean isExit() {
        return stackDeep <= 0;
    }
}

Use case

No response

Related issues

目前我找到了以下issue编号,https://github.com/apache/skywalking/issues/6900;本次的新特性可以帮助他们在GlobalFilter获取traceId

Are you willing to submit a pull request to implement this on your own?

• Yes I am willing to submit a pull request on my own!

Code of Conduct

• I agree to follow this project's Code of Conduct

[wu-sheng]

GatewayFilterInterceptor should not have a field to hold the context. Because it is a Singleton Class, this will cause race conditions.

[yqw570994511]

sorry, there was an error in the previous code. I have revised my code.

package org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x;

import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.ServerWebExchangeDecorator;
import org.springframework.web.server.adapter.DefaultServerWebExchange;

import java.lang.reflect.Method;
import java.util.concurrent.atomic.AtomicInteger;

import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.SPRING_CLOUD_GATEWAY;

public class GatewayFilterInterceptor implements InstanceMethodsAroundInterceptor {

    private static final ThreadLocal<AtomicInteger> STACK_DEEP = ThreadLocal.withInitial(() -> new AtomicInteger(0));

    @Override
    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                             MethodInterceptResult result) throws Throwable {
        STACK_DEEP.get().getAndIncrement();
        if (isEntry()) {
            ServerWebExchange exchange = (ServerWebExchange) allArguments[0];

            EnhancedInstance enhancedInstance = getInstance(exchange);

            AbstractSpan span = ContextManager.createLocalSpan("SpringCloudGateway/GatewayFilter");
            if (enhancedInstance != null && enhancedInstance.getSkyWalkingDynamicField() != null) {
                ContextManager.continued((ContextSnapshot) enhancedInstance.getSkyWalkingDynamicField());
            }
            span.setComponent(SPRING_CLOUD_GATEWAY);
        }
    }

    public static EnhancedInstance getInstance(Object o) {
        EnhancedInstance instance = null;
        if (o instanceof DefaultServerWebExchange) {
            instance = (EnhancedInstance) o;
        } else if (o instanceof ServerWebExchangeDecorator) {
            ServerWebExchange delegate = ((ServerWebExchangeDecorator) o).getDelegate();
            return getInstance(delegate);
        }
        return instance;
    }

    @Override
    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                              Object ret) throws Throwable {
        STACK_DEEP.get().getAndDecrement();
        if (ContextManager.isActive()) {
            if (isExit()) {
                STACK_DEEP.remove();
                ContextManager.stopSpan();
            }
        }
        return ret;
    }

    @Override
    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
                                      Class<?>[] argumentsTypes, Throwable t) {
        ContextManager.activeSpan().log(t);
    }

    private boolean isEntry() {
        return STACK_DEEP.get().intValue() == 1;
    }

    private boolean isExit() {
        return STACK_DEEP.get().intValue() <= 0;
    }
}

[wu-sheng]

STACK_DEEP is thread based, are GatewayFilter's implementation filters always executed in a same thread?

I am not familiar with GW codes, but usually we received issue reports about GW plugin because of x-thread scenarios.

[yqw570994511]

The main implementation code of GatewayFilter is as follows:

It can be seen that each GatewayFilter is running in the same thread.

The reason for using STACK_DEEP is to solve the problem of nested GatewayFilters.

There are two types of filters in the gateway, namely GatewayFilter and GlobalFilter, which are ultimately converted to GatewayFilter for unified processing.

The official provides two classes, GatewayFilterAdapter and OrderedGatewayFilter, to wrap GlobalFilter. When a GlobalFilter implements Ordered, it will be wrapped into GatewayFilterAdapter first, and then GatewayFilterAdapter will be wrapped into OrderedGatewayFilter.

The code for this part is as follows:

The definition and filter method of OrderedGatewayFilter class are as follows:

At this time, executing filter() on the OrderedGatewayFilter instance will trigger two filter() calls, but in fact we only need to create a localSpan at the entry point. The subsequent filter can already obtain the traceId, so I used STACK_DEEP to handle this situation

[wu-sheng]

OK, once they are in the same thread, I think it is fine.

When you submit a pull request, please be sure global filter are used in test scenarios, and the test cases could passed when the latest expected span files. And you may need to consider newer version, and the test scenarios may not cover the latest release of Spring GW.

[wu-sheng]

Further codes discussion should happen here, apache/skywalking-java#736

[wuwen5]

@yqw570994511 Referring to this will be helpful for you.

https://github.com/apache/skywalking-java/blob/main/docs/en/setup/service-agent/java-agent/Application-toolkit-webflux.md#fetch-trace-context-relative-ids

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain){
        // fetch trace ID
        String traceId = WebFluxSkyWalkingTraceContext.traceId(exchange);
        
        // fetch segment ID
        String segmentId = WebFluxSkyWalkingTraceContext.segmentId(exchange);
        
        // fetch span ID
        int spanId = WebFluxSkyWalkingTraceContext.spanId(exchange);
        
        return chain.filter(exchange);
    }

apache/skywalking-java#539

[wu-sheng]

The final status of this discussion ends at #12860