sofastack · EvenLjj · Mar 25, 2024 · Feb 27, 2024
diff --git a/.../bootstrap-api/src/main/java/com/alipay/sofa/rpc/bootstrap/DefaultClientProxyInvoker.java b/.../bootstrap-api/src/main/java/com/alipay/sofa/rpc/bootstrap/DefaultClientProxyInvoker.java
@@ -133,6 +133,7 @@ protected void decorateRequest(SofaRequest request) {
         // 额外属性通过HEAD传递给服务端
         request.addRequestProp(RemotingConstants.HEAD_APP_NAME, consumerConfig.getAppName());
         request.addRequestProp(RemotingConstants.HEAD_PROTOCOL, consumerConfig.getProtocol());
+        request.addRequestProp(RemotingConstants.HEAD_INVOKE_TYPE, request.getInvokeType());
 
         customRequest(request, internalContext);
 

diff --git a/...opentracing/src/main/java/com/alipay/sofa/rpc/filter/sofatracer/ProviderTracerFilter.java b/...opentracing/src/main/java/com/alipay/sofa/rpc/filter/sofatracer/ProviderTracerFilter.java
@@ -19,9 +19,7 @@
 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.common.RpcConstants;
 import com.alipay.sofa.rpc.config.ProviderConfig;
-import com.alipay.sofa.rpc.context.RpcInternalContext;
 import com.alipay.sofa.rpc.core.exception.SofaRpcException;
 import com.alipay.sofa.rpc.core.request.SofaRequest;
 import com.alipay.sofa.rpc.core.response.SofaResponse;
@@ -32,10 +30,6 @@
 import com.alipay.sofa.rpc.module.SofaTracerModule;
 import com.alipay.sofa.rpc.tracer.sofatracer.log.tags.RpcSpanTags;
 
-import static com.alipay.sofa.rpc.common.RemotingConstants.HEAD_APP_NAME;
-import static com.alipay.sofa.rpc.common.RemotingConstants.HEAD_INVOKE_TYPE;
-import static com.alipay.sofa.rpc.common.RemotingConstants.HEAD_PROTOCOL;
-
 /**
  * @author <a href="mailto:zhanggeng.zg@antfin.com">zhanggeng</a>
  */
@@ -50,34 +44,14 @@ public boolean needToLoad(FilterInvoker invoker) {
 
     @Override
     public SofaResponse invoke(FilterInvoker invoker, SofaRequest request) throws SofaRpcException {
-        SofaTracerSpan serverSpan = null;
-        try {
-            SofaTraceContext sofaTraceContext = SofaTraceContextHolder.getSofaTraceContext();
-            serverSpan = sofaTraceContext.getCurrentSpan();
-            if (serverSpan != null) {
-                RpcInternalContext context = RpcInternalContext.getContext();
-                serverSpan.setTag(RpcSpanTags.SERVICE, request.getTargetServiceUniqueName());
-                serverSpan.setTag(RpcSpanTags.METHOD, request.getMethodName());
-                serverSpan.setTag(RpcSpanTags.REMOTE_IP, context.getRemoteHostName()); // 客户端地址
-
-                // 从请求里获取ConsumerTracerFilter额外传递的信息
-                serverSpan.setTag(RpcSpanTags.REMOTE_APP, (String) request.getRequestProp(HEAD_APP_NAME));
-                serverSpan.setTag(RpcSpanTags.PROTOCOL, (String) request.getRequestProp(HEAD_PROTOCOL));
-                serverSpan.setTag(RpcSpanTags.INVOKE_TYPE, (String) request.getRequestProp(HEAD_INVOKE_TYPE));
-
-                ProviderConfig providerConfig = (ProviderConfig) invoker.getConfig();
-                serverSpan.setTag(RpcSpanTags.LOCAL_APP, providerConfig.getAppName());
-
-                serverSpan.setTag(RpcSpanTags.SERVER_THREAD_POOL_WAIT_TIME,
-                    (Number) context.getAttachment(RpcConstants.INTERNAL_KEY_PROCESS_WAIT_TIME));
-            }
-            return invoker.invoke(request);
-        } finally {
-            if (serverSpan != null) {
-                serverSpan.setTag(RpcSpanTags.SERVER_BIZ_TIME,
-                    (Number) RpcInternalContext.getContext().getAttachment(RpcConstants.INTERNAL_KEY_IMPL_ELAPSE));
-            }
+        SofaTraceContext sofaTraceContext = SofaTraceContextHolder.getSofaTraceContext();
+        SofaTracerSpan serverSpan = sofaTraceContext.getCurrentSpan();
+        if (serverSpan != null) {
+            ProviderConfig providerConfig = (ProviderConfig) invoker.getConfig();
+            serverSpan.setTag(RpcSpanTags.LOCAL_APP, providerConfig.getAppName());
         }
+        return invoker.invoke(request);
+
     }
 
 }
diff --git a/...tracer-opentracing/src/main/java/com/alipay/sofa/rpc/tracer/sofatracer/RpcSofaTracer.java b/...tracer-opentracing/src/main/java/com/alipay/sofa/rpc/tracer/sofatracer/RpcSofaTracer.java
@@ -61,6 +61,10 @@
 import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
 
+import static com.alipay.sofa.rpc.common.RemotingConstants.HEAD_APP_NAME;
+import static com.alipay.sofa.rpc.common.RemotingConstants.HEAD_INVOKE_TYPE;
+import static com.alipay.sofa.rpc.common.RemotingConstants.HEAD_PROTOCOL;
+
 /**
  * SofaTracer
  *
@@ -490,11 +494,19 @@ public void serverReceived(SofaRequest request) {
 
         // Record server receive event
         serverSpan.log(LogData.SERVER_RECV_EVENT_VALUE);
+        // Retrieve logging context information from the request
+        RpcInternalContext context = RpcInternalContext.getContext();
+        serverSpan.setTag(RpcSpanTags.LOCAL_APP, request.getTargetAppName());
+        serverSpan.setTag(RpcSpanTags.SERVICE, request.getTargetServiceUniqueName());
+        serverSpan.setTag(RpcSpanTags.METHOD, request.getMethodName());
+        serverSpan.setTag(RpcSpanTags.PROTOCOL, (String) request.getRequestProp(HEAD_PROTOCOL));
+        serverSpan.setTag(RpcSpanTags.INVOKE_TYPE, (String) request.getRequestProp(HEAD_INVOKE_TYPE));
+        serverSpan.setTag(RpcSpanTags.REMOTE_IP, context.getRemoteHostName());
+        serverSpan.setTag(RpcSpanTags.REMOTE_APP, (String) request.getRequestProp(HEAD_APP_NAME));
         //放到线程上下文
         sofaTraceContext.push(serverSpan);
         //rpc 上下文
         if (RpcInternalContext.isAttachmentEnable()) {
-            RpcInternalContext context = RpcInternalContext.getContext();
             context.setAttachment(RpcConstants.INTERNAL_KEY_TRACE_ID, spanContext.getTraceId());
             context.setAttachment(RpcConstants.INTERNAL_KEY_SPAN_ID, spanContext.getSpanId());
         }
@@ -564,6 +576,10 @@ public void serverSend(SofaRequest request, SofaResponse response, Throwable exc
 
         RpcInternalContext context = RpcInternalContext.getContext();
         RpcInvokeContext invokeContext = RpcInvokeContext.getContext();
+        serverSpan.setTag(RpcSpanTags.SERVER_THREAD_POOL_WAIT_TIME,
+            (Number) context.getAttachment(RpcConstants.INTERNAL_KEY_PROCESS_WAIT_TIME));
+        serverSpan.setTag(RpcSpanTags.SERVER_BIZ_TIME,
+            (Number) RpcInternalContext.getContext().getAttachment(RpcConstants.INTERNAL_KEY_IMPL_ELAPSE));
         serverSpan.setTag(RpcSpanTags.RESP_SERIALIZE_TIME,
             (Number) context.getAttachment(RpcConstants.INTERNAL_KEY_RESP_SERIALIZE_TIME));
         serverSpan.setTag(RpcSpanTags.REQ_DESERIALIZE_TIME,

diff --git a/.../tracer-opentracing/src/test/java/com/alipay/sofa/rpc/event/SofaTracerSubscriberTest.java b/.../tracer-opentracing/src/test/java/com/alipay/sofa/rpc/event/SofaTracerSubscriberTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.event;
+
+import com.alipay.common.tracer.core.holder.SofaTraceContextHolder;
+import com.alipay.common.tracer.core.span.SofaTracerSpan;
+import com.alipay.sofa.rpc.common.RemotingConstants;
+import com.alipay.sofa.rpc.common.RpcOptions;
+import com.alipay.sofa.rpc.context.RpcInternalContext;
+import com.alipay.sofa.rpc.core.request.SofaRequest;
+import com.alipay.sofa.rpc.tracer.sofatracer.log.tags.RpcSpanTags;
+import io.opentracing.tag.Tags;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.Map;
+
+import static com.alipay.sofa.rpc.common.RemotingConstants.HEAD_APP_NAME;
+import static com.alipay.sofa.rpc.common.RemotingConstants.HEAD_INVOKE_TYPE;
+import static com.alipay.sofa.rpc.common.RemotingConstants.HEAD_PROTOCOL;
+
+/**
+ * @author Even
+ * @date 2024/2/27 19:48
+ */
+public class SofaTracerSubscriberTest {
+
+    @BeforeClass
+    public static void beforeClass() {
+        System.getProperties().put(RpcOptions.DEFAULT_TRACER, "sofaTracer");
+    }
+
+    @Test
+    public void testClientSendAndServerReceiveTracerEvent() {
+        SofaRequest sofaRequest = new SofaRequest();
+        sofaRequest.setMethodName("testService");
+        sofaRequest.setTimeout(1000);
+        sofaRequest.setInvokeType("sync");
+        sofaRequest.setTargetServiceUniqueName("testInterface:1.0");
+        sofaRequest.setTargetAppName("targetAppName");
+        EventBus.post(new ClientStartInvokeEvent(sofaRequest));
+        SofaTracerSpan currentClientSpan = SofaTraceContextHolder.getSofaTraceContext().getCurrentSpan();
+        Map<String, String> clientTagsWithStr = currentClientSpan.getTagsWithStr();
+        Assert.assertEquals("client", clientTagsWithStr.get(Tags.SPAN_KIND.getKey()));
+        Assert.assertEquals(sofaRequest.getTargetServiceUniqueName(), clientTagsWithStr.get(RpcSpanTags.SERVICE));
+        Assert.assertEquals(sofaRequest.getMethodName(), clientTagsWithStr.get(RpcSpanTags.METHOD));
+        Assert.assertEquals(Thread.currentThread().getName(), clientTagsWithStr.get(RpcSpanTags.CURRENT_THREAD_NAME));
+
+        Assert.assertNull(sofaRequest.getRequestProps());
+        EventBus.post(new ClientBeforeSendEvent(sofaRequest));
+        Map traceContext = (Map) sofaRequest.getRequestProps().get(RemotingConstants.RPC_TRACE_NAME);
+        Assert.assertNotNull(traceContext);
+
+        sofaRequest.getRequestProps().put(HEAD_PROTOCOL, "tr");
+        sofaRequest.getRequestProps().put(HEAD_INVOKE_TYPE, sofaRequest.getInvokeType());
+        sofaRequest.getRequestProps().put(HEAD_APP_NAME, "callerAppName");
+        RpcInternalContext.getContext().setRemoteAddress("127.0.0.1", 12200);
+        EventBus.post(new ServerReceiveEvent(sofaRequest));
+        SofaTracerSpan currentServerSpan = SofaTraceContextHolder.getSofaTraceContext().getCurrentSpan();
+        Map<String, String> serverTagsWithStr = currentServerSpan.getTagsWithStr();
+        Assert.assertEquals("server", serverTagsWithStr.get(Tags.SPAN_KIND.getKey()));
+        Assert.assertEquals(sofaRequest.getTargetServiceUniqueName(), serverTagsWithStr.get(RpcSpanTags.SERVICE));
+        Assert.assertEquals(sofaRequest.getMethodName(), serverTagsWithStr.get(RpcSpanTags.METHOD));
+        Assert.assertEquals(sofaRequest.getTargetAppName(), serverTagsWithStr.get(RpcSpanTags.LOCAL_APP));
+        Assert.assertEquals(sofaRequest.getInvokeType(), serverTagsWithStr.get(RpcSpanTags.INVOKE_TYPE));
+        Assert.assertEquals("tr", serverTagsWithStr.get(RpcSpanTags.PROTOCOL));
+        Assert.assertEquals("127.0.0.1", serverTagsWithStr.get(RpcSpanTags.REMOTE_IP));
+        Assert.assertEquals("callerAppName", serverTagsWithStr.get(RpcSpanTags.REMOTE_APP));
+        SofaTraceContextHolder.getSofaTraceContext().clear();
+    }
+
+    @AfterClass
+    public static void afterClass() {
+        System.getProperties().remove(RpcOptions.DEFAULT_TRACER);
+    }
+
+}