Skip to content

Commit f388fcb

Browse files
xuzhiguangxuzhiguang
committed
feat(threadpool): support tracing for invokeAll and invokeAny
1 parent 59237c4 commit f388fcb

6 files changed

Lines changed: 456 additions & 2 deletions

File tree

‎CHANGES.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ Release Notes.
1111
* Add a Jetty 12 server plugin (`jetty-server-12.x`). Jetty 12 removed the `HttpChannel` handle target and moved request handling to the async `Server#handle(Request, Response, Callback)` core API, so it needs a separate plugin from the merged `jetty-server`.
1212
* Add a Struts 7 plugin (`struts2-7.x`) for Jakarta Struts, whose `DefaultActionInvocation` moved to `org.apache.struts2`.
1313
* Added support for Lettuce reactive Redis commands.
14+
* Add tracing support for `invokeAll` and `invokeAny` in the JDK thread pool plugin.
1415
* Add Spring AI 1.x plugin and GenAI layer.
1516
* Fix httpclient-5.x plugin injecting sw8 propagation headers into ClickHouse HTTP requests (port 8123), causing HTTP 400. Add `PROPAGATION_EXCLUDE_PORTS` config to skip tracing (including header injection) for specified ports in the classic client interceptor.
1617
* Add Spring RabbitMQ 2.x - 4.x plugin.

‎apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptor.java‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
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;
20+
21+
import java.lang.reflect.Method;
22+
import java.util.ArrayList;
23+
import java.util.Collection;
24+
import java.util.List;
25+
import java.util.concurrent.Callable;
26+
import org.apache.skywalking.apm.agent.core.context.ContextManager;
27+
import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
28+
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
29+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
30+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
31+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
32+
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
33+
import org.apache.skywalking.apm.plugin.wrapper.SwCallableWrapper;
34+
35+
public class ThreadPoolInvokeMethodInterceptor implements InstanceMethodsAroundInterceptor {
36+
37+
private static final String OPERATION_NAME_PREFIX = "ThreadPoolExecutor/";
38+
39+
@Override
40+
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
41+
MethodInterceptResult result) throws Throwable {
42+
if (!shouldEnhance(allArguments)) {
43+
return;
44+
}
45+
46+
AbstractSpan span = ContextManager.createLocalSpan(OPERATION_NAME_PREFIX + method.getName());
47+
span.setComponent(ComponentsDefine.JDK_THREADING);
48+
49+
ContextSnapshot contextSnapshot = ContextManager.capture();
50+
Collection<?> callables = (Collection<?>) allArguments[0];
51+
List<Object> wrappedCallables = new ArrayList<>(callables.size());
52+
for (Object callable : callables) {
53+
wrappedCallables.add(wrap(callable, contextSnapshot));
54+
}
55+
allArguments[0] = wrappedCallables;
56+
}
57+
58+
@Override
59+
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
60+
Object ret) throws Throwable {
61+
if (shouldEnhance(allArguments)) {
62+
ContextManager.stopSpan();
63+
}
64+
return ret;
65+
}
66+
67+
@Override
68+
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
69+
Class<?>[] argumentsTypes, Throwable t) {
70+
if (shouldEnhance(allArguments)) {
71+
ContextManager.activeSpan().log(t);
72+
}
73+
}
74+
75+
private boolean shouldEnhance(Object[] allArguments) {
76+
return ContextManager.isActive()
77+
&& allArguments != null
78+
&& allArguments.length > 0
79+
&& allArguments[0] instanceof Collection;
80+
}
81+
82+
private Object wrap(Object callable, ContextSnapshot contextSnapshot) {
83+
if (!(callable instanceof Callable) || callable instanceof SwCallableWrapper || hasCapturedContext(callable)) {
84+
return callable;
85+
}
86+
return new SwCallableWrapper((Callable) callable, contextSnapshot);
87+
}
88+
89+
private boolean hasCapturedContext(Object callable) {
90+
return callable instanceof EnhancedInstance
91+
&& ((EnhancedInstance) callable).getSkyWalkingDynamicField() instanceof ContextSnapshot;
92+
}
93+
}

‎apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInstrumentation.java‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,10 +37,16 @@ public class ThreadPoolExecutorInstrumentation extends ClassInstanceMethodsEnhan
3737

3838
private static final String INTERCEPT_SUBMIT_METHOD = "submit";
3939

40+
private static final String INTERCEPT_INVOKE_ALL_METHOD = "invokeAll";
41+
42+
private static final String INTERCEPT_INVOKE_ANY_METHOD = "invokeAny";
43+
4044
private static final String INTERCEPT_EXECUTE_METHOD_HANDLE = "org.apache.skywalking.apm.plugin.ThreadPoolExecuteMethodInterceptor";
4145

4246
private static final String INTERCEPT_SUBMIT_METHOD_HANDLE = "org.apache.skywalking.apm.plugin.ThreadPoolSubmitMethodInterceptor";
4347

48+
private static final String INTERCEPT_INVOKE_METHOD_HANDLE = "org.apache.skywalking.apm.plugin.ThreadPoolInvokeMethodInterceptor";
49+
4450
@Override
4551
public boolean isBootstrapInstrumentation() {
4652
return true;
@@ -86,6 +92,23 @@ public String getMethodsInterceptor() {
8692
return INTERCEPT_SUBMIT_METHOD_HANDLE;
8793
}
8894

95+
@Override
96+
public boolean isOverrideArgs() {
97+
return true;
98+
}
99+
},
100+
new InstanceMethodsInterceptPoint() {
101+
@Override
102+
public ElementMatcher<MethodDescription> getMethodsMatcher() {
103+
return ElementMatchers.named(INTERCEPT_INVOKE_ALL_METHOD)
104+
.or(ElementMatchers.named(INTERCEPT_INVOKE_ANY_METHOD));
105+
}
106+
107+
@Override
108+
public String getMethodsInterceptor() {
109+
return INTERCEPT_INVOKE_METHOD_HANDLE;
110+
}
111+
89112
@Override
90113
public boolean isOverrideArgs() {
91114
return true;

‎apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/test/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptorTest.java‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,129 @@
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+
package org.apache.skywalking.apm.plugin;
19+
20+
import static org.hamcrest.CoreMatchers.is;
21+
import static org.hamcrest.MatcherAssert.assertThat;
22+
import java.lang.reflect.Method;
23+
import java.util.Collection;
24+
import java.util.Collections;
25+
import java.util.List;
26+
import java.util.concurrent.AbstractExecutorService;
27+
import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
28+
import org.apache.skywalking.apm.agent.core.context.ContextManager;
29+
import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
30+
import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
31+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
32+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
33+
import org.apache.skywalking.apm.agent.test.helper.SegmentHelper;
34+
import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule;
35+
import org.apache.skywalking.apm.agent.test.tools.SegmentStorage;
36+
import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint;
37+
import org.apache.skywalking.apm.agent.test.tools.SpanAssert;
38+
import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
39+
import org.junit.Rule;
40+
import org.junit.Test;
41+
import org.junit.runner.RunWith;
42+
import org.mockito.Mock;
43+
44+
@RunWith(TracingSegmentRunner.class)
45+
public class ThreadPoolInvokeMethodInterceptorTest {
46+
47+
@SegmentStoragePoint
48+
private SegmentStorage segmentStorage;
49+
50+
@Rule
51+
public AgentServiceRule agentServiceRule = new AgentServiceRule();
52+
53+
@Mock
54+
private EnhancedInstance enhancedInstance;
55+
56+
@Mock
57+
private MethodInterceptResult result;
58+
59+
private final ThreadPoolInvokeMethodInterceptor interceptor = new ThreadPoolInvokeMethodInterceptor();
60+
61+
@Test
62+
public void shouldIgnoreUnexpectedArguments() throws Throwable {
63+
Object[][] unexpectedArguments = new Object[][] {
64+
null,
65+
new Object[0],
66+
new Object[] {null},
67+
new Object[] {"not-a-collection"}
68+
};
69+
70+
for (Object[] arguments : unexpectedArguments) {
71+
ContextManager.createEntrySpan("parent", new ContextCarrier());
72+
73+
interceptor.beforeMethod(enhancedInstance, invokeAllMethod(), arguments, null, result);
74+
interceptor.handleMethodException(
75+
enhancedInstance, invokeAllMethod(), arguments, null, new IllegalStateException("ignored"));
76+
interceptor.afterMethod(enhancedInstance, invokeAllMethod(), arguments, null, null);
77+
78+
assertThat(ContextManager.isActive(), is(true));
79+
ContextManager.stopSpan();
80+
}
81+
82+
assertThat(segmentStorage.getTraceSegments().size(), is(4));
83+
for (TraceSegment traceSegment : segmentStorage.getTraceSegments()) {
84+
List<AbstractTracingSpan> spans = SegmentHelper.getSpans(traceSegment);
85+
assertThat(spans.size(), is(1));
86+
assertThat(spans.get(0).getOperationName(), is("parent"));
87+
SpanAssert.assertOccurException(spans.get(0), false);
88+
}
89+
}
90+
91+
@Test
92+
public void shouldTraceEmptyCollection() throws Throwable {
93+
Object[] arguments = new Object[] {Collections.emptyList()};
94+
ContextManager.createEntrySpan("parent", new ContextCarrier());
95+
96+
interceptor.beforeMethod(enhancedInstance, invokeAllMethod(), arguments, null, result);
97+
interceptor.afterMethod(enhancedInstance, invokeAllMethod(), arguments, null, null);
98+
ContextManager.stopSpan();
99+
100+
List<AbstractTracingSpan> spans = SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0));
101+
assertThat(spans.size(), is(2));
102+
assertThat(spans.get(0).getOperationName(), is("ThreadPoolExecutor/invokeAll"));
103+
}
104+
105+
@Test
106+
public void shouldTraceAlternativeInvokeAllSignature() throws Throwable {
107+
Object[] arguments = new Object[] {Collections.emptyList(), "alternative"};
108+
ContextManager.createEntrySpan("parent", new ContextCarrier());
109+
110+
interceptor.beforeMethod(enhancedInstance, alternativeInvokeAllMethod(), arguments, null, result);
111+
interceptor.afterMethod(enhancedInstance, alternativeInvokeAllMethod(), arguments, null, null);
112+
ContextManager.stopSpan();
113+
114+
List<AbstractTracingSpan> spans = SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0));
115+
assertThat(spans.size(), is(2));
116+
assertThat(spans.get(0).getOperationName(), is("ThreadPoolExecutor/invokeAll"));
117+
}
118+
119+
private Method invokeAllMethod() throws NoSuchMethodException {
120+
return AbstractExecutorService.class.getMethod("invokeAll", Collection.class);
121+
}
122+
123+
private Method alternativeInvokeAllMethod() throws NoSuchMethodException {
124+
return getClass().getDeclaredMethod("invokeAll", Collection.class, String.class);
125+
}
126+
127+
private void invokeAll(Collection<?> callables, String alternative) {
128+
}
129+
}

0 commit comments

Comments
 (0)