enhance triple stream tracelog

core/api/src/main/java/com/alipay/sofa/rpc/common/RpcConstants.java

@@ -128,15 +128,15 @@ public class RpcConstants {
     /**
      * 调用方式：客户端流
      */
-    public static final String  INVOKER_TYPE_CLIENT_STREAMING                = "clientStream";
+    public static final String  INVOKER_TYPE_CLIENT_STREAMING                = "client_stream";
     /**
      * 调用方式：服务端流
      */
-    public static final String  INVOKER_TYPE_SERVER_STREAMING                = "serverStream";
+    public static final String  INVOKER_TYPE_SERVER_STREAMING                = "server_stream";
     /**
      * 调用方式：双向流
      */
-    public static final String  INVOKER_TYPE_BI_STREAMING                    = "bidirectionalStream";
+    public static final String  INVOKER_TYPE_BI_STREAMING                    = "bi_stream";
 
     /**
      * Hessian序列化 [不推荐]

core/api/src/main/java/com/alipay/sofa/rpc/core/request/SofaRequest.java

@@ -306,7 +306,10 @@ public SofaRequest setData(AbstractByteBuf data) {
      * @return 如果是Future和Callback，是异步请求
      */
     public boolean isAsync() {
-        return invokeType != null && (RpcConstants.INVOKER_TYPE_CALLBACK.equals(invokeType)
-            || RpcConstants.INVOKER_TYPE_FUTURE.equals(invokeType));
+        return RpcConstants.INVOKER_TYPE_CALLBACK.equals(invokeType)
+            || RpcConstants.INVOKER_TYPE_FUTURE.equals(invokeType)
+            || RpcConstants.INVOKER_TYPE_BI_STREAMING.equals(invokeType)
+            || RpcConstants.INVOKER_TYPE_SERVER_STREAMING.equals(invokeType)
+            || RpcConstants.INVOKER_TYPE_CLIENT_STREAMING.equals(invokeType);
     }
 }

core/api/src/test/java/com/alipay/sofa/rpc/config/ConsumerConfigTest.java

@@ -1,3 +1,19 @@
+/*
+ * 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 com.alipay.sofa.rpc.config;
 
 import com.alipay.sofa.rpc.invoke.Invoker;
@@ -37,5 +53,4 @@ public void testMethodTimeout() {
         consumerConfig.getConfigValueCache(true);
         Assert.assertEquals(-1, consumerConfig.getMethodTimeout("invoke"));
     }
-
 }

remoting/pom.xml

@@ -88,6 +88,11 @@
                 <artifactId>sofa-rpc-log-common-tools</artifactId>
                 <version>${project.parent.version}</version>
             </dependency>
+            <dependency>
+                <groupId>com.alipay.sofa</groupId>
+                <artifactId>sofa-rpc-tracer-opentracing</artifactId>
+                <version>${project.parent.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 

remoting/remoting-triple/pom.xml

@@ -66,5 +66,9 @@
             <groupId>com.alipay.sofa</groupId>
             <artifactId>tracer-core</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.alipay.sofa</groupId>
+            <artifactId>sofa-rpc-tracer-opentracing</artifactId>
+        </dependency>
     </dependencies>
 </project>

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/interceptor/ClientHeaderClientInterceptor.java

@@ -16,12 +16,20 @@
  */
 package com.alipay.sofa.rpc.interceptor;
 
+import com.alipay.common.tracer.core.context.trace.SofaTraceContext;
+import com.alipay.common.tracer.core.holder.SofaTraceContextHolder;
+import com.alipay.common.tracer.core.span.SofaTracerSpan;
 import com.alipay.sofa.rpc.config.ConsumerConfig;
+import com.alipay.sofa.rpc.context.RpcInternalContext;
 import com.alipay.sofa.rpc.context.RpcInvokeContext;
 import com.alipay.sofa.rpc.context.RpcRunningState;
 import com.alipay.sofa.rpc.core.request.SofaRequest;
+import com.alipay.sofa.rpc.core.response.SofaResponse;
+import com.alipay.sofa.rpc.event.ClientAsyncReceiveEvent;
+import com.alipay.sofa.rpc.event.EventBus;
 import com.alipay.sofa.rpc.server.triple.TripleContants;
 import com.alipay.sofa.rpc.tracer.sofatracer.TripleTracerAdapter;
+import com.alipay.sofa.rpc.utils.TripleExceptionUtils;
 import io.grpc.CallOptions;
 import io.grpc.Channel;
 import io.grpc.ClientCall;
@@ -59,12 +67,15 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT
 
             @Override
             public void start(Listener<RespT> responseListener, Metadata requestHeader) {
-
+                RpcInternalContext internalContext = RpcInternalContext.getContext();
                 RpcInvokeContext context = RpcInvokeContext.getContext();
                 SofaRequest sofaRequest = (SofaRequest) context.get(TripleContants.SOFA_REQUEST_KEY);
 
-                ConsumerConfig consumerConfig = (ConsumerConfig) context.get(TripleContants.SOFA_CONSUMER_CONFIG_KEY);
-                TripleTracerAdapter.beforeSend(sofaRequest, consumerConfig, requestHeader);
+                ConsumerConfig<?> consumerConfig = (ConsumerConfig<?>) context
+                    .get(TripleContants.SOFA_CONSUMER_CONFIG_KEY);
+                TripleTracerAdapter.beforeSend(sofaRequest, consumerConfig, requestHeader, method);
+                SofaTraceContext sofaTraceContext = SofaTraceContextHolder.getSofaTraceContext();
+                SofaTracerSpan clientSpan = sofaTraceContext.getCurrentSpan();
                 if (RpcRunningState.isDebugMode()) {
                     LOGGER.info("[2]prepare to send from client:{}", requestHeader);
                 }
@@ -80,18 +91,48 @@ public void onHeaders(Metadata responseHeader) {
 
                     @Override
                     public void onMessage(RespT message) {
-                        if (RpcRunningState.isDebugMode()) {
-                            LOGGER.info("[4]response message received from server:{}", message);
+                        // onMessage -> onNext()
+                        try {
+                            if (sofaRequest.isAsync()) {
+                                RpcInvokeContext.setContext(context);
+                                sofaTraceContext.push(clientSpan);
+                            }
+                            if (RpcRunningState.isDebugMode()) {
+                                LOGGER.info("[4]response message received from server:{}", message);
+                            }
+                            super.onMessage(message);
+                        } finally {
+                            if (sofaRequest.isAsync()) {
+                                sofaTraceContext.clear();
+                                RpcInvokeContext.removeContext();
+                            }
                         }
-                        super.onMessage(message);
                     }
 
                     @Override
                     public void onClose(Status status, Metadata trailers) {
-                        if (RpcRunningState.isDebugMode()) {
-                            LOGGER.info("[5]response close received from server:{},trailers:{}", status, trailers);
+                        // onClose -> onComplete() or onError()
+                        try {
+                            if (sofaRequest.isAsync()) {
+                                RpcInvokeContext.setContext(context);
+                                sofaTraceContext.push(clientSpan);
+                            }
+                            if (RpcRunningState.isDebugMode()) {
+                                LOGGER.info("[5]response close received from server:{},trailers:{}", status, trailers);
+                            }
+                            super.onClose(status, trailers);
+                        } finally {
+                            if (sofaRequest.isAsync()) {
+                                Throwable throwable = TripleExceptionUtils.getThrowableFromStatus(status);
+                                RpcInternalContext.setContext(internalContext);
+                                if (EventBus.isEnable(ClientAsyncReceiveEvent.class)) {
+                                    EventBus.post(new ClientAsyncReceiveEvent(consumerConfig, null,
+                                        sofaRequest, new SofaResponse(), throwable));
+                                }
+                                RpcInvokeContext.removeContext();
+                                RpcInternalContext.removeAllContext();
+                            }
                         }
-                        super.onClose(status, trailers);
                     }
 
                     @Override
@@ -104,6 +145,15 @@ public void onReady() {
                 }, requestHeader);
             }
 
+            @Override
+            public void sendMessage(ReqT message) {
+                try {
+                    super.sendMessage(message);
+                } catch (Throwable t) {
+                    LOGGER.error("Client invoke grpc sendMessage meet error:", t);
+                    throw t;
+                }
+            }
         };
     }
 }