fix the response time of gRPC is not right. (#7179)

CHANGES.md

@@ -15,6 +15,7 @@ Release Notes.
 * Remove deprecated gRPC method, `io.grpc.ManagedChannelBuilder#nameResolverFactory`. See [gRPC-java 7133](https://github.com/grpc/grpc-java/issues/7133) for more details.
 * Add `Neo4j-4.x` plugin.
 * Correct `profile.duration` to `profile.max_duration` in the default `agent.config` file.
+* Fix the reponse time of gRPC.
 
 #### OAP-Backend
 * Disable Spring sleuth meter analyzer by default.

apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/server/ServerInterceptor.java

@@ -23,11 +23,6 @@ import io.grpc.ServerCall;
 import io.grpc.ServerCallHandler;
 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.trace.AbstractSpan;
-import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
-import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
-import org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil;
 import org.apache.skywalking.apm.util.StringUtil;
 
 public class ServerInterceptor implements io.grpc.ServerInterceptor {
@@ -43,15 +38,7 @@ public class ServerInterceptor implements io.grpc.ServerInterceptor {
                 next.setHeadValue(contextValue);
             }
         }
-
-        final AbstractSpan span = ContextManager.createEntrySpan(OperationNameFormatUtil.formatOperationName(call.getMethodDescriptor()), contextCarrier);
-        span.setComponent(ComponentsDefine.GRPC);
-        span.setLayer(SpanLayer.RPC_FRAMEWORK);
-        try {
-            return new TracingServerCallListener<>(handler.startCall(new TracingServerCall<>(call, ContextManager.capture()), headers), call
-                .getMethodDescriptor(), ContextManager.capture());
-        } finally {
-            ContextManager.stopSpan();
-        }
+        return new TracingServerCallListener<>(handler.startCall(new TracingServerCall<>(call), headers), call
+                .getMethodDescriptor(), contextCarrier);
     }
 }

apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/server/TracingServerCall.java

@@ -23,26 +23,22 @@ import io.grpc.Metadata;
 import io.grpc.ServerCall;
 import io.grpc.Status;
 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.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.network.trace.component.ComponentsDefine;
-import org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil;
-
 import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.RESPONSE_ON_CLOSE_OPERATION_NAME;
 import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.RESPONSE_ON_MESSAGE_OPERATION_NAME;
 import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.SERVER;
+import org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil;
 
 public class TracingServerCall<REQUEST, RESPONSE> extends ForwardingServerCall.SimpleForwardingServerCall<REQUEST, RESPONSE> {
 
     private final String operationPrefix;
-    private final ContextSnapshot contextSnapshot;
 
-    protected TracingServerCall(ServerCall<REQUEST, RESPONSE> delegate, ContextSnapshot contextSnapshot) {
+    protected TracingServerCall(ServerCall<REQUEST, RESPONSE> delegate) {
         super(delegate);
         this.operationPrefix = OperationNameFormatUtil.formatOperationName(delegate.getMethodDescriptor()) + SERVER;
-        this.contextSnapshot = contextSnapshot;
     }
 
     @Override
@@ -52,7 +48,6 @@ public class TracingServerCall<REQUEST, RESPONSE> extends ForwardingServerCall.S
             final AbstractSpan span = ContextManager.createLocalSpan(operationPrefix + RESPONSE_ON_MESSAGE_OPERATION_NAME);
             span.setComponent(ComponentsDefine.GRPC);
             span.setLayer(SpanLayer.RPC_FRAMEWORK);
-            ContextManager.continued(contextSnapshot);
 
             try {
                 super.sendMessage(message);
@@ -72,7 +67,6 @@ public class TracingServerCall<REQUEST, RESPONSE> extends ForwardingServerCall.S
         final AbstractSpan span = ContextManager.createLocalSpan(operationPrefix + RESPONSE_ON_CLOSE_OPERATION_NAME);
         span.setComponent(ComponentsDefine.GRPC);
         span.setLayer(SpanLayer.RPC_FRAMEWORK);
-        ContextManager.continued(contextSnapshot);
         switch (status.getCode()) {
             case OK:
                 break;
@@ -108,4 +102,5 @@ public class TracingServerCall<REQUEST, RESPONSE> extends ForwardingServerCall.S
             ContextManager.stopSpan();
         }
     }
+
 }

apm-sniffer/apm-sdk-plugin/grpc-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/grpc/v1/server/TracingServerCallListener.java

@@ -21,30 +21,33 @@ package org.apache.skywalking.apm.plugin.grpc.v1.server;
 import io.grpc.ForwardingServerCallListener;
 import io.grpc.MethodDescriptor;
 import io.grpc.ServerCall;
+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.ContextSnapshot;
 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.network.trace.component.ComponentsDefine;
-import org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil;
-
 import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.REQUEST_ON_CANCEL_OPERATION_NAME;
 import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.REQUEST_ON_COMPLETE_OPERATION_NAME;
 import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.REQUEST_ON_MESSAGE_OPERATION_NAME;
 import static org.apache.skywalking.apm.plugin.grpc.v1.Constants.SERVER;
+import org.apache.skywalking.apm.plugin.grpc.v1.OperationNameFormatUtil;
 
 public class TracingServerCallListener<REQUEST> extends ForwardingServerCallListener.SimpleForwardingServerCallListener<REQUEST> {
-
-    private final ContextSnapshot contextSnapshot;
     private final MethodDescriptor.MethodType methodType;
     private final String operationPrefix;
+    private final String operation;
+    private final ContextCarrier contextCarrier;
 
-    protected TracingServerCallListener(ServerCall.Listener<REQUEST> delegate, MethodDescriptor<REQUEST, ?> descriptor,
-        ContextSnapshot contextSnapshot) {
+    private AbstractSpan asyncSpan;
+    private ContextSnapshot contextSnapshot;
+
+    protected TracingServerCallListener(ServerCall.Listener<REQUEST> delegate, MethodDescriptor<REQUEST, ?> descriptor, ContextCarrier contextCarrier) {
         super(delegate);
-        this.contextSnapshot = contextSnapshot;
         this.methodType = descriptor.getType();
         this.operationPrefix = OperationNameFormatUtil.formatOperationName(descriptor) + SERVER;
+        this.operation = OperationNameFormatUtil.formatOperationName(descriptor);
+        this.contextCarrier = contextCarrier;
     }
 
     @Override
@@ -74,7 +77,6 @@ public class TracingServerCallListener<REQUEST> extends ForwardingServerCallList
         span.setComponent(ComponentsDefine.GRPC);
         span.setLayer(SpanLayer.RPC_FRAMEWORK);
         ContextManager.continued(contextSnapshot);
-
         try {
             super.onCancel();
         } catch (Throwable t) {
@@ -91,7 +93,6 @@ public class TracingServerCallListener<REQUEST> extends ForwardingServerCallList
         span.setComponent(ComponentsDefine.GRPC);
         span.setLayer(SpanLayer.RPC_FRAMEWORK);
         ContextManager.continued(contextSnapshot);
-
         try {
             super.onHalfClose();
         } catch (Throwable t) {
@@ -101,4 +102,22 @@ public class TracingServerCallListener<REQUEST> extends ForwardingServerCallList
             ContextManager.stopSpan();
         }
     }
-}
+
+    @Override
+    public void onComplete() {
+        super.onComplete();
+        asyncSpan.asyncFinish();
+    }
+
+    @Override
+    public void onReady() {
+        final AbstractSpan span = ContextManager.createEntrySpan(operation, contextCarrier);
+        span.setComponent(ComponentsDefine.GRPC);
+        span.setLayer(SpanLayer.RPC_FRAMEWORK);
+        contextSnapshot = ContextManager.capture();
+        asyncSpan = span.prepareForAsync();
+        ContextManager.stopSpan(asyncSpan);
+
+        super.onReady();
+    }
+}
\ No newline at end of file

test/plugin/scenarios/correlation-autotag-scenario/config/expectedData.yaml

@@ -150,11 +150,6 @@ segmentItems:
             spanType: Local
             peer: ''
             skipAnalysis: false
-            refs:
-              - {parentEndpoint: Greeter.sayHello, networkAddress: '', refType: CrossThread,
-                 parentSpanId: 0, parentTraceSegmentId: not null,
-                 parentServiceInstance: not null, parentService: correlation-autotag-scenario,
-                 traceId: not null}
           - operationName: Greeter.sayHello/server/Request/onMessage
             operationId: 0
             parentSpanId: -1
@@ -186,11 +181,6 @@ segmentItems:
             spanType: Local
             peer: ''
             skipAnalysis: false
-            refs:
-              - {parentEndpoint: Greeter.sayHello, networkAddress: '', refType: CrossThread,
-                 parentSpanId: 0, parentTraceSegmentId: not null,
-                 parentServiceInstance: not null, parentService: correlation-autotag-scenario,
-                 traceId: not null}
           - operationName: Greeter.sayHello/server/Request/onMessage
             operationId: 0
             parentSpanId: -1
@@ -224,11 +214,6 @@ segmentItems:
             skipAnalysis: false
             tags:
               - {key: status_code, value: OK}
-            refs:
-              - {parentEndpoint: Greeter.sayHello, networkAddress: '', refType: CrossThread,
-                 parentSpanId: 0, parentTraceSegmentId: not null,
-                 parentServiceInstance: not null, parentService: correlation-autotag-scenario,
-                 traceId: not null}
           - operationName: Greeter.sayHello/server/Request/onComplete
             operationId: 0
             parentSpanId: -1