Skip to content

Commit a751e32

Browse files
authored
Support across thread tracing for SOFA-RPC (#675)
1 parent 466f173 commit a751e32

File tree

15 files changed

+624
-27
lines changed

15 files changed

+624
-27
lines changed

CHANGES.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ Release Notes.
1818
* Archive the expired plugins `impala-jdbc-2.6.x-plugin`.
1919
* Fix a bug in Spring Cloud Gateway if HttpClientFinalizer#send does not invoke, the span created at NettyRoutingFilterInterceptor can not stop.
2020
* Fix not tracing in HttpClient v5 when HttpHost(arg[0]) is null but `RoutingSupport#determineHost` works.
21+
* Support across thread tracing for SOFA-RPC.
2122

2223
#### Documentation
2324
* Update docs to describe `expired-plugins`.
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
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+
*
17+
*/
18+
19+
package org.apache.skywalking.apm.plugin.sofarpc;
20+
21+
import com.alipay.remoting.InvokeCallback;
22+
import java.util.concurrent.Executor;
23+
import lombok.AccessLevel;
24+
import lombok.Getter;
25+
import org.apache.skywalking.apm.agent.core.context.ContextManager;
26+
import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
27+
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
28+
29+
public class InvokeCallbackWrapper implements InvokeCallback {
30+
31+
@Getter(AccessLevel.PACKAGE)
32+
private ContextSnapshot contextSnapshot;
33+
@Getter(AccessLevel.PACKAGE)
34+
private final InvokeCallback invokeCallback;
35+
36+
public InvokeCallbackWrapper(InvokeCallback invokeCallback) {
37+
if (ContextManager.isActive()) {
38+
this.contextSnapshot = ContextManager.capture();
39+
}
40+
this.invokeCallback = invokeCallback;
41+
}
42+
43+
@Override
44+
public void onResponse(final Object o) {
45+
ContextManager.createLocalSpan("Thread/" + invokeCallback.getClass().getName() + "/onResponse");
46+
if (contextSnapshot != null) {
47+
ContextManager.continued(contextSnapshot);
48+
}
49+
try {
50+
invokeCallback.onResponse(o);
51+
} catch (Throwable t) {
52+
ContextManager.activeSpan().log(t);
53+
throw t;
54+
} finally {
55+
contextSnapshot = null;
56+
ContextManager.stopSpan();
57+
}
58+
59+
}
60+
61+
@Override
62+
public void onException(final Throwable throwable) {
63+
ContextManager.createLocalSpan("Thread/" + invokeCallback.getClass().getName() + "/onException");
64+
if (contextSnapshot != null) {
65+
ContextManager.continued(contextSnapshot);
66+
}
67+
if (throwable != null) {
68+
AbstractSpan abstractSpan = ContextManager.activeSpan();
69+
if (abstractSpan != null) {
70+
abstractSpan.log(throwable);
71+
}
72+
}
73+
try {
74+
invokeCallback.onException(throwable);
75+
} catch (Throwable t) {
76+
ContextManager.activeSpan().log(t);
77+
throw t;
78+
} finally {
79+
contextSnapshot = null;
80+
ContextManager.stopSpan();
81+
}
82+
}
83+
84+
@Override
85+
public Executor getExecutor() {
86+
return invokeCallback.getExecutor();
87+
}
88+
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
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+
*
17+
*/
18+
19+
package org.apache.skywalking.apm.plugin.sofarpc;
20+
21+
import net.bytebuddy.description.method.MethodDescription;
22+
import net.bytebuddy.matcher.ElementMatcher;
23+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
24+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
25+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
26+
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
27+
import org.apache.skywalking.apm.agent.core.plugin.match.NameMatch;
28+
29+
import static net.bytebuddy.matcher.ElementMatchers.named;
30+
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
31+
32+
public class SofaBoltCallbackInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
33+
34+
private static final String ENHANCE_CLASS = "com.alipay.remoting.BaseRemoting";
35+
private static final String INVOKE_METHOD_INTERCEPTOR = "org.apache.skywalking.apm.plugin.sofarpc.SofaBoltCallbackInvokeInterceptor";
36+
private static final String INVOKE_METHOD = "invokeWithCallback";
37+
38+
@Override
39+
protected ClassMatch enhanceClass() {
40+
return NameMatch.byName(ENHANCE_CLASS);
41+
}
42+
43+
@Override
44+
public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
45+
return null;
46+
}
47+
48+
@Override
49+
public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
50+
return new InstanceMethodsInterceptPoint[] {
51+
new InstanceMethodsInterceptPoint() {
52+
@Override
53+
public ElementMatcher<MethodDescription> getMethodsMatcher() {
54+
return named(INVOKE_METHOD).and(
55+
takesArguments(4));
56+
}
57+
58+
@Override
59+
public String getMethodsInterceptor() {
60+
return INVOKE_METHOD_INTERCEPTOR;
61+
}
62+
63+
@Override
64+
public boolean isOverrideArgs() {
65+
return true;
66+
}
67+
}
68+
};
69+
}
70+
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
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+
*
17+
*/
18+
19+
package org.apache.skywalking.apm.plugin.sofarpc;
20+
21+
import com.alipay.remoting.InvokeCallback;
22+
import java.lang.reflect.Method;
23+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
24+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
25+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
26+
27+
public class SofaBoltCallbackInvokeInterceptor implements InstanceMethodsAroundInterceptor {
28+
@Override
29+
public void beforeMethod(EnhancedInstance objInst,
30+
Method method,
31+
Object[] allArguments,
32+
Class<?>[] argumentsTypes,
33+
MethodInterceptResult result) {
34+
if (allArguments[2] instanceof InvokeCallback) {
35+
allArguments[2] = new InvokeCallbackWrapper((InvokeCallback) allArguments[2]);
36+
}
37+
}
38+
39+
@Override
40+
public Object afterMethod(EnhancedInstance objInst,
41+
Method method,
42+
Object[] allArguments,
43+
Class<?>[] argumentsTypes,
44+
Object ret) {
45+
return ret;
46+
}
47+
48+
@Override
49+
public void handleMethodException(EnhancedInstance objInst,
50+
Method method,
51+
Object[] allArguments,
52+
Class<?>[] argumentsTypes,
53+
Throwable t) {
54+
}
55+
}

apm-sniffer/apm-sdk-plugin/sofarpc-plugin/src/main/resources/skywalking-plugin.def

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,3 +16,4 @@
1616

1717
sofarpc=org.apache.skywalking.apm.plugin.sofarpc.SofaRpcConsumerInstrumentation
1818
sofarpc=org.apache.skywalking.apm.plugin.sofarpc.SofaRpcProviderInstrumentation
19+
sofarpc=org.apache.skywalking.apm.plugin.sofarpc.SofaBoltCallbackInstrumentation
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,169 @@
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+
*
17+
*/
18+
19+
package org.apache.skywalking.apm.plugin.sofarpc;
20+
21+
import com.alipay.remoting.InvokeCallback;
22+
import java.util.List;
23+
import java.util.concurrent.CountDownLatch;
24+
import java.util.concurrent.Executor;
25+
import java.util.concurrent.Executors;
26+
import org.apache.skywalking.apm.agent.core.context.ContextManager;
27+
import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
28+
import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
29+
import org.apache.skywalking.apm.agent.test.helper.SegmentHelper;
30+
import org.apache.skywalking.apm.agent.test.helper.SpanHelper;
31+
import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule;
32+
import org.apache.skywalking.apm.agent.test.tools.SegmentStorage;
33+
import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint;
34+
import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
35+
import org.junit.Assert;
36+
import org.junit.Before;
37+
import org.junit.Rule;
38+
import org.junit.Test;
39+
import org.junit.runner.RunWith;
40+
import org.mockito.junit.MockitoJUnit;
41+
import org.mockito.junit.MockitoRule;
42+
43+
import static org.hamcrest.CoreMatchers.is;
44+
import static org.junit.Assert.assertEquals;
45+
import static org.junit.Assert.assertThat;
46+
47+
@RunWith(TracingSegmentRunner.class)
48+
public class InvokeCallbackWrapperTest {
49+
50+
@SegmentStoragePoint
51+
private SegmentStorage segmentStorage;
52+
53+
private Executor executor = Executors.newFixedThreadPool(1);
54+
55+
@Rule
56+
public AgentServiceRule agentServiceRule = new AgentServiceRule();
57+
@Rule
58+
public MockitoRule rule = MockitoJUnit.rule();
59+
60+
private InvokeCallback callback;
61+
62+
@Before
63+
public void before() {
64+
callback = new InvokeCallback() {
65+
@Override
66+
public void onResponse(final Object o) {
67+
}
68+
69+
@Override
70+
public void onException(final Throwable throwable) {
71+
}
72+
73+
@Override
74+
public Executor getExecutor() {
75+
return null;
76+
}
77+
};
78+
}
79+
80+
static class WrapperWrapper implements InvokeCallback {
81+
82+
private InvokeCallback callback;
83+
84+
private CountDownLatch countDownLatch;
85+
86+
public CountDownLatch getCountDownLatch() {
87+
return countDownLatch;
88+
}
89+
90+
public WrapperWrapper(InvokeCallback callback) {
91+
this.countDownLatch = new CountDownLatch(1);
92+
this.callback = callback;
93+
}
94+
95+
@Override
96+
public void onResponse(final Object o) {
97+
callback.onResponse(o);
98+
countDownLatch.countDown();
99+
}
100+
101+
@Override
102+
public void onException(final Throwable throwable) {
103+
callback.onException(throwable);
104+
countDownLatch.countDown();
105+
}
106+
107+
@Override
108+
public Executor getExecutor() {
109+
return null;
110+
}
111+
}
112+
113+
@Test
114+
public void testConstruct() {
115+
InvokeCallbackWrapper wrapper = new InvokeCallbackWrapper(callback);
116+
Assert.assertSame(callback, wrapper.getInvokeCallback());
117+
Assert.assertNull(wrapper.getContextSnapshot());
118+
119+
ContextManager.createEntrySpan("sofarpc", null);
120+
wrapper = new InvokeCallbackWrapper(callback);
121+
Assert.assertSame(callback, wrapper.getInvokeCallback());
122+
Assert.assertEquals(ContextManager.getGlobalTraceId(), wrapper.getContextSnapshot().getTraceId().getId());
123+
Assert.assertEquals("sofarpc", wrapper.getContextSnapshot().getParentEndpoint());
124+
ContextManager.stopSpan();
125+
}
126+
127+
@Test
128+
public void testOnResponse() throws InterruptedException {
129+
ContextManager.createEntrySpan("sofarpc", null);
130+
InvokeCallbackWrapper wrapper = new InvokeCallbackWrapper(callback);
131+
final WrapperWrapper wrapperWrapper = new WrapperWrapper(wrapper);
132+
executor.execute(() -> wrapperWrapper.onResponse(null));
133+
ContextManager.stopSpan();
134+
wrapperWrapper.getCountDownLatch().await();
135+
136+
assertThat(segmentStorage.getTraceSegments().size(), is(2));
137+
TraceSegment traceSegment = segmentStorage.getTraceSegments().get(0);
138+
List<AbstractTracingSpan> spans = SegmentHelper.getSpans(traceSegment);
139+
assertThat(spans.size(), is(1));
140+
141+
TraceSegment traceSegment2 = segmentStorage.getTraceSegments().get(1);
142+
List<AbstractTracingSpan> spans2 = SegmentHelper.getSpans(traceSegment2);
143+
assertThat(spans2.size(), is(1));
144+
assertEquals("sofarpc", traceSegment2.getRef().getParentEndpoint());
145+
}
146+
147+
@Test
148+
public void testOnException() throws InterruptedException {
149+
ContextManager.createEntrySpan("sofarpc", null);
150+
InvokeCallbackWrapper wrapper = new InvokeCallbackWrapper(callback);
151+
final WrapperWrapper wrapperWrapper = new WrapperWrapper(wrapper);
152+
final Throwable throwable = new Throwable();
153+
executor.execute(() -> wrapperWrapper.onException(throwable));
154+
ContextManager.stopSpan();
155+
wrapperWrapper.getCountDownLatch().await();
156+
157+
assertThat(segmentStorage.getTraceSegments().size(), is(2));
158+
TraceSegment traceSegment = segmentStorage.getTraceSegments().get(0);
159+
List<AbstractTracingSpan> spans = SegmentHelper.getSpans(traceSegment);
160+
assertThat(spans.size(), is(1));
161+
162+
TraceSegment traceSegment2 = segmentStorage.getTraceSegments().get(1);
163+
List<AbstractTracingSpan> spans2 = SegmentHelper.getSpans(traceSegment2);
164+
assertThat(spans2.size(), is(1));
165+
assertThat(SpanHelper.getLogs(spans2.get(0)).size(), is(1));
166+
167+
}
168+
169+
}

0 commit comments

Comments
 (0)