Skip to content

Commit 1fcee98

Browse files
xuzhiguangxuzhiguang
authored
Add tracing support for ThreadPoolExecutor invokeAll and invokeAny (#816)
1 parent 59237c4 commit 1fcee98

8 files changed

Lines changed: 666 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 (`jdk-threadpool-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: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
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.v2.InstanceMethodsAroundInterceptorV2;
31+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.v2.MethodInvocationContext;
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 InstanceMethodsAroundInterceptorV2 {
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+
MethodInvocationContext context) 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+
context.setContext(span);
49+
50+
ContextSnapshot contextSnapshot = ContextManager.capture();
51+
Collection<?> callables = (Collection<?>) allArguments[0];
52+
List<Object> wrappedCallables = new ArrayList<>(callables.size());
53+
for (Object callable : callables) {
54+
wrappedCallables.add(wrap(callable, contextSnapshot));
55+
}
56+
allArguments[0] = wrappedCallables;
57+
}
58+
59+
@Override
60+
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
61+
Object ret, MethodInvocationContext context) throws Throwable {
62+
AbstractSpan span = (AbstractSpan) context.getContext();
63+
if (span != null) {
64+
ContextManager.stopSpan(span);
65+
}
66+
return ret;
67+
}
68+
69+
@Override
70+
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
71+
Class<?>[] argumentsTypes, Throwable t, MethodInvocationContext context) {
72+
AbstractSpan span = (AbstractSpan) context.getContext();
73+
if (span != null) {
74+
span.log(t);
75+
}
76+
}
77+
78+
private boolean shouldEnhance(Object[] allArguments) {
79+
return ContextManager.isActive()
80+
&& allArguments != null
81+
&& allArguments.length > 0
82+
&& allArguments[0] instanceof Collection;
83+
}
84+
85+
private Object wrap(Object callable, ContextSnapshot contextSnapshot) {
86+
if (!(callable instanceof Callable) || callable instanceof SwCallableWrapper || hasCapturedContext(callable)) {
87+
return callable;
88+
}
89+
return new SwCallableWrapper((Callable) callable, contextSnapshot);
90+
}
91+
92+
private boolean hasCapturedContext(Object callable) {
93+
return callable instanceof EnhancedInstance
94+
&& ((EnhancedInstance) callable).getSkyWalkingDynamicField() instanceof ContextSnapshot;
95+
}
96+
}

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

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
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.define;
20+
21+
import java.util.Collection;
22+
import java.util.concurrent.TimeUnit;
23+
import net.bytebuddy.description.method.MethodDescription;
24+
import net.bytebuddy.matcher.ElementMatcher;
25+
import net.bytebuddy.matcher.ElementMatchers;
26+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
27+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.v2.ClassInstanceMethodsEnhancePluginDefineV2;
28+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.v2.InstanceMethodsInterceptV2Point;
29+
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
30+
import org.apache.skywalking.apm.agent.core.plugin.match.NameMatch;
31+
32+
public class ThreadPoolExecutorInvokeInstrumentation extends ClassInstanceMethodsEnhancePluginDefineV2 {
33+
34+
private static final String ENHANCE_CLASS = "java.util.concurrent.ThreadPoolExecutor";
35+
36+
private static final String INTERCEPT_INVOKE_ALL_METHOD = "invokeAll";
37+
38+
private static final String INTERCEPT_INVOKE_ANY_METHOD = "invokeAny";
39+
40+
private static final String INTERCEPT_INVOKE_METHOD_HANDLE =
41+
"org.apache.skywalking.apm.plugin.ThreadPoolInvokeMethodInterceptor";
42+
43+
@Override
44+
public boolean isBootstrapInstrumentation() {
45+
return true;
46+
}
47+
48+
@Override
49+
protected ClassMatch enhanceClass() {
50+
return NameMatch.byName(ENHANCE_CLASS);
51+
}
52+
53+
@Override
54+
public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
55+
return new ConstructorInterceptPoint[0];
56+
}
57+
58+
@Override
59+
public InstanceMethodsInterceptV2Point[] getInstanceMethodsInterceptV2Points() {
60+
return new InstanceMethodsInterceptV2Point[] {
61+
new InstanceMethodsInterceptV2Point() {
62+
@Override
63+
public ElementMatcher<MethodDescription> getMethodsMatcher() {
64+
return ElementMatchers.named(INTERCEPT_INVOKE_ALL_METHOD)
65+
.or(ElementMatchers.named(INTERCEPT_INVOKE_ANY_METHOD))
66+
.and(ElementMatchers.takesArguments(Collection.class)
67+
.or(ElementMatchers.takesArguments(
68+
Collection.class, long.class, TimeUnit.class)));
69+
}
70+
71+
@Override
72+
public String getMethodsInterceptorV2() {
73+
return INTERCEPT_INVOKE_METHOD_HANDLE;
74+
}
75+
76+
@Override
77+
public boolean isOverrideArgs() {
78+
return true;
79+
}
80+
}
81+
};
82+
}
83+
}

‎apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/resources/skywalking-plugin.def‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,3 +15,4 @@
1515
# limitations under the License.
1616

1717
jdk-threadpool-plugin=org.apache.skywalking.apm.plugin.define.ThreadPoolExecutorInstrumentation
18+
jdk-threadpool-plugin=org.apache.skywalking.apm.plugin.define.ThreadPoolExecutorInvokeInstrumentation

0 commit comments

Comments
 (0)