Skip to content

Commit 4db5636

Browse files
author
liujianjun.ljj
committed
enhance triple tracelog in server
1 parent e1ddc67 commit 4db5636

File tree

9 files changed

+300
-103
lines changed

9 files changed

+300
-103
lines changed

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

+3-3
Original file line numberDiff line numberDiff line change
@@ -128,15 +128,15 @@ public class RpcConstants {
128128
/**
129129
* 调用方式：客户端流
130130
*/
131-
public static final String INVOKER_TYPE_CLIENT_STREAMING = "clientStream";
131+
public static final String INVOKER_TYPE_CLIENT_STREAMING = "client_stream";
132132
/**
133133
* 调用方式：服务端流
134134
*/
135-
public static final String INVOKER_TYPE_SERVER_STREAMING = "serverStream";
135+
public static final String INVOKER_TYPE_SERVER_STREAMING = "server_stream";
136136
/**
137137
* 调用方式：双向流
138138
*/
139-
public static final String INVOKER_TYPE_BI_STREAMING = "bidirectionalStream";
139+
public static final String INVOKER_TYPE_BI_STREAMING = "bi_stream";
140140

141141
/**
142142
* Hessian序列化 [不推荐]

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

+5-2
Original file line numberDiff line numberDiff line change
@@ -306,7 +306,10 @@ public SofaRequest setData(AbstractByteBuf data) {
306306
* @return 如果是Future和Callback，是异步请求
307307
*/
308308
public boolean isAsync() {
309-
return invokeType != null && (RpcConstants.INVOKER_TYPE_CALLBACK.equals(invokeType)
310-
|| RpcConstants.INVOKER_TYPE_FUTURE.equals(invokeType));
309+
return RpcConstants.INVOKER_TYPE_CALLBACK.equals(invokeType)
310+
|| RpcConstants.INVOKER_TYPE_FUTURE.equals(invokeType)
311+
|| RpcConstants.INVOKER_TYPE_BI_STREAMING.equals(invokeType)
312+
|| RpcConstants.INVOKER_TYPE_SERVER_STREAMING.equals(invokeType)
313+
|| RpcConstants.INVOKER_TYPE_CLIENT_STREAMING.equals(invokeType);
311314
}
312315
}

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

+16-1
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,19 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
117
package com.alipay.sofa.rpc.config;
218

319
import com.alipay.sofa.rpc.invoke.Invoker;
@@ -37,5 +53,4 @@ public void testMethodTimeout() {
3753
consumerConfig.getConfigValueCache(true);
3854
Assert.assertEquals(-1, consumerConfig.getMethodTimeout("invoke"));
3955
}
40-
4156
}

‎remoting/pom.xml

+5
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,11 @@
8888
<artifactId>sofa-rpc-log-common-tools</artifactId>
8989
<version>${project.parent.version}</version>
9090
</dependency>
91+
<dependency>
92+
<groupId>com.alipay.sofa</groupId>
93+
<artifactId>sofa-rpc-tracer-opentracing</artifactId>
94+
<version>${project.parent.version}</version>
95+
</dependency>
9196
</dependencies>
9297
</dependencyManagement>
9398

‎remoting/remoting-triple/pom.xml

+4
Original file line numberDiff line numberDiff line change
@@ -66,5 +66,9 @@
6666
<groupId>com.alipay.sofa</groupId>
6767
<artifactId>tracer-core</artifactId>
6868
</dependency>
69+
<dependency>
70+
<groupId>com.alipay.sofa</groupId>
71+
<artifactId>sofa-rpc-tracer-opentracing</artifactId>
72+
</dependency>
6973
</dependencies>
7074
</project>

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

+74-9
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,21 @@
1616
*/
1717
package com.alipay.sofa.rpc.interceptor;
1818

19+
import com.alipay.common.tracer.core.context.trace.SofaTraceContext;
20+
import com.alipay.common.tracer.core.holder.SofaTraceContextHolder;
21+
import com.alipay.common.tracer.core.span.SofaTracerSpan;
1922
import com.alipay.sofa.rpc.config.ConsumerConfig;
23+
import com.alipay.sofa.rpc.context.RpcInternalContext;
2024
import com.alipay.sofa.rpc.context.RpcInvokeContext;
2125
import com.alipay.sofa.rpc.context.RpcRunningState;
26+
import com.alipay.sofa.rpc.core.exception.RpcErrorType;
27+
import com.alipay.sofa.rpc.core.exception.SofaRouteException;
28+
import com.alipay.sofa.rpc.core.exception.SofaRpcException;
29+
import com.alipay.sofa.rpc.core.exception.SofaTimeOutException;
2230
import com.alipay.sofa.rpc.core.request.SofaRequest;
31+
import com.alipay.sofa.rpc.core.response.SofaResponse;
32+
import com.alipay.sofa.rpc.event.ClientAsyncReceiveEvent;
33+
import com.alipay.sofa.rpc.event.EventBus;
2334
import com.alipay.sofa.rpc.server.triple.TripleContants;
2435
import com.alipay.sofa.rpc.tracer.sofatracer.TripleTracerAdapter;
2536
import io.grpc.CallOptions;
@@ -59,12 +70,15 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT
5970

6071
@Override
6172
public void start(Listener<RespT> responseListener, Metadata requestHeader) {
62-
73+
RpcInternalContext internalContext = RpcInternalContext.getContext();
6374
RpcInvokeContext context = RpcInvokeContext.getContext();
6475
SofaRequest sofaRequest = (SofaRequest) context.get(TripleContants.SOFA_REQUEST_KEY);
6576

66-
ConsumerConfig consumerConfig = (ConsumerConfig) context.get(TripleContants.SOFA_CONSUMER_CONFIG_KEY);
67-
TripleTracerAdapter.beforeSend(sofaRequest, consumerConfig, requestHeader);
77+
ConsumerConfig<?> consumerConfig = (ConsumerConfig<?>) context
78+
.get(TripleContants.SOFA_CONSUMER_CONFIG_KEY);
79+
TripleTracerAdapter.beforeSend(sofaRequest, consumerConfig, requestHeader, method);
80+
SofaTraceContext sofaTraceContext = SofaTraceContextHolder.getSofaTraceContext();
81+
SofaTracerSpan clientSpan = sofaTraceContext.getCurrentSpan();
6882
if (RpcRunningState.isDebugMode()) {
6983
LOGGER.info("[2]prepare to send from client:{}", requestHeader);
7084
}
@@ -80,18 +94,48 @@ public void onHeaders(Metadata responseHeader) {
8094

8195
@Override
8296
public void onMessage(RespT message) {
83-
if (RpcRunningState.isDebugMode()) {
84-
LOGGER.info("[4]response message received from server:{}", message);
97+
// onMessage -> onNext()
98+
try {
99+
if (sofaRequest.isAsync()) {
100+
RpcInvokeContext.setContext(context);
101+
sofaTraceContext.push(clientSpan);
102+
}
103+
if (RpcRunningState.isDebugMode()) {
104+
LOGGER.info("[4]response message received from server:{}", message);
105+
}
106+
super.onMessage(message);
107+
} finally {
108+
if (sofaRequest.isAsync()) {
109+
sofaTraceContext.clear();
110+
RpcInvokeContext.removeContext();
111+
}
85112
}
86-
super.onMessage(message);
87113
}
88114

89115
@Override
90116
public void onClose(Status status, Metadata trailers) {
91-
if (RpcRunningState.isDebugMode()) {
92-
LOGGER.info("[5]response close received from server:{},trailers:{}", status, trailers);
117+
// onClose -> onComplete() or onError()
118+
try {
119+
if (sofaRequest.isAsync()) {
120+
RpcInvokeContext.setContext(context);
121+
sofaTraceContext.push(clientSpan);
122+
}
123+
if (RpcRunningState.isDebugMode()) {
124+
LOGGER.info("[5]response close received from server:{},trailers:{}", status, trailers);
125+
}
126+
super.onClose(status, trailers);
127+
} finally {
128+
if (sofaRequest.isAsync()) {
129+
Throwable throwable = getThrowableFromStatus(status);
130+
RpcInternalContext.setContext(internalContext);
131+
if (EventBus.isEnable(ClientAsyncReceiveEvent.class)) {
132+
EventBus.post(new ClientAsyncReceiveEvent(consumerConfig, null,
133+
sofaRequest, new SofaResponse(), throwable));
134+
}
135+
RpcInvokeContext.removeContext();
136+
RpcInternalContext.removeAllContext();
137+
}
93138
}
94-
super.onClose(status, trailers);
95139
}
96140

97141
@Override
@@ -104,6 +148,27 @@ public void onReady() {
104148
}, requestHeader);
105149
}
106150

151+
@Override
152+
public void sendMessage(ReqT message) {
153+
try {
154+
super.sendMessage(message);
155+
} catch (Throwable t) {
156+
LOGGER.error("Client invoke grpc sendMessage meet error:", t);
157+
throw t;
158+
}
159+
}
107160
};
108161
}
162+
163+
private static Throwable getThrowableFromStatus(Status status) {
164+
if (status.getCode() == Status.OK.getCode()) {
165+
return null;
166+
} else if (status.getCode() == Status.UNAVAILABLE.getCode()) {
167+
return new SofaRouteException(status.getDescription(), status.getCause());
168+
} else if (status.getCode() == Status.DEADLINE_EXCEEDED.getCode()) {
169+
return new SofaTimeOutException(status.getDescription(), status.getCause());
170+
} else {
171+
return new SofaRpcException(RpcErrorType.UNKNOWN, status.getDescription(), status.getCause());
172+
}
173+
}
109174
}

0 commit comments

Comments
 (0)