Fix Span not finished in gateway plugin when the gateway request timeout.

Author: wuwen5

CHANGES.md

@@ -16,6 +16,7 @@ Release Notes.
 * Grpc plugin support trace client async generic call(without grpc stubs), support Method type: `UNARY`、`SERVER_STREAMING`.
 * Enhance Apache ShenYu (incubating) plugin: support trace `grpc`,`sofarpc`,`motan`,`tars` rpc proxy.
 * Add primary endpoint name to log events.
+* Fix Span not finished in gateway plugin when the gateway request timeout.
 
 #### 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/HttpClientFinalizerResponseConnectionInterceptor.java

@@ -24,6 +24,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.plugin.spring.cloud.gateway.v3x.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;
 
@@ -39,7 +41,7 @@ public class HttpClientFinalizerResponseConnectionInterceptor implements Instanc
 
     @Override
     public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
-            MethodInterceptResult result) throws Throwable {
+            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) -> {
@@ -53,20 +55,48 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
                     cache.getSpan().errorOccurred();
                 }
                 Tags.HTTP_RESPONSE_STATUS_CODE.set(cache.getSpan(), response.status().code());
-                cache.getSpan().asyncFinish();
-            }
-
-            if (cache.getSpan1() != null) {
-                cache.getSpan1().asyncFinish();
             }
+            
             return publisher;
         };
     }
 
     @Override
     public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
-            Object ret) throws Throwable {
-        return ret;
+                              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;
+                    }
+                    // receive the response. Stop 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

apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-3.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v3x/HttpClientFinalizerInterceptorTest.java

@@ -40,6 +40,7 @@
 import org.powermock.modules.junit4.PowerMockRunner;
 import org.powermock.modules.junit4.PowerMockRunnerDelegate;
 import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 import reactor.netty.Connection;
 import reactor.netty.NettyOutbound;
 import reactor.netty.http.client.HttpClientRequest;
@@ -151,9 +152,13 @@ private void executeSendRequest() throws Throwable {
         Object[] responseConnectionArguments = new Object[]{originalResponseConnectionBiFunction};
         responseConnectionInterceptor
                 .beforeMethod(enhancedInstance, null, responseConnectionArguments, null, null);
-        responseConnectionInterceptor.afterMethod(enhancedInstance, null, new Object[0], null, enhancedInstance);
+        Flux flux = Flux.just(0);
+
+        flux = (Flux) responseConnectionInterceptor.afterMethod(enhancedInstance, null, new Object[0], null, flux);
+        
         ((BiFunction<? super HttpClientResponse, ? super Connection, ? extends Publisher<Void>>) responseConnectionArguments[0])
                 .apply(mockResponse, null);
+        flux.blockFirst();
     }
 
     private void assertUpstreamSpan(AbstractSpan span) {

test/plugin/scenarios/gateway-3.x-scenario/gateway-projectA-scenario/src/main/resources/application.yml

@@ -20,8 +20,14 @@ server:
 spring:
   cloud:
     gateway:
+      httpclient:
+        connect-timeout: 2000
       routes:
       - id: provider_route
         uri: http://localhost:18070
         predicates:
         - Path=/provider/b/*
+      - id: provider_timeout
+        uri: http://1.2.3.4:18070
+        predicates:
+        - Path=/provider/timeout/*