Support for tracing webflux-6.x and gateway-4.x

.github/workflows/plugins-jdk17-test.1.yaml

@@ -57,6 +57,7 @@ jobs:
         case:
           - spring-6.x-scenario
           - resteasy-6.x-scenario
+          - gateway-4.x-scenario
     steps:
       - uses: actions/checkout@v2
         with:

CHANGES.md

@@ -7,6 +7,8 @@ Release Notes.
 
 * Fix NoSuchMethodError in mvc-annotation-commons and change deprecated method.
 * fix forkjoinpool plugin in JDK11。
+* Support for tracing spring-cloud-gateway 4.x in gateway-4.x-plugin.
+
 
 #### Documentation
 

apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-3.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v3x/define/AbstractGatewayV3EnhancePluginDefine.java

@@ -20,13 +20,16 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
 
 /**
- * This abstract class defines the <code>witnessClasses()</code> method,
- * and other plugin define classes need to inherit from this class
+ * This abstract class defines the <code>witnessClasses()</code> method, and other plugin define classes need to inherit
+ * from this class
  */
 public abstract class AbstractGatewayV3EnhancePluginDefine extends ClassInstanceMethodsEnhancePluginDefine {
 
     @Override
     protected String[] witnessClasses() {
-        return new String[]{"org.springframework.cloud.gateway.config.GatewayLoadBalancerProperties"};
+        return new String[] {
+            "org.springframework.cloud.gateway.config.GatewayLoadBalancerProperties",
+            "org.springframework.web.client.AsyncRestTemplate"
+        };
     }
 }

apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-4.x-plugin/pom.xml

@@ -0,0 +1,78 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one or more
+  ~ contributor license agreements.  See the NOTICE file distributed with
+  ~ this work for additional information regarding copyright ownership.
+  ~ The ASF licenses this file to You under the Apache License, Version 2.0
+  ~ (the "License"); you may not use this file except in compliance with
+  ~ the License.  You may obtain a copy of the License at
+  ~
+  ~     http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  ~
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>optional-spring-cloud</artifactId>
+        <groupId>org.apache.skywalking</groupId>
+        <version>9.2.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>apm-spring-cloud-gateway-4.x-plugin</artifactId>
+    <packaging>jar</packaging>
+    <url>http://maven.apache.org</url>
+
+    <properties>
+        <spring-cloud-starter-gateway.version>4.0.0</spring-cloud-starter-gateway.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-gateway</artifactId>
+            <version>${spring-cloud-starter-gateway.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>apm-spring-webflux-6.x-plugin</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <version>3.2.4</version>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                        <configuration>
+                            <filters>
+                                <filter>
+                                    <artifact>org.apache.skywalking:apm-spring-webflux-6.x-plugin</artifact>
+                                    <excludes>
+                                        <exclude>skywalking-plugin.def</exclude>
+                                    </excludes>
+                                </filter>
+                            </filters>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v4x/HttpClientFinalizerConstructorInterceptor.java

@@ -0,0 +1,43 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.apache.skywalking.apm.plugin.spring.cloud.gateway.v4x;
+
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor;
+import org.apache.skywalking.apm.plugin.spring.cloud.gateway.v4x.define.EnhanceObjectCache;
+import reactor.netty.http.client.HttpClientConfig;
+
+/**
+ * Intercept the constructor and inject {@link EnhanceObjectCache}.
+ * <p>
+ * The first constructor argument is {@link reactor.netty.http.client.HttpClientConfig} class instance which can get the
+ * request uri string.
+ */
+public class HttpClientFinalizerConstructorInterceptor implements InstanceConstructorInterceptor {
+
+    @Override
+    public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
+        final HttpClientConfig httpClientConfig = (HttpClientConfig) allArguments[0];
+        if (httpClientConfig == null) {
+            return;
+        }
+        final EnhanceObjectCache enhanceObjectCache = new EnhanceObjectCache();
+        enhanceObjectCache.setUrl(httpClientConfig.uri());
+        objInst.setSkyWalkingDynamicField(enhanceObjectCache);
+    }
+}

apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v4x/HttpClientFinalizerResponseConnectionInterceptor.java

@@ -0,0 +1,107 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.apache.skywalking.apm.plugin.spring.cloud.gateway.v4x;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import org.apache.skywalking.apm.agent.core.context.tag.Tags;
+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.apache.skywalking.apm.plugin.spring.cloud.gateway.v4x.define.EnhanceObjectCache;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.SignalType;
+import reactor.netty.Connection;
+import reactor.netty.http.client.HttpClientResponse;
+
+import java.lang.reflect.Method;
+import java.util.function.BiFunction;
+
+/**
+ * This class intercept <code>responseConnection</code> method.
+ * <p>
+ * After downstream service response, finish the span in the {@link EnhanceObjectCache}.
+ */
+public class HttpClientFinalizerResponseConnectionInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override
+    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
+            MethodInterceptResult result) {
+        BiFunction<? super HttpClientResponse, ? super Connection, ? extends Publisher> finalReceiver = (BiFunction<? super HttpClientResponse, ? super Connection, ? extends Publisher>) allArguments[0];
+        EnhanceObjectCache cache = (EnhanceObjectCache) objInst.getSkyWalkingDynamicField();
+        allArguments[0] = (BiFunction<HttpClientResponse, Connection, Publisher>) (response, connection) -> {
+            Publisher publisher = finalReceiver.apply(response, connection);
+            if (cache == null) {
+                return publisher;
+            }
+            // receive the response.
+            if (cache.getSpan() != null) {
+                if (response.status().code() >= HttpResponseStatus.BAD_REQUEST.code()) {
+                    cache.getSpan().errorOccurred();
+                }
+                Tags.HTTP_RESPONSE_STATUS_CODE.set(cache.getSpan(), response.status().code());
+            }
+
+            return publisher;
+        };
+    }
+
+    @Override
+    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
+                              Object ret) {
+        Flux<?> responseFlux = (Flux<?>) ret;
+
+        responseFlux = responseFlux
+                .doOnError(e -> {
+                    EnhanceObjectCache cache = (EnhanceObjectCache) objInst.getSkyWalkingDynamicField();
+                    if (cache == null) {
+                        return;
+                    }
+
+                    if (cache.getSpan() != null) {
+                        cache.getSpan().errorOccurred();
+                        cache.getSpan().log(e);
+                    }
+                })
+                .doFinally(signalType -> {
+                    EnhanceObjectCache cache = (EnhanceObjectCache) objInst.getSkyWalkingDynamicField();
+                    if (cache == null) {
+                        return;
+                    }
+                    // do finally. Finish the span.
+                    if (cache.getSpan() != null) {
+                        if (signalType == SignalType.CANCEL) {
+                            cache.getSpan().errorOccurred();
+                        }
+                        cache.getSpan().asyncFinish();
+                    }
+
+                    if (cache.getSpan1() != null) {
+                        cache.getSpan1().asyncFinish();
+                    }
+                });
+
+        return responseFlux;
+    }
+
+    @Override
+    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+            Class<?>[] argumentsTypes, Throwable t) {
+
+    }
+}

apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v4x/HttpClientFinalizerSendInterceptor.java

@@ -0,0 +1,115 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.apache.skywalking.apm.plugin.spring.cloud.gateway.v4x;
+
+import org.apache.skywalking.apm.agent.core.context.CarrierItem;
+import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.tag.Tags;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+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.apache.skywalking.apm.plugin.spring.cloud.gateway.v4x.define.EnhanceObjectCache;
+import org.apache.skywalking.apm.util.StringUtil;
+import org.reactivestreams.Publisher;
+import reactor.netty.NettyOutbound;
+import reactor.netty.http.client.HttpClientRequest;
+
+import java.lang.reflect.Method;
+import java.net.URL;
+import java.util.function.BiFunction;
+
+import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.SPRING_CLOUD_GATEWAY;
+
+/**
+ * This class intercept <code>send</code> method.
+ * <p>
+ * In before method, create a new BiFunction lambda expression for setting <code>ContextCarrier</code> to http header
+ * and replace the original lambda in argument
+ */
+public class HttpClientFinalizerSendInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override
+    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
+            MethodInterceptResult result) throws Throwable {
+        EnhanceObjectCache enhanceObjectCache = (EnhanceObjectCache) objInst.getSkyWalkingDynamicField();
+        if (enhanceObjectCache == null) {
+            return;
+        }
+
+        /*
+          In this plug-in, the HttpClientFinalizerSendInterceptor depends on the NettyRoutingFilterInterceptor
+          When the NettyRoutingFilterInterceptor is not executed, the HttpClientFinalizerSendInterceptor has no meaning to be executed independently
+          and using ContextManager.activeSpan() method would cause NPE as active span does not exist.
+         */
+        if (!ContextManager.isActive()) {
+            return;
+        }
+
+        AbstractSpan span = ContextManager.activeSpan();
+        span.prepareForAsync();
+
+        if (StringUtil.isNotEmpty(enhanceObjectCache.getUrl())) {
+            URL url = new URL(enhanceObjectCache.getUrl());
+
+            ContextCarrier contextCarrier = new ContextCarrier();
+            AbstractSpan abstractSpan = ContextManager.createExitSpan(
+                    "SpringCloudGateway/sendRequest", contextCarrier, getPeer(url));
+            Tags.URL.set(abstractSpan, enhanceObjectCache.getUrl());
+            abstractSpan.prepareForAsync();
+            abstractSpan.setComponent(SPRING_CLOUD_GATEWAY);
+            abstractSpan.setLayer(SpanLayer.HTTP);
+            ContextManager.stopSpan(abstractSpan);
+
+            BiFunction<? super HttpClientRequest, ? super NettyOutbound, ? extends Publisher<Void>> finalSender = (BiFunction<? super HttpClientRequest, ? super NettyOutbound, ? extends Publisher<Void>>) allArguments[0];
+            allArguments[0] = (BiFunction<HttpClientRequest, NettyOutbound, Publisher<Void>>) (request, outbound) -> {
+                Publisher publisher = finalSender.apply(request, outbound);
+
+                CarrierItem next = contextCarrier.items();
+                while (next.hasNext()) {
+                    next = next.next();
+                    request.requestHeaders().remove(next.getHeadKey());
+                    request.requestHeaders().set(next.getHeadKey(), next.getHeadValue());
+                }
+                return publisher;
+            };
+            enhanceObjectCache.setCacheSpan(abstractSpan);
+        }
+        ContextManager.stopSpan(span);
+        enhanceObjectCache.setSpan1(span);
+    }
+
+    private String getPeer(URL url) {
+        return url.getHost() + ":" + url.getPort();
+    }
+
+    @Override
+    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
+            Object ret) {
+        ((EnhancedInstance) ret).setSkyWalkingDynamicField(objInst.getSkyWalkingDynamicField());
+        return ret;
+    }
+
+    @Override
+    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+            Class<?>[] argumentsTypes, Throwable t) {
+
+    }
+}