Fix springmvc reactive api can't collect HTTP statusCode. (#6671)

CHANGES.md

@@ -29,6 +29,7 @@ Release Notes.
 * Fix bug that springmvn-annotation-4.x-plugin, witness class does not exist in some versions.
 * Add Redis command parameters to 'db.statement' field on Lettuce span UI for displaying more info
 * Fix NullPointerException with `ReactiveRequestHolder.getHeaders`.
+* Fix springmvc reactive api can't collect HTTP statusCode.
 * Fix bug that asynchttpclient plugin does not record the response status code
 
 #### OAP-Backend

apm-sniffer/apm-sdk-plugin/spring-plugins/mvc-annotation-5.x-plugin/pom.xml

@@ -33,6 +33,7 @@
         <spring-core.version>5.0.0.RELEASE</spring-core.version>
         <spring-webmvc.version>5.0.0.RELEASE</spring-webmvc.version>
         <javax-servlet-api.version>3.0.1</javax-servlet-api.version>
+        <spring-webflux.version>5.0.0.RELEASE</spring-webflux.version>
     </properties>
 
     <dependencies>
@@ -60,5 +61,11 @@
             <version>${project.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-webflux</artifactId>
+            <version>${spring-webflux.version}</version>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 </project>

apm-sniffer/apm-sdk-plugin/spring-plugins/mvc-annotation-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/mvc/v5/InvokeInterceptor.java

@@ -19,12 +19,16 @@ package org.apache.skywalking.apm.plugin.spring.mvc.v5;
 
 import java.lang.reflect.Method;
 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.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.mvc.commons.ReactiveRequestHolder;
 import org.apache.skywalking.apm.plugin.spring.mvc.commons.ReactiveResponseHolder;
+import org.springframework.http.HttpStatus;
 import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Mono;
 
 import static org.apache.skywalking.apm.plugin.spring.mvc.commons.Constants.REQUEST_KEY_IN_RUNTIME_CONTEXT;
 import static org.apache.skywalking.apm.plugin.spring.mvc.commons.Constants.RESPONSE_KEY_IN_RUNTIME_CONTEXT;
@@ -37,10 +41,12 @@ public class InvokeInterceptor implements InstanceMethodsAroundInterceptor {
                              final Class<?>[] argumentsTypes,
                              final MethodInterceptResult result) throws Throwable {
         ServerWebExchange exchange = (ServerWebExchange) allArguments[0];
+        final ReactiveResponseHolder responseHolder = new ReactiveResponseHolder(exchange.getResponse());
         ContextManager.getRuntimeContext()
-                      .put(RESPONSE_KEY_IN_RUNTIME_CONTEXT, new ReactiveResponseHolder(exchange.getResponse()));
+                .put(RESPONSE_KEY_IN_RUNTIME_CONTEXT, responseHolder);
         ContextManager.getRuntimeContext()
-                      .put(REQUEST_KEY_IN_RUNTIME_CONTEXT, new ReactiveRequestHolder(exchange.getRequest()));
+                .put(REQUEST_KEY_IN_RUNTIME_CONTEXT, new ReactiveRequestHolder(exchange.getRequest()));
+        objInst.setSkyWalkingDynamicField(responseHolder);
     }
 
     @Override
@@ -49,7 +55,20 @@ public class InvokeInterceptor implements InstanceMethodsAroundInterceptor {
                               final Object[] allArguments,
                               final Class<?>[] argumentsTypes,
                               final Object ret) throws Throwable {
-        return ret;
+        ServerWebExchange exchange = (ServerWebExchange) allArguments[0];
+        return ((Mono) ret).doFinally(s -> {
+            ReactiveResponseHolder responseHolder = (ReactiveResponseHolder) objInst.getSkyWalkingDynamicField();
+            AbstractSpan span = responseHolder.getSpan();
+            if (span == null) {
+                return;
+            }
+            HttpStatus httpStatus = exchange.getResponse().getStatusCode();
+            if (httpStatus != null && httpStatus.isError()) {
+                span.errorOccurred();
+                Tags.STATUS_CODE.set(span, Integer.toString(httpStatus.value()));
+            }
+            span.asyncFinish();
+        });
     }
 
     @Override

apm-sniffer/apm-sdk-plugin/spring-plugins/mvc-annotation-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/mvc/v5/define/reactive/InvocableHandlerMethodInstrumentation.java

@@ -45,7 +45,7 @@ public class InvocableHandlerMethodInstrumentation extends ClassInstanceMethodsE
             new InstanceMethodsInterceptPoint() {
                 @Override
                 public ElementMatcher<MethodDescription> getMethodsMatcher() {
-                    return named("getMethodArgumentValues").and(
+                    return named("invoke").and(
                         takesArgumentWithType(0, "org.springframework.web.server.ServerWebExchange"));
                 }
 

apm-sniffer/apm-sdk-plugin/spring-plugins/mvc-annotation-commons/src/main/java/org/apache/skywalking/apm/plugin/spring/mvc/commons/ReactiveResponseHolder.java

@@ -17,16 +17,27 @@
 
 package org.apache.skywalking.apm.plugin.spring.mvc.commons;
 
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
 import org.springframework.http.server.reactive.ServerHttpResponse;
 
 public class ReactiveResponseHolder implements ResponseHolder {
 
     private final ServerHttpResponse response;
 
+    private AbstractSpan span;
+
     public ReactiveResponseHolder(final ServerHttpResponse response) {
         this.response = response;
     }
 
+    public void setSpan(AbstractSpan span) {
+        this.span = span;
+    }
+
+    public AbstractSpan getSpan() {
+        return this.span;
+    }
+
     @Override
     public int statusCode() {
         return response.getStatusCode().value();

apm-sniffer/apm-sdk-plugin/spring-plugins/mvc-annotation-commons/src/main/java/org/apache/skywalking/apm/plugin/spring/mvc/commons/interceptor/AbstractMethodInterceptor.java

@@ -40,6 +40,7 @@ import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.spring.mvc.commons.EnhanceRequireObjectCache;
 import org.apache.skywalking.apm.plugin.spring.mvc.commons.RequestHolder;
 import org.apache.skywalking.apm.plugin.spring.mvc.commons.ResponseHolder;
+import org.apache.skywalking.apm.plugin.spring.mvc.commons.ReactiveResponseHolder;
 import org.apache.skywalking.apm.plugin.spring.mvc.commons.SpringMVCPluginConfig;
 import org.apache.skywalking.apm.plugin.spring.mvc.commons.exception.IllegalMethodStackDepthException;
 import org.apache.skywalking.apm.plugin.spring.mvc.commons.exception.ServletResponseNotFoundException;
@@ -185,7 +186,11 @@ public abstract class AbstractMethodInterceptor implements InstanceMethodsAround
                     span.errorOccurred();
                     Tags.STATUS_CODE.set(span, Integer.toString(response.statusCode()));
                 }
-
+                if (response instanceof ReactiveResponseHolder) {
+                    ReactiveResponseHolder reactiveResponse = (ReactiveResponseHolder) response;
+                    AbstractSpan async = span.prepareForAsync();
+                    reactiveResponse.setSpan(async);
+                }
                 ContextManager.getRuntimeContext().remove(REQUEST_KEY_IN_RUNTIME_CONTEXT);
                 ContextManager.getRuntimeContext().remove(RESPONSE_KEY_IN_RUNTIME_CONTEXT);
                 ContextManager.getRuntimeContext().remove(CONTROLLER_METHOD_STACK_DEPTH);

test/plugin/scenarios/springmvc-reactive-scenario/config/expectedData.yaml

@@ -16,7 +16,7 @@
 
 segmentItems:
 - serviceName: springmvc-reactive-scenario
-  segmentSize: nq 0
+  segmentSize: ge 2
   segments:
   - segmentId: not null
     spans:
@@ -36,6 +36,35 @@ segmentItems:
       - {key: db.instance, value: test}
       - {key: db.statement, value: not null}
       skipAnalysis: 'false'
+    - operationName: '/testcase/error'
+      operationId: 0
+      parentSpanId: 0
+      spanId: 2
+      spanLayer: Http
+      startTime: not null
+      endTime: not null
+      isError: false
+      componentId: not null
+      tags:
+      - {key: url, value: not null}
+      - {key: http.method, value: GET}
+      skipAnalysis: 'false'
+    - operationName: 'future/get:/testcase/error'
+      operationId: 0
+      parentSpanId: 0
+      spanId: 3
+      spanLayer: not null
+      startTime: not null
+      endTime: not null
+      isError: true
+      componentId: not null
+      logs:
+      - logEvent:
+        - {key: event, value: error}
+        - {key: error.kind, value: not null}
+        - {key: message, value: not null}
+        - {key: stack, value: not null}
+      skipAnalysis: 'false'
     - operationName: '{GET}/testcase/{test}'
       operationId: 0
       parentSpanId: -1
@@ -44,7 +73,31 @@ segmentItems:
       startTime: not null
       endTime: not null
       componentId: 14
+      isError: false
+      tags:
+      - {key: url, value: not null}
+      - {key: http.method, value: GET}
+      skipAnalysis: 'false'
+  - segmentId: not null
+    spans:
+    - operationName: '{GET}/testcase/error'
+      operationId: 0
+      parentSpanId: -1
+      spanId: 0
+      spanLayer: Http
+      startTime: not null
+      endTime: not null
+      componentId: 14
+      spanType: Entry
+      isError: true
       tags:
       - {key: url, value: not null}
       - {key: http.method, value: GET}
+      - {key: status_code, value: '500'}
+      logs:
+      - logEvent:
+        - {key: event, value: error}
+        - {key: error.kind, value: not null}
+        - {key: message, value: not null}
+        - {key: stack, value: not null}
       skipAnalysis: 'false'
\ No newline at end of file

test/plugin/scenarios/springmvc-reactive-scenario/src/main/java/test/apache/skywalking/apm/testcase/sc/springmvcreactive/controller/Controller.java

@@ -19,11 +19,14 @@ package test.apache.skywalking.apm.testcase.sc.springmvcreactive.controller;
 
 import java.sql.SQLException;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
+import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.client.AsyncRestTemplate;
 import reactor.core.publisher.Mono;
 import test.apache.skywalking.apm.testcase.sc.springmvcreactive.service.TestService;
 
@@ -41,6 +44,17 @@ public class Controller {
     @GetMapping("/testcase/{test}")
     public Mono<String> hello(@RequestBody(required = false) String body, @PathVariable("test") String test) throws SQLException {
         testService.executeSQL();
+        ListenableFuture<ResponseEntity<String>> forEntity = new AsyncRestTemplate().getForEntity("http://localhost:8080/testcase/error", String.class);
+        try {
+            forEntity.get();
+        } catch (Exception e) {
+        }
         return Mono.just("Hello World");
     }
+
+    @GetMapping("/testcase/error")
+    public Mono<String> error() {
+        throw new RuntimeException("this is Error");
+    }
+
 }