From f388fcbf11dab06bcbe12c329d4619890dca4992 Mon Sep 17 00:00:00 2001 From: xuzhiguang <13520773230@163.com> Date: Wed, 29 Jul 2026 15:32:48 +0800 Subject: [PATCH 1/2] feat(threadpool): support tracing for invokeAll and invokeAny --- CHANGES.md | 1 + .../ThreadPoolInvokeMethodInterceptor.java | 93 ++++++++++ .../ThreadPoolExecutorInstrumentation.java | 23 +++ ...ThreadPoolInvokeMethodInterceptorTest.java | 129 ++++++++++++++ .../config/expectedData.yaml | 164 +++++++++++++++++- .../testcase/jdk/threading/Application.java | 48 ++++- 6 files changed, 456 insertions(+), 2 deletions(-) create mode 100644 apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptor.java create mode 100644 apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/test/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptorTest.java diff --git a/CHANGES.md b/CHANGES.md index 8b4e789c2f..be584db540 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -11,6 +11,7 @@ Release Notes. * 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`. * Add a Struts 7 plugin (`struts2-7.x`) for Jakarta Struts, whose `DefaultActionInvocation` moved to `org.apache.struts2`. * Added support for Lettuce reactive Redis commands. +* Add tracing support for `invokeAll` and `invokeAny` in the JDK thread pool plugin. * Add Spring AI 1.x plugin and GenAI layer. * 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. * Add Spring RabbitMQ 2.x - 4.x plugin. diff --git a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptor.java b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptor.java new file mode 100644 index 0000000000..a9f7fbc494 --- /dev/null +++ b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptor.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 org.apache.skywalking.apm.plugin; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.Callable; +import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.ContextSnapshot; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; +import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; +import org.apache.skywalking.apm.plugin.wrapper.SwCallableWrapper; + +public class ThreadPoolInvokeMethodInterceptor implements InstanceMethodsAroundInterceptor { + + private static final String OPERATION_NAME_PREFIX = "ThreadPoolExecutor/"; + + @Override + public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + MethodInterceptResult result) throws Throwable { + if (!shouldEnhance(allArguments)) { + return; + } + + AbstractSpan span = ContextManager.createLocalSpan(OPERATION_NAME_PREFIX + method.getName()); + span.setComponent(ComponentsDefine.JDK_THREADING); + + ContextSnapshot contextSnapshot = ContextManager.capture(); + Collection callables = (Collection) allArguments[0]; + List wrappedCallables = new ArrayList<>(callables.size()); + for (Object callable : callables) { + wrappedCallables.add(wrap(callable, contextSnapshot)); + } + allArguments[0] = wrappedCallables; + } + + @Override + public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + Object ret) throws Throwable { + if (shouldEnhance(allArguments)) { + ContextManager.stopSpan(); + } + return ret; + } + + @Override + public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, + Class[] argumentsTypes, Throwable t) { + if (shouldEnhance(allArguments)) { + ContextManager.activeSpan().log(t); + } + } + + private boolean shouldEnhance(Object[] allArguments) { + return ContextManager.isActive() + && allArguments != null + && allArguments.length > 0 + && allArguments[0] instanceof Collection; + } + + private Object wrap(Object callable, ContextSnapshot contextSnapshot) { + if (!(callable instanceof Callable) || callable instanceof SwCallableWrapper || hasCapturedContext(callable)) { + return callable; + } + return new SwCallableWrapper((Callable) callable, contextSnapshot); + } + + private boolean hasCapturedContext(Object callable) { + return callable instanceof EnhancedInstance + && ((EnhancedInstance) callable).getSkyWalkingDynamicField() instanceof ContextSnapshot; + } +} diff --git a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInstrumentation.java b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInstrumentation.java index d057a8f317..a895d61d53 100644 --- a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInstrumentation.java +++ b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInstrumentation.java @@ -37,10 +37,16 @@ public class ThreadPoolExecutorInstrumentation extends ClassInstanceMethodsEnhan private static final String INTERCEPT_SUBMIT_METHOD = "submit"; + private static final String INTERCEPT_INVOKE_ALL_METHOD = "invokeAll"; + + private static final String INTERCEPT_INVOKE_ANY_METHOD = "invokeAny"; + private static final String INTERCEPT_EXECUTE_METHOD_HANDLE = "org.apache.skywalking.apm.plugin.ThreadPoolExecuteMethodInterceptor"; private static final String INTERCEPT_SUBMIT_METHOD_HANDLE = "org.apache.skywalking.apm.plugin.ThreadPoolSubmitMethodInterceptor"; + private static final String INTERCEPT_INVOKE_METHOD_HANDLE = "org.apache.skywalking.apm.plugin.ThreadPoolInvokeMethodInterceptor"; + @Override public boolean isBootstrapInstrumentation() { return true; @@ -86,6 +92,23 @@ public String getMethodsInterceptor() { return INTERCEPT_SUBMIT_METHOD_HANDLE; } + @Override + public boolean isOverrideArgs() { + return true; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return ElementMatchers.named(INTERCEPT_INVOKE_ALL_METHOD) + .or(ElementMatchers.named(INTERCEPT_INVOKE_ANY_METHOD)); + } + + @Override + public String getMethodsInterceptor() { + return INTERCEPT_INVOKE_METHOD_HANDLE; + } + @Override public boolean isOverrideArgs() { return true; diff --git a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/test/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptorTest.java b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/test/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptorTest.java new file mode 100644 index 0000000000..319a1da528 --- /dev/null +++ b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/test/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptorTest.java @@ -0,0 +1,129 @@ +/* + * 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 org.apache.skywalking.apm.plugin; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; +import java.lang.reflect.Method; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.AbstractExecutorService; +import org.apache.skywalking.apm.agent.core.context.ContextCarrier; +import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan; +import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; +import org.apache.skywalking.apm.agent.test.helper.SegmentHelper; +import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule; +import org.apache.skywalking.apm.agent.test.tools.SegmentStorage; +import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint; +import org.apache.skywalking.apm.agent.test.tools.SpanAssert; +import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; + +@RunWith(TracingSegmentRunner.class) +public class ThreadPoolInvokeMethodInterceptorTest { + + @SegmentStoragePoint + private SegmentStorage segmentStorage; + + @Rule + public AgentServiceRule agentServiceRule = new AgentServiceRule(); + + @Mock + private EnhancedInstance enhancedInstance; + + @Mock + private MethodInterceptResult result; + + private final ThreadPoolInvokeMethodInterceptor interceptor = new ThreadPoolInvokeMethodInterceptor(); + + @Test + public void shouldIgnoreUnexpectedArguments() throws Throwable { + Object[][] unexpectedArguments = new Object[][] { + null, + new Object[0], + new Object[] {null}, + new Object[] {"not-a-collection"} + }; + + for (Object[] arguments : unexpectedArguments) { + ContextManager.createEntrySpan("parent", new ContextCarrier()); + + interceptor.beforeMethod(enhancedInstance, invokeAllMethod(), arguments, null, result); + interceptor.handleMethodException( + enhancedInstance, invokeAllMethod(), arguments, null, new IllegalStateException("ignored")); + interceptor.afterMethod(enhancedInstance, invokeAllMethod(), arguments, null, null); + + assertThat(ContextManager.isActive(), is(true)); + ContextManager.stopSpan(); + } + + assertThat(segmentStorage.getTraceSegments().size(), is(4)); + for (TraceSegment traceSegment : segmentStorage.getTraceSegments()) { + List spans = SegmentHelper.getSpans(traceSegment); + assertThat(spans.size(), is(1)); + assertThat(spans.get(0).getOperationName(), is("parent")); + SpanAssert.assertOccurException(spans.get(0), false); + } + } + + @Test + public void shouldTraceEmptyCollection() throws Throwable { + Object[] arguments = new Object[] {Collections.emptyList()}; + ContextManager.createEntrySpan("parent", new ContextCarrier()); + + interceptor.beforeMethod(enhancedInstance, invokeAllMethod(), arguments, null, result); + interceptor.afterMethod(enhancedInstance, invokeAllMethod(), arguments, null, null); + ContextManager.stopSpan(); + + List spans = SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0)); + assertThat(spans.size(), is(2)); + assertThat(spans.get(0).getOperationName(), is("ThreadPoolExecutor/invokeAll")); + } + + @Test + public void shouldTraceAlternativeInvokeAllSignature() throws Throwable { + Object[] arguments = new Object[] {Collections.emptyList(), "alternative"}; + ContextManager.createEntrySpan("parent", new ContextCarrier()); + + interceptor.beforeMethod(enhancedInstance, alternativeInvokeAllMethod(), arguments, null, result); + interceptor.afterMethod(enhancedInstance, alternativeInvokeAllMethod(), arguments, null, null); + ContextManager.stopSpan(); + + List spans = SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0)); + assertThat(spans.size(), is(2)); + assertThat(spans.get(0).getOperationName(), is("ThreadPoolExecutor/invokeAll")); + } + + private Method invokeAllMethod() throws NoSuchMethodException { + return AbstractExecutorService.class.getMethod("invokeAll", Collection.class); + } + + private Method alternativeInvokeAllMethod() throws NoSuchMethodException { + return getClass().getDeclaredMethod("invokeAll", Collection.class, String.class); + } + + private void invokeAll(Collection callables, String alternative) { + } +} diff --git a/test/plugin/scenarios/jdk-threadpool-scenario/config/expectedData.yaml b/test/plugin/scenarios/jdk-threadpool-scenario/config/expectedData.yaml index 21c509cc2f..2c49d15499 100644 --- a/test/plugin/scenarios/jdk-threadpool-scenario/config/expectedData.yaml +++ b/test/plugin/scenarios/jdk-threadpool-scenario/config/expectedData.yaml @@ -15,7 +15,7 @@ # limitations under the License. segmentItems: - serviceName: jdk-threadpool-scenario - segmentSize: ge 14 + segmentSize: ge 20 segments: - segmentId: not null spans: @@ -344,6 +344,54 @@ segmentItems: traceId: not null} - segmentId: not null spans: + - operationName: ThreadPoolExecutor/invokeAny + operationId: 0 + parentSpanId: 0 + spanId: 4 + spanLayer: Unknown + startTime: nq 0 + endTime: nq 0 + componentId: 80 + isError: false + spanType: Local + peer: '' + skipAnalysis: false + - operationName: ThreadPoolExecutor/invokeAny + operationId: 0 + parentSpanId: 0 + spanId: 3 + spanLayer: Unknown + startTime: nq 0 + endTime: nq 0 + componentId: 80 + isError: false + spanType: Local + peer: '' + skipAnalysis: false + - operationName: ThreadPoolExecutor/invokeAll + operationId: 0 + parentSpanId: 0 + spanId: 2 + spanLayer: Unknown + startTime: nq 0 + endTime: nq 0 + componentId: 80 + isError: false + spanType: Local + peer: '' + skipAnalysis: false + - operationName: ThreadPoolExecutor/invokeAll + operationId: 0 + parentSpanId: 0 + spanId: 1 + spanLayer: Unknown + startTime: nq 0 + endTime: nq 0 + componentId: 80 + isError: false + spanType: Local + peer: '' + skipAnalysis: false - operationName: GET:/greet/{username} operationId: 0 parentSpanId: -1 @@ -360,6 +408,120 @@ segmentItems: - {key: url, value: 'http://localhost:8080/greet/skywalking'} - {key: http.method, value: GET} - {key: http.status_code, value: '200'} + - segmentId: not null + spans: + - operationName: SwCallableWrapper/batch-callable-thread + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Unknown + startTime: nq 0 + endTime: nq 0 + componentId: 80 + isError: false + spanType: Local + peer: '' + skipAnalysis: false + refs: + - {parentEndpoint: 'GET:/greet/{username}', networkAddress: '', refType: CrossThread, + parentSpanId: not null, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: jdk-threadpool-scenario, + traceId: not null} + - segmentId: not null + spans: + - operationName: SwCallableWrapper/batch-callable-thread + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Unknown + startTime: nq 0 + endTime: nq 0 + componentId: 80 + isError: false + spanType: Local + peer: '' + skipAnalysis: false + refs: + - {parentEndpoint: 'GET:/greet/{username}', networkAddress: '', refType: CrossThread, + parentSpanId: not null, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: jdk-threadpool-scenario, + traceId: not null} + - segmentId: not null + spans: + - operationName: SwCallableWrapper/batch-callable-thread + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Unknown + startTime: nq 0 + endTime: nq 0 + componentId: 80 + isError: false + spanType: Local + peer: '' + skipAnalysis: false + refs: + - {parentEndpoint: 'GET:/greet/{username}', networkAddress: '', refType: CrossThread, + parentSpanId: not null, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: jdk-threadpool-scenario, + traceId: not null} + - segmentId: not null + spans: + - operationName: SwCallableWrapper/batch-callable-thread + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Unknown + startTime: nq 0 + endTime: nq 0 + componentId: 80 + isError: false + spanType: Local + peer: '' + skipAnalysis: false + refs: + - {parentEndpoint: 'GET:/greet/{username}', networkAddress: '', refType: CrossThread, + parentSpanId: not null, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: jdk-threadpool-scenario, + traceId: not null} + - segmentId: not null + spans: + - operationName: SwCallableWrapper/batch-callable-thread + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Unknown + startTime: nq 0 + endTime: nq 0 + componentId: 80 + isError: false + spanType: Local + peer: '' + skipAnalysis: false + refs: + - {parentEndpoint: 'GET:/greet/{username}', networkAddress: '', refType: CrossThread, + parentSpanId: not null, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: jdk-threadpool-scenario, + traceId: not null} + - segmentId: not null + spans: + - operationName: SwCallableWrapper/batch-callable-thread + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Unknown + startTime: nq 0 + endTime: nq 0 + componentId: 80 + isError: false + spanType: Local + peer: '' + skipAnalysis: false + refs: + - {parentEndpoint: 'GET:/greet/{username}', networkAddress: '', refType: CrossThread, + parentSpanId: not null, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: jdk-threadpool-scenario, + traceId: not null} - segmentId: not null spans: - operationName: GET:/threadpool diff --git a/test/plugin/scenarios/jdk-threadpool-scenario/src/main/java/test/apache/skywalking/apm/testcase/jdk/threading/Application.java b/test/plugin/scenarios/jdk-threadpool-scenario/src/main/java/test/apache/skywalking/apm/testcase/jdk/threading/Application.java index 094c0aed0e..73d3aec4b1 100644 --- a/test/plugin/scenarios/jdk-threadpool-scenario/src/main/java/test/apache/skywalking/apm/testcase/jdk/threading/Application.java +++ b/test/plugin/scenarios/jdk-threadpool-scenario/src/main/java/test/apache/skywalking/apm/testcase/jdk/threading/Application.java @@ -17,11 +17,19 @@ package test.apache.skywalking.apm.testcase.jdk.threading; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @@ -48,6 +56,7 @@ static class TestController { private final RestTemplate restTemplate; private final ExecutorService executorService; private final ExecutorService executorService2; + private final ExecutorService executorService3; public TestController(final RestTemplate restTemplate) { this.restTemplate = restTemplate; @@ -67,6 +76,15 @@ public Thread newThread(Runnable r) { return thread; } }); + this.executorService3 = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS, + new LinkedBlockingQueue(), new ThreadFactory() { + @Override + public Thread newThread(Runnable r) { + Thread thread = new Thread(r); + thread.setName("batch-callable-thread"); + return thread; + } + }); } @GetMapping("/healthCheck") @@ -75,7 +93,8 @@ public String healthCheck() { } @GetMapping("/greet/{username}") - public String testCase(@PathVariable final String username) throws ExecutionException, InterruptedException { + public String testCase(@PathVariable final String username) + throws ExecutionException, InterruptedException, TimeoutException { Runnable runnable = new Runnable() { @Override public void run() { @@ -98,9 +117,36 @@ public String call() { executorService2.submit(runnable); executorService2.submit(callable).get(); + Callable firstCallable = new Callable() { + @Override + public String call() { + return "first"; + } + }; + Callable secondCallable = new Callable() { + @Override + public String call() { + return "second"; + } + }; + List> invokeAllCallables = Collections.unmodifiableList( + Arrays.asList(firstCallable, secondCallable)); + verifyInvokeAllResults(executorService3.invokeAll(invokeAllCallables)); + verifyInvokeAllResults(executorService3.invokeAll(invokeAllCallables, 10, TimeUnit.SECONDS)); + + List> invokeAnyCallables = Collections.singletonList(firstCallable); + executorService3.invokeAny(invokeAnyCallables); + executorService3.invokeAny(invokeAnyCallables, 10, TimeUnit.SECONDS); + return username; } + private void verifyInvokeAllResults(List> results) throws ExecutionException, InterruptedException { + if (!"first".equals(results.get(0).get()) || !"second".equals(results.get(1).get())) { + throw new IllegalStateException("invokeAll results do not preserve task iteration order"); + } + } + @GetMapping("/threadpool") public String threadpool() { return "threadpool"; From bcf709269fab4b66abefb2801f96efa5bcd6ed19 Mon Sep 17 00:00:00 2001 From: xuzhiguang <13520773230@163.com> Date: Fri, 31 Jul 2026 19:29:28 +0800 Subject: [PATCH 2/2] fix(threadpool): avoid duplicate invoke spans Match invokeAll and invokeAny only on ThreadPoolExecutor and use MethodInvocationContext to carry the invocation span lifecycle. --- .../ThreadPoolInvokeMethodInterceptor.java | 23 ++- .../ThreadPoolExecutorInstrumentation.java | 23 --- ...readPoolExecutorInvokeInstrumentation.java | 83 ++++++++ .../src/main/resources/skywalking-plugin.def | 1 + ...ThreadPoolInvokeMethodInterceptorTest.java | 182 +++++++++++++++--- .../config/expectedData.yaml | 16 +- .../testcase/jdk/threading/Application.java | 10 +- 7 files changed, 268 insertions(+), 70 deletions(-) create mode 100644 apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInvokeInstrumentation.java diff --git a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptor.java b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptor.java index a9f7fbc494..eee82d491e 100644 --- a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptor.java +++ b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptor.java @@ -27,24 +27,25 @@ import org.apache.skywalking.apm.agent.core.context.ContextSnapshot; import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; -import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; -import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.v2.InstanceMethodsAroundInterceptorV2; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.v2.MethodInvocationContext; import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; import org.apache.skywalking.apm.plugin.wrapper.SwCallableWrapper; -public class ThreadPoolInvokeMethodInterceptor implements InstanceMethodsAroundInterceptor { +public class ThreadPoolInvokeMethodInterceptor implements InstanceMethodsAroundInterceptorV2 { private static final String OPERATION_NAME_PREFIX = "ThreadPoolExecutor/"; @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, - MethodInterceptResult result) throws Throwable { + MethodInvocationContext context) throws Throwable { if (!shouldEnhance(allArguments)) { return; } AbstractSpan span = ContextManager.createLocalSpan(OPERATION_NAME_PREFIX + method.getName()); span.setComponent(ComponentsDefine.JDK_THREADING); + context.setContext(span); ContextSnapshot contextSnapshot = ContextManager.capture(); Collection callables = (Collection) allArguments[0]; @@ -57,18 +58,20 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, - Object ret) throws Throwable { - if (shouldEnhance(allArguments)) { - ContextManager.stopSpan(); + Object ret, MethodInvocationContext context) throws Throwable { + AbstractSpan span = (AbstractSpan) context.getContext(); + if (span != null) { + ContextManager.stopSpan(span); } return ret; } @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, - Class[] argumentsTypes, Throwable t) { - if (shouldEnhance(allArguments)) { - ContextManager.activeSpan().log(t); + Class[] argumentsTypes, Throwable t, MethodInvocationContext context) { + AbstractSpan span = (AbstractSpan) context.getContext(); + if (span != null) { + span.log(t); } } diff --git a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInstrumentation.java b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInstrumentation.java index a895d61d53..d057a8f317 100644 --- a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInstrumentation.java +++ b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInstrumentation.java @@ -37,16 +37,10 @@ public class ThreadPoolExecutorInstrumentation extends ClassInstanceMethodsEnhan private static final String INTERCEPT_SUBMIT_METHOD = "submit"; - private static final String INTERCEPT_INVOKE_ALL_METHOD = "invokeAll"; - - private static final String INTERCEPT_INVOKE_ANY_METHOD = "invokeAny"; - private static final String INTERCEPT_EXECUTE_METHOD_HANDLE = "org.apache.skywalking.apm.plugin.ThreadPoolExecuteMethodInterceptor"; private static final String INTERCEPT_SUBMIT_METHOD_HANDLE = "org.apache.skywalking.apm.plugin.ThreadPoolSubmitMethodInterceptor"; - private static final String INTERCEPT_INVOKE_METHOD_HANDLE = "org.apache.skywalking.apm.plugin.ThreadPoolInvokeMethodInterceptor"; - @Override public boolean isBootstrapInstrumentation() { return true; @@ -92,23 +86,6 @@ public String getMethodsInterceptor() { return INTERCEPT_SUBMIT_METHOD_HANDLE; } - @Override - public boolean isOverrideArgs() { - return true; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - return ElementMatchers.named(INTERCEPT_INVOKE_ALL_METHOD) - .or(ElementMatchers.named(INTERCEPT_INVOKE_ANY_METHOD)); - } - - @Override - public String getMethodsInterceptor() { - return INTERCEPT_INVOKE_METHOD_HANDLE; - } - @Override public boolean isOverrideArgs() { return true; diff --git a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInvokeInstrumentation.java b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInvokeInstrumentation.java new file mode 100644 index 0000000000..aad4ff1939 --- /dev/null +++ b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/define/ThreadPoolExecutorInvokeInstrumentation.java @@ -0,0 +1,83 @@ +/* + * 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 org.apache.skywalking.apm.plugin.define; + +import java.util.Collection; +import java.util.concurrent.TimeUnit; +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; +import net.bytebuddy.matcher.ElementMatchers; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.v2.ClassInstanceMethodsEnhancePluginDefineV2; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.v2.InstanceMethodsInterceptV2Point; +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; +import org.apache.skywalking.apm.agent.core.plugin.match.NameMatch; + +public class ThreadPoolExecutorInvokeInstrumentation extends ClassInstanceMethodsEnhancePluginDefineV2 { + + private static final String ENHANCE_CLASS = "java.util.concurrent.ThreadPoolExecutor"; + + private static final String INTERCEPT_INVOKE_ALL_METHOD = "invokeAll"; + + private static final String INTERCEPT_INVOKE_ANY_METHOD = "invokeAny"; + + private static final String INTERCEPT_INVOKE_METHOD_HANDLE = + "org.apache.skywalking.apm.plugin.ThreadPoolInvokeMethodInterceptor"; + + @Override + public boolean isBootstrapInstrumentation() { + return true; + } + + @Override + protected ClassMatch enhanceClass() { + return NameMatch.byName(ENHANCE_CLASS); + } + + @Override + public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[0]; + } + + @Override + public InstanceMethodsInterceptV2Point[] getInstanceMethodsInterceptV2Points() { + return new InstanceMethodsInterceptV2Point[] { + new InstanceMethodsInterceptV2Point() { + @Override + public ElementMatcher getMethodsMatcher() { + return ElementMatchers.named(INTERCEPT_INVOKE_ALL_METHOD) + .or(ElementMatchers.named(INTERCEPT_INVOKE_ANY_METHOD)) + .and(ElementMatchers.takesArguments(Collection.class) + .or(ElementMatchers.takesArguments( + Collection.class, long.class, TimeUnit.class))); + } + + @Override + public String getMethodsInterceptorV2() { + return INTERCEPT_INVOKE_METHOD_HANDLE; + } + + @Override + public boolean isOverrideArgs() { + return true; + } + } + }; + } +} diff --git a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/resources/skywalking-plugin.def index 2162787cd1..fe74b21a99 100644 --- a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/resources/skywalking-plugin.def +++ b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/resources/skywalking-plugin.def @@ -15,3 +15,4 @@ # limitations under the License. jdk-threadpool-plugin=org.apache.skywalking.apm.plugin.define.ThreadPoolExecutorInstrumentation +jdk-threadpool-plugin=org.apache.skywalking.apm.plugin.define.ThreadPoolExecutorInvokeInstrumentation diff --git a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/test/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptorTest.java b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/test/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptorTest.java index 319a1da528..13ad5de9cb 100644 --- a/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/test/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptorTest.java +++ b/apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/test/java/org/apache/skywalking/apm/plugin/ThreadPoolInvokeMethodInterceptorTest.java @@ -17,25 +17,32 @@ package org.apache.skywalking.apm.plugin; +import static org.hamcrest.CoreMatchers.instanceOf; import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.CoreMatchers.sameInstance; import static org.hamcrest.MatcherAssert.assertThat; import java.lang.reflect.Method; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.concurrent.AbstractExecutorService; +import java.util.concurrent.Callable; +import java.util.concurrent.TimeUnit; import org.apache.skywalking.apm.agent.core.context.ContextCarrier; import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.ContextSnapshot; import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan; import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; -import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.v2.MethodInvocationContext; import org.apache.skywalking.apm.agent.test.helper.SegmentHelper; import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule; import org.apache.skywalking.apm.agent.test.tools.SegmentStorage; import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint; import org.apache.skywalking.apm.agent.test.tools.SpanAssert; import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner; +import org.apache.skywalking.apm.plugin.wrapper.SwCallableWrapper; import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; @@ -53,13 +60,86 @@ public class ThreadPoolInvokeMethodInterceptorTest { @Mock private EnhancedInstance enhancedInstance; - @Mock - private MethodInterceptResult result; - private final ThreadPoolInvokeMethodInterceptor interceptor = new ThreadPoolInvokeMethodInterceptor(); @Test - public void shouldIgnoreUnexpectedArguments() throws Throwable { + public void shouldWrapCallablesInOrderWithoutMutatingOriginalCollection() throws Throwable { + Callable first = new StringCallable("first"); + Callable second = new StringCallable("second"); + List> original = Collections.unmodifiableList(Arrays.asList(first, second)); + Object[] arguments = new Object[] {original}; + MethodInvocationContext context = new MethodInvocationContext(); + ContextManager.createEntrySpan("parent", new ContextCarrier()); + + interceptor.beforeMethod(enhancedInstance, invokeAllMethod(), arguments, null, context); + + assertThat(arguments[0] == original, is(false)); + assertThat(arguments[0], instanceOf(List.class)); + List wrapped = (List) arguments[0]; + assertThat(wrapped.size(), is(2)); + assertThat(wrapped.get(0), instanceOf(SwCallableWrapper.class)); + assertThat(wrapped.get(1), instanceOf(SwCallableWrapper.class)); + assertThat(original.get(0), sameInstance(first)); + assertThat(original.get(1), sameInstance(second)); + + interceptor.afterMethod(enhancedInstance, invokeAllMethod(), arguments, null, null, context); + assertThat(ContextManager.isActive(), is(true)); + ContextManager.stopSpan(); + + assertInvocationSpan("ThreadPoolExecutor/invokeAll", false); + } + + @Test + public void shouldSkipCallablesThatAlreadyCarryTracingContext() throws Throwable { + ContextManager.createEntrySpan("parent", new ContextCarrier()); + Callable delegate = new StringCallable("wrapped"); + SwCallableWrapper alreadyWrapped = new SwCallableWrapper(delegate, ContextManager.capture()); + CapturedCallable capturedCallable = new CapturedCallable(ContextManager.capture()); + Object[] arguments = new Object[] {Arrays.asList(alreadyWrapped, capturedCallable)}; + MethodInvocationContext context = new MethodInvocationContext(); + + interceptor.beforeMethod(enhancedInstance, invokeAnyMethod(), arguments, null, context); + + List wrapped = (List) arguments[0]; + assertThat(wrapped.get(0), sameInstance((Object) alreadyWrapped)); + assertThat(wrapped.get(1), sameInstance((Object) capturedCallable)); + + interceptor.afterMethod(enhancedInstance, invokeAnyMethod(), arguments, null, null, context); + ContextManager.stopSpan(); + + assertInvocationSpan("ThreadPoolExecutor/invokeAny", false); + } + + @Test + public void shouldTraceAllInvokeOverloads() throws Throwable { + Method[] methods = new Method[] { + invokeAllMethod(), timedInvokeAllMethod(), invokeAnyMethod(), timedInvokeAnyMethod() + }; + String[] operationNames = new String[] { + "ThreadPoolExecutor/invokeAll", "ThreadPoolExecutor/invokeAll", + "ThreadPoolExecutor/invokeAny", "ThreadPoolExecutor/invokeAny" + }; + + for (Method method : methods) { + Object[] arguments = method.getParameterTypes().length == 1 + ? new Object[] {Collections.emptyList()} + : new Object[] {Collections.emptyList(), 1L, TimeUnit.SECONDS}; + MethodInvocationContext context = new MethodInvocationContext(); + ContextManager.createEntrySpan("parent", new ContextCarrier()); + + interceptor.beforeMethod(enhancedInstance, method, arguments, method.getParameterTypes(), context); + interceptor.afterMethod(enhancedInstance, method, arguments, method.getParameterTypes(), null, context); + ContextManager.stopSpan(); + } + + assertThat(segmentStorage.getTraceSegments().size(), is(4)); + for (int i = 0; i < operationNames.length; i++) { + assertInvocationSpan(segmentStorage.getTraceSegments().get(i), operationNames[i], false); + } + } + + @Test + public void shouldIgnoreUnexpectedArgumentsWithoutStoppingParentSpan() throws Throwable { Object[][] unexpectedArguments = new Object[][] { null, new Object[0], @@ -68,13 +148,15 @@ public void shouldIgnoreUnexpectedArguments() throws Throwable { }; for (Object[] arguments : unexpectedArguments) { + MethodInvocationContext context = new MethodInvocationContext(); ContextManager.createEntrySpan("parent", new ContextCarrier()); - interceptor.beforeMethod(enhancedInstance, invokeAllMethod(), arguments, null, result); + interceptor.beforeMethod(enhancedInstance, invokeAllMethod(), arguments, null, context); interceptor.handleMethodException( - enhancedInstance, invokeAllMethod(), arguments, null, new IllegalStateException("ignored")); - interceptor.afterMethod(enhancedInstance, invokeAllMethod(), arguments, null, null); + enhancedInstance, invokeAllMethod(), arguments, null, new IllegalStateException("ignored"), context); + interceptor.afterMethod(enhancedInstance, invokeAllMethod(), arguments, null, null, context); + assertThat(context.getContext(), is((Object) null)); assertThat(ContextManager.isActive(), is(true)); ContextManager.stopSpan(); } @@ -89,41 +171,85 @@ public void shouldIgnoreUnexpectedArguments() throws Throwable { } @Test - public void shouldTraceEmptyCollection() throws Throwable { - Object[] arguments = new Object[] {Collections.emptyList()}; + public void shouldLogExceptionOnInvocationSpan() throws Throwable { + Object[] arguments = new Object[] {Collections.singletonList(new StringCallable("callable"))}; + MethodInvocationContext context = new MethodInvocationContext(); ContextManager.createEntrySpan("parent", new ContextCarrier()); - interceptor.beforeMethod(enhancedInstance, invokeAllMethod(), arguments, null, result); - interceptor.afterMethod(enhancedInstance, invokeAllMethod(), arguments, null, null); + interceptor.beforeMethod(enhancedInstance, invokeAnyMethod(), arguments, null, context); + interceptor.handleMethodException( + enhancedInstance, invokeAnyMethod(), arguments, null, new IllegalStateException("test"), context); + interceptor.afterMethod(enhancedInstance, invokeAnyMethod(), arguments, null, null, context); + + assertThat(ContextManager.isActive(), is(true)); ContextManager.stopSpan(); - List spans = SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0)); - assertThat(spans.size(), is(2)); - assertThat(spans.get(0).getOperationName(), is("ThreadPoolExecutor/invokeAll")); + assertInvocationSpan("ThreadPoolExecutor/invokeAny", true); } - @Test - public void shouldTraceAlternativeInvokeAllSignature() throws Throwable { - Object[] arguments = new Object[] {Collections.emptyList(), "alternative"}; - ContextManager.createEntrySpan("parent", new ContextCarrier()); - - interceptor.beforeMethod(enhancedInstance, alternativeInvokeAllMethod(), arguments, null, result); - interceptor.afterMethod(enhancedInstance, alternativeInvokeAllMethod(), arguments, null, null); - ContextManager.stopSpan(); + private void assertInvocationSpan(String operationName, boolean hasException) { + assertInvocationSpan(segmentStorage.getTraceSegments().get(0), operationName, hasException); + } - List spans = SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0)); + private void assertInvocationSpan(TraceSegment traceSegment, String operationName, boolean hasException) { + List spans = SegmentHelper.getSpans(traceSegment); assertThat(spans.size(), is(2)); - assertThat(spans.get(0).getOperationName(), is("ThreadPoolExecutor/invokeAll")); + assertThat(spans.get(0).getOperationName(), is(operationName)); + SpanAssert.assertOccurException(spans.get(0), hasException); + assertThat(spans.get(1).getOperationName(), is("parent")); } private Method invokeAllMethod() throws NoSuchMethodException { return AbstractExecutorService.class.getMethod("invokeAll", Collection.class); } - private Method alternativeInvokeAllMethod() throws NoSuchMethodException { - return getClass().getDeclaredMethod("invokeAll", Collection.class, String.class); + private Method timedInvokeAllMethod() throws NoSuchMethodException { + return AbstractExecutorService.class.getMethod( + "invokeAll", Collection.class, long.class, TimeUnit.class); + } + + private Method invokeAnyMethod() throws NoSuchMethodException { + return AbstractExecutorService.class.getMethod("invokeAny", Collection.class); + } + + private Method timedInvokeAnyMethod() throws NoSuchMethodException { + return AbstractExecutorService.class.getMethod( + "invokeAny", Collection.class, long.class, TimeUnit.class); } - private void invokeAll(Collection callables, String alternative) { + private static class StringCallable implements Callable { + private final String value; + + private StringCallable(String value) { + this.value = value; + } + + @Override + public String call() { + return value; + } + } + + private static class CapturedCallable implements Callable, EnhancedInstance { + private Object dynamicField; + + private CapturedCallable(ContextSnapshot contextSnapshot) { + this.dynamicField = contextSnapshot; + } + + @Override + public String call() { + return "captured"; + } + + @Override + public Object getSkyWalkingDynamicField() { + return dynamicField; + } + + @Override + public void setSkyWalkingDynamicField(Object value) { + this.dynamicField = value; + } } } diff --git a/test/plugin/scenarios/jdk-threadpool-scenario/config/expectedData.yaml b/test/plugin/scenarios/jdk-threadpool-scenario/config/expectedData.yaml index 2c49d15499..285049878b 100644 --- a/test/plugin/scenarios/jdk-threadpool-scenario/config/expectedData.yaml +++ b/test/plugin/scenarios/jdk-threadpool-scenario/config/expectedData.yaml @@ -344,10 +344,10 @@ segmentItems: traceId: not null} - segmentId: not null spans: - - operationName: ThreadPoolExecutor/invokeAny + - operationName: ThreadPoolExecutor/invokeAll operationId: 0 parentSpanId: 0 - spanId: 4 + spanId: 1 spanLayer: Unknown startTime: nq 0 endTime: nq 0 @@ -356,10 +356,10 @@ segmentItems: spanType: Local peer: '' skipAnalysis: false - - operationName: ThreadPoolExecutor/invokeAny + - operationName: ThreadPoolExecutor/invokeAll operationId: 0 parentSpanId: 0 - spanId: 3 + spanId: 2 spanLayer: Unknown startTime: nq 0 endTime: nq 0 @@ -368,10 +368,10 @@ segmentItems: spanType: Local peer: '' skipAnalysis: false - - operationName: ThreadPoolExecutor/invokeAll + - operationName: ThreadPoolExecutor/invokeAny operationId: 0 parentSpanId: 0 - spanId: 2 + spanId: 3 spanLayer: Unknown startTime: nq 0 endTime: nq 0 @@ -380,10 +380,10 @@ segmentItems: spanType: Local peer: '' skipAnalysis: false - - operationName: ThreadPoolExecutor/invokeAll + - operationName: ThreadPoolExecutor/invokeAny operationId: 0 parentSpanId: 0 - spanId: 1 + spanId: 4 spanLayer: Unknown startTime: nq 0 endTime: nq 0 diff --git a/test/plugin/scenarios/jdk-threadpool-scenario/src/main/java/test/apache/skywalking/apm/testcase/jdk/threading/Application.java b/test/plugin/scenarios/jdk-threadpool-scenario/src/main/java/test/apache/skywalking/apm/testcase/jdk/threading/Application.java index 73d3aec4b1..49cdfbb374 100644 --- a/test/plugin/scenarios/jdk-threadpool-scenario/src/main/java/test/apache/skywalking/apm/testcase/jdk/threading/Application.java +++ b/test/plugin/scenarios/jdk-threadpool-scenario/src/main/java/test/apache/skywalking/apm/testcase/jdk/threading/Application.java @@ -53,6 +53,14 @@ public RestTemplate restTemplate() { @RestController static class TestController { + private static class NestedThreadPoolExecutor extends ThreadPoolExecutor { + private NestedThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, + TimeUnit unit, LinkedBlockingQueue workQueue, + ThreadFactory threadFactory) { + super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory); + } + } + private final RestTemplate restTemplate; private final ExecutorService executorService; private final ExecutorService executorService2; @@ -76,7 +84,7 @@ public Thread newThread(Runnable r) { return thread; } }); - this.executorService3 = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS, + this.executorService3 = new NestedThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), new ThreadFactory() { @Override public Thread newThread(Runnable r) {