From 7ec97a0bcbf13aa698d6330dd86860dbe759276b Mon Sep 17 00:00:00 2001 From: Ayush0612005 Date: Mon, 21 Sep 2026 16:35:38 +0530 Subject: [PATCH 1/2] Fix HttpClient 5.x callback span handling --- CHANGES.md | 1 + .../apm/plugin/httpclient/v5/Constants.java | 2 + .../HttpAsyncClientDoExecuteInterceptor.java | 2 +- .../v5/IOSessionImplPollInterceptor.java | 1 + .../apm/plugin/httpclient/v5/OwnedSpans.java | 47 +++++ .../wrapper/AsyncResponseConsumerWrapper.java | 7 + .../v5/wrapper/FutureCallbackWrapper.java | 30 ++- .../v5/FutureCallbackWrapperTest.java | 185 ++++++++++++++++++ 8 files changed, 264 insertions(+), 11 deletions(-) create mode 100644 apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/OwnedSpans.java create mode 100644 apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/FutureCallbackWrapperTest.java diff --git a/CHANGES.md b/CHANGES.md index a175738df6..034cd6a168 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -4,6 +4,7 @@ Release Notes. 9.8.0 ------------------ +* Fix `httpclient-5.x-plugin` closing the caller thread's active span when `FutureCallback` executes on the caller thread (apache/skywalking#14097). * Fix the `spring-ai-1.x-plugin` `ChatModelStreamInterceptor` leaking its async span when `ChatModel#stream(Prompt)` fails synchronously, which silently dropped the whole `TraceSegment` diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/Constants.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/Constants.java index 2497ca8cbc..84bff48dc2 100644 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/Constants.java +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/Constants.java @@ -20,4 +20,6 @@ public class Constants { public static String SKYWALKING_CONTEXT_SNAPSHOT = "skywalking-context-snapshot"; + + public static String SKYWALKING_LOCAL_SPAN = "skywalking-local-span"; } diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientDoExecuteInterceptor.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientDoExecuteInterceptor.java index 68267fcc6f..06b60dc5da 100644 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientDoExecuteInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientDoExecuteInterceptor.java @@ -39,7 +39,7 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr HttpContext context = (HttpContext) allArguments[4]; FutureCallback callback = (FutureCallback) allArguments[5]; allArguments[2] = new AsyncResponseConsumerWrapper(consumer); - allArguments[5] = new FutureCallbackWrapper(callback); + allArguments[5] = new FutureCallbackWrapper(callback, context); if (ContextManager.isActive()) { context.setAttribute(Constants.SKYWALKING_CONTEXT_SNAPSHOT, ContextManager.capture()); } diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/IOSessionImplPollInterceptor.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/IOSessionImplPollInterceptor.java index fc8ef190d4..f28848114e 100644 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/IOSessionImplPollInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/IOSessionImplPollInterceptor.java @@ -62,6 +62,7 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA AbstractSpan localSpan = ContextManager.createLocalSpan("httpasyncclient/local"); localSpan.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT); localSpan.setLayer(SpanLayer.HTTP); + httpContext.setAttribute(Constants.SKYWALKING_LOCAL_SPAN, localSpan); ContextManager.continued(snapshot); final ContextCarrier contextCarrier = new ContextCarrier(); diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/OwnedSpans.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/OwnedSpans.java new file mode 100644 index 0000000000..17cee16443 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/OwnedSpans.java @@ -0,0 +1,47 @@ +/* + * 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.httpclient.v5; + +import org.apache.hc.core5.http.protocol.HttpContext; +import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; + +/** + * The async HttpClient callbacks may run on a thread other than the I/O thread that created the spans, for example + * the business thread when {@code HttpAsyncClients.classic(...)} is used. Only the span created for the request may + * be finished by the plugin, and only when it is the active span of the current thread. + */ +public final class OwnedSpans { + + private OwnedSpans() { + } + + /** + * @return the local span created for this request if, and only if, it is the active span of the current thread. + */ + public static AbstractSpan activeOwnedSpan(HttpContext context) { + if (context == null || !ContextManager.isActive()) { + return null; + } + Object owned = context.getAttribute(Constants.SKYWALKING_LOCAL_SPAN); + if (owned != null && owned == ContextManager.activeSpan()) { + return (AbstractSpan) owned; + } + return null; + } +} diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapper.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapper.java index 9dec7d109a..2a41cfc849 100644 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapper.java +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapper.java @@ -28,6 +28,7 @@ import org.apache.skywalking.apm.agent.core.context.ContextManager; import org.apache.skywalking.apm.agent.core.context.tag.Tags; import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.plugin.httpclient.v5.OwnedSpans; import java.io.IOException; import java.nio.ByteBuffer; @@ -52,6 +53,12 @@ public void consumeResponse(HttpResponse response, EntityDetails entityDetails, span.errorOccurred(); } ContextManager.stopSpan(); + // consumeResponse runs on the I/O thread that created the local span. Finish it here, otherwise it + // would stay on this thread's span stack when the FutureCallback runs on another thread. + AbstractSpan localSpan = OwnedSpans.activeOwnedSpan(context); + if (localSpan != null) { + ContextManager.stopSpan(localSpan); + } } consumer.consumeResponse(response, entityDetails, context, resultCallback); } diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java index f606856edf..5052131a1e 100644 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java @@ -18,20 +18,28 @@ package org.apache.skywalking.apm.plugin.httpclient.v5.wrapper; import org.apache.hc.core5.concurrent.FutureCallback; +import org.apache.hc.core5.http.protocol.HttpContext; import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.plugin.httpclient.v5.OwnedSpans; public class FutureCallbackWrapper implements FutureCallback { - private FutureCallback callback; + private final FutureCallback callback; + private final HttpContext context; - public FutureCallbackWrapper(FutureCallback callback) { + public FutureCallbackWrapper(FutureCallback callback, HttpContext context) { this.callback = callback; + this.context = context; } @Override public void completed(T o) { - if (ContextManager.isActive()) { - ContextManager.stopSpan(); + // The callback may run on the caller thread (e.g. HttpAsyncClients.classic), whose active span + // does not belong to this request. Only finish the span created for this request. + AbstractSpan span = OwnedSpans.activeOwnedSpan(context); + if (span != null) { + ContextManager.stopSpan(span); } if (callback != null) { callback.completed(o); @@ -40,9 +48,10 @@ public void completed(T o) { @Override public void failed(Exception e) { - if (ContextManager.isActive()) { - ContextManager.activeSpan().errorOccurred().log(e); - ContextManager.stopSpan(); + AbstractSpan span = OwnedSpans.activeOwnedSpan(context); + if (span != null) { + span.errorOccurred().log(e); + ContextManager.stopSpan(span); } if (callback != null) { callback.failed(e); @@ -51,9 +60,10 @@ public void failed(Exception e) { @Override public void cancelled() { - if (ContextManager.isActive()) { - ContextManager.activeSpan().errorOccurred(); - ContextManager.stopSpan(); + AbstractSpan span = OwnedSpans.activeOwnedSpan(context); + if (span != null) { + span.errorOccurred(); + ContextManager.stopSpan(span); } if (callback != null) { callback.cancelled(); diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/FutureCallbackWrapperTest.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/FutureCallbackWrapperTest.java new file mode 100644 index 0000000000..7bb6dc6046 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/FutureCallbackWrapperTest.java @@ -0,0 +1,185 @@ +/* + * 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.httpclient.v5; + +import org.apache.hc.core5.concurrent.FutureCallback; +import org.apache.hc.core5.http.HttpResponse; +import org.apache.hc.core5.http.nio.AsyncResponseConsumer; +import org.apache.hc.core5.http.protocol.BasicHttpContext; +import org.apache.hc.core5.http.protocol.HttpContext; +import org.apache.skywalking.apm.agent.core.boot.ServiceManager; +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.AbstractSpan; +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.test.helper.SegmentHelper; +import org.apache.skywalking.apm.agent.test.helper.SpanHelper; +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.TracingSegmentRunner; +import org.apache.skywalking.apm.plugin.httpclient.v5.wrapper.AsyncResponseConsumerWrapper; +import org.apache.skywalking.apm.plugin.httpclient.v5.wrapper.FutureCallbackWrapper; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; + +import java.util.List; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@RunWith(TracingSegmentRunner.class) +public class FutureCallbackWrapperTest { + + @SegmentStoragePoint + private SegmentStorage segmentStorage; + + @Rule + public AgentServiceRule agentServiceRule = new AgentServiceRule(); + @Rule + public MockitoRule rule = MockitoJUnit.rule(); + + @Mock + private FutureCallback delegate; + + @Mock + private AsyncResponseConsumer consumer; + @Mock + private HttpResponse response; + + private HttpContext httpContext; + + @Before + public void setUp() { + ServiceManager.INSTANCE.boot(); + httpContext = new BasicHttpContext(); + } + + @Test + public void completedOnIoThreadStopsOwnedSpan() { + AbstractSpan owned = ContextManager.createLocalSpan("httpasyncclient/local"); + httpContext.setAttribute(Constants.SKYWALKING_LOCAL_SPAN, owned); + + new FutureCallbackWrapper<>(delegate, httpContext).completed("ok"); + + assertThat(ContextManager.isActive(), is(false)); + assertThat(segmentStorage.getTraceSegments().size(), is(1)); + verify(delegate).completed("ok"); + } + + @Test + public void failedOnIoThreadMarksOwnedSpanAsError() { + AbstractSpan owned = ContextManager.createLocalSpan("httpasyncclient/local"); + httpContext.setAttribute(Constants.SKYWALKING_LOCAL_SPAN, owned); + Exception cause = new RuntimeException("boom"); + + new FutureCallbackWrapper<>(delegate, httpContext).failed(cause); + + assertThat(ContextManager.isActive(), is(false)); + assertThat(segmentStorage.getTraceSegments().size(), is(1)); + List spans = SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0)); + assertThat(SpanHelper.getErrorOccurred(spans.get(0)), is(true)); + verify(delegate).failed(cause); + } + + @Test + public void completedOnCallerThreadKeepsCallerSpanActive() throws Exception { + // The request span is created on an "I/O thread", as IOSessionImplPollInterceptor does. + Thread ioThread = new Thread(() -> { + AbstractSpan owned = ContextManager.createLocalSpan("httpasyncclient/local"); + httpContext.setAttribute(Constants.SKYWALKING_LOCAL_SPAN, owned); + }); + ioThread.start(); + ioThread.join(); + + // The callback then runs on the caller/business thread, which has its own active Entry span. + AbstractSpan entry = ContextManager.createEntrySpan("/business", null); + + new FutureCallbackWrapper<>(delegate, httpContext).completed("ok"); + + assertThat(ContextManager.isActive(), is(true)); + assertThat(ContextManager.activeSpan() == entry, is(true)); + verify(delegate).completed("ok"); + + ContextManager.stopSpan(entry); + assertThat(segmentStorage.getTraceSegments().size(), is(1)); + TraceSegment segment = segmentStorage.getTraceSegments().get(0); + assertThat(SegmentHelper.getSpans(segment).size(), is(1)); + } + + /** + * Full lifecycle of the reported scenario: spans are created and finished on the I/O thread, the FutureCallback + * then runs on the business thread which owns an Entry span that must survive. + */ + @Test + public void ioThreadSegmentIsFinishedAndCallerEntrySpanSurvives() throws Exception { + when(response.getCode()).thenReturn(200); + Throwable[] ioError = new Throwable[1]; + Thread ioThread = new Thread(() -> { + try { + // what IOSessionImplPollInterceptor does + AbstractSpan local = ContextManager.createLocalSpan("httpasyncclient/local"); + httpContext.setAttribute(Constants.SKYWALKING_LOCAL_SPAN, local); + ContextManager.createExitSpan("/hello", new ContextCarrier(), "127.0.0.1:8080"); + // response headers arrive on the I/O thread + new AsyncResponseConsumerWrapper<>(consumer).consumeResponse(response, null, httpContext, null); + } catch (Throwable t) { + ioError[0] = t; + } + }); + ioThread.start(); + ioThread.join(); + assertThat(ioError[0] == null, is(true)); + + // local + exit span are both finished, so the I/O thread's segment is complete + assertThat(segmentStorage.getTraceSegments().size(), is(1)); + assertThat(SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0)).size(), is(2)); + + // entity read to EOF on the business thread triggers the callback there + AbstractSpan entry = ContextManager.createEntrySpan("/business", null); + new FutureCallbackWrapper<>(delegate, httpContext).completed("body"); + + assertThat(ContextManager.isActive(), is(true)); + assertThat(ContextManager.activeSpan() == entry, is(true)); + ContextManager.stopSpan(entry); + + assertThat(segmentStorage.getTraceSegments().size(), is(2)); + assertThat(SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(1)).size(), is(1)); + verify(delegate).completed("body"); + } + + @Test + public void nullContextNeverStopsForeignSpan() { + AbstractSpan entry = ContextManager.createEntrySpan("/business", null); + + new FutureCallbackWrapper<>(delegate, null).completed("ok"); + + assertThat(ContextManager.isActive(), is(true)); + ContextManager.stopSpan(entry); + verify(delegate).completed("ok"); + } +} From 13a5e173f84e92d5e2a11a01aaec0329665614a4 Mon Sep 17 00:00:00 2001 From: Ayush0612005 Date: Tue, 22 Sep 2026 21:34:13 +0530 Subject: [PATCH 2/2] Fix HttpClient 5.x callback span handling --- .../plugin/httpclient/v5/AsyncExitSpan.java | 91 ++++++++++++ .../apm/plugin/httpclient/v5/Constants.java | 2 - .../HttpAsyncClientDoExecuteInterceptor.java | 31 ++-- .../v5/IOSessionImplPollInterceptor.java | 93 ------------ .../apm/plugin/httpclient/v5/OwnedSpans.java | 47 ------ .../wrapper/AsyncRequestProducerWrapper.java | 125 ++++++++++++++++ .../wrapper/AsyncResponseConsumerWrapper.java | 80 +++++----- .../v5/wrapper/FutureCallbackWrapper.java | 35 ++--- .../src/main/resources/skywalking-plugin.def | 1 - .../v5/FutureCallbackWrapperTest.java | 139 ++++++------------ 10 files changed, 322 insertions(+), 322 deletions(-) create mode 100644 apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncExitSpan.java delete mode 100644 apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/IOSessionImplPollInterceptor.java delete mode 100644 apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/OwnedSpans.java create mode 100644 apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncRequestProducerWrapper.java diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncExitSpan.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncExitSpan.java new file mode 100644 index 0000000000..2565b2579c --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncExitSpan.java @@ -0,0 +1,91 @@ +/* + * 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.httpclient.v5; + +import org.apache.hc.core5.http.HttpHost; +import org.apache.skywalking.apm.agent.core.context.tag.Tags; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; + +public class AsyncExitSpan { + private final HttpHost target; + private volatile Thread creator = Thread.currentThread(); + private AbstractSpan span; + + public AsyncExitSpan(HttpHost target) { + this.target = target; + } + + public HttpHost getTarget() { + return target; + } + + public boolean claimCreation() { + if (creator != Thread.currentThread()) { + return false; + } + creator = null; + return true; + } + + public void callerReturned() { + creator = null; + } + + public synchronized void start(AbstractSpan span) { + this.span = span; + } + + public synchronized void onResponse(int statusCode) { + if (span != null) { + Tags.HTTP_RESPONSE_STATUS_CODE.set(span, statusCode); + if (statusCode >= 400) { + span.errorOccurred(); + } + } + } + + public synchronized void finish() { + end(false, null); + } + + public synchronized void fail(Throwable cause) { + end(true, cause); + } + + public synchronized void abort() { + end(true, null); + } + + private void end(boolean error, Throwable cause) { + if (span == null) { + return; + } + + if (error) { + span.errorOccurred(); + } + + if (cause != null) { + span.log(cause); + } + + span.asyncFinish(); + span = null; + } +} diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/Constants.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/Constants.java index 84bff48dc2..2497ca8cbc 100644 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/Constants.java +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/Constants.java @@ -20,6 +20,4 @@ public class Constants { public static String SKYWALKING_CONTEXT_SNAPSHOT = "skywalking-context-snapshot"; - - public static String SKYWALKING_LOCAL_SPAN = "skywalking-local-span"; } diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientDoExecuteInterceptor.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientDoExecuteInterceptor.java index 06b60dc5da..71e0b1ca2d 100644 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientDoExecuteInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientDoExecuteInterceptor.java @@ -18,42 +18,53 @@ package org.apache.skywalking.apm.plugin.httpclient.v5; +import java.lang.reflect.Method; import org.apache.hc.core5.concurrent.FutureCallback; +import org.apache.hc.core5.http.HttpHost; +import org.apache.hc.core5.http.nio.AsyncRequestProducer; import org.apache.hc.core5.http.nio.AsyncResponseConsumer; -import org.apache.hc.core5.http.protocol.HttpContext; import org.apache.skywalking.apm.agent.core.context.ContextManager; 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.plugin.httpclient.v5.wrapper.AsyncRequestProducerWrapper; import org.apache.skywalking.apm.plugin.httpclient.v5.wrapper.AsyncResponseConsumerWrapper; import org.apache.skywalking.apm.plugin.httpclient.v5.wrapper.FutureCallbackWrapper; -import java.lang.reflect.Method; - public class HttpAsyncClientDoExecuteInterceptor implements InstanceMethodsAroundInterceptor { @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, MethodInterceptResult result) throws Throwable { + if (!ContextManager.isActive()) { + return; + } + + AsyncExitSpan exitSpan = new AsyncExitSpan((HttpHost) allArguments[0]); + + AsyncRequestProducer producer = (AsyncRequestProducer) allArguments[1]; AsyncResponseConsumer consumer = (AsyncResponseConsumer) allArguments[2]; - HttpContext context = (HttpContext) allArguments[4]; FutureCallback callback = (FutureCallback) allArguments[5]; - allArguments[2] = new AsyncResponseConsumerWrapper(consumer); - allArguments[5] = new FutureCallbackWrapper(callback, context); - if (ContextManager.isActive()) { - context.setAttribute(Constants.SKYWALKING_CONTEXT_SNAPSHOT, ContextManager.capture()); - } + + allArguments[1] = new AsyncRequestProducerWrapper(producer, exitSpan); + allArguments[2] = new AsyncResponseConsumerWrapper(consumer, exitSpan); + allArguments[5] = new FutureCallbackWrapper(callback, exitSpan); } @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) throws Throwable { + if (allArguments[1] instanceof AsyncRequestProducerWrapper) { + ((AsyncRequestProducerWrapper) allArguments[1]).getExitSpan().callerReturned(); + } return ret; } @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Throwable t) { - + if (allArguments[1] instanceof AsyncRequestProducerWrapper) { + ((AsyncRequestProducerWrapper) allArguments[1]).getExitSpan().fail(t); + } } } diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/IOSessionImplPollInterceptor.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/IOSessionImplPollInterceptor.java deleted file mode 100644 index f28848114e..0000000000 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/IOSessionImplPollInterceptor.java +++ /dev/null @@ -1,93 +0,0 @@ -/* - * 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.httpclient.v5; - -import org.apache.hc.client5.http.protocol.HttpClientContext; -import org.apache.hc.core5.http.message.BasicHttpRequest; -import org.apache.hc.core5.http.nio.command.RequestExecutionCommand; -import org.apache.hc.core5.http.protocol.HttpContext; -import org.apache.hc.core5.reactor.Command; -import org.apache.skywalking.apm.agent.core.context.CarrierItem; -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.tag.Tags; -import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; -import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; -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 java.lang.reflect.Method; -import java.net.URI; - -public class IOSessionImplPollInterceptor implements InstanceMethodsAroundInterceptor { - - @Override - public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, - MethodInterceptResult result) throws Throwable { - - } - - @Override - public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, - Object ret) throws Throwable { - Command command = (Command) ret; - if (!(command instanceof RequestExecutionCommand)) { - return ret; - } - HttpContext httpContext = ((RequestExecutionCommand) command).getContext(); - ContextSnapshot snapshot = (ContextSnapshot) httpContext.getAttribute(Constants.SKYWALKING_CONTEXT_SNAPSHOT); - if (snapshot == null) { - return ret; - } - httpContext.removeAttribute(Constants.SKYWALKING_CONTEXT_SNAPSHOT); - AbstractSpan localSpan = ContextManager.createLocalSpan("httpasyncclient/local"); - localSpan.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT); - localSpan.setLayer(SpanLayer.HTTP); - httpContext.setAttribute(Constants.SKYWALKING_LOCAL_SPAN, localSpan); - ContextManager.continued(snapshot); - - final ContextCarrier contextCarrier = new ContextCarrier(); - BasicHttpRequest request = (BasicHttpRequest) httpContext.getAttribute(HttpClientContext.HTTP_REQUEST); - URI uri = request.getUri(); - - String operationName = uri.getPath(); - int port = uri.getPort(); - AbstractSpan span = ContextManager - .createExitSpan(operationName, contextCarrier, uri.getHost() + ":" + (port == -1 ? 80 : port)); - span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT); - Tags.URL.set(span, uri.toURL().toString()); - Tags.HTTP.METHOD.set(span, request.getMethod()); - SpanLayer.asHttp(span); - CarrierItem next = contextCarrier.items(); - while (next.hasNext()) { - next = next.next(); - request.setHeader(next.getHeadKey(), next.getHeadValue()); - } - return ret; - } - - @Override - public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, - Class[] argumentsTypes, Throwable t) { - - } -} diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/OwnedSpans.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/OwnedSpans.java deleted file mode 100644 index 17cee16443..0000000000 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/OwnedSpans.java +++ /dev/null @@ -1,47 +0,0 @@ -/* - * 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.httpclient.v5; - -import org.apache.hc.core5.http.protocol.HttpContext; -import org.apache.skywalking.apm.agent.core.context.ContextManager; -import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; - -/** - * The async HttpClient callbacks may run on a thread other than the I/O thread that created the spans, for example - * the business thread when {@code HttpAsyncClients.classic(...)} is used. Only the span created for the request may - * be finished by the plugin, and only when it is the active span of the current thread. - */ -public final class OwnedSpans { - - private OwnedSpans() { - } - - /** - * @return the local span created for this request if, and only if, it is the active span of the current thread. - */ - public static AbstractSpan activeOwnedSpan(HttpContext context) { - if (context == null || !ContextManager.isActive()) { - return null; - } - Object owned = context.getAttribute(Constants.SKYWALKING_LOCAL_SPAN); - if (owned != null && owned == ContextManager.activeSpan()) { - return (AbstractSpan) owned; - } - return null; - } -} diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncRequestProducerWrapper.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncRequestProducerWrapper.java new file mode 100644 index 0000000000..5a81b592e0 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncRequestProducerWrapper.java @@ -0,0 +1,125 @@ +/* + * 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.httpclient.v5.wrapper; + +import java.io.IOException; +import org.apache.hc.core5.http.HttpException; +import org.apache.hc.core5.http.HttpRequest; +import org.apache.hc.core5.http.nio.AsyncRequestProducer; +import org.apache.hc.core5.http.nio.DataStreamChannel; +import org.apache.hc.core5.http.nio.RequestChannel; +import org.apache.hc.core5.http.protocol.HttpContext; +import org.apache.skywalking.apm.agent.core.context.CarrierItem; +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.tag.Tags; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; +import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; +import org.apache.skywalking.apm.plugin.httpclient.v5.AsyncExitSpan; + +public class AsyncRequestProducerWrapper implements AsyncRequestProducer { + + private final AsyncRequestProducer producer; + private final AsyncExitSpan exitSpan; + + public AsyncRequestProducerWrapper(AsyncRequestProducer producer, AsyncExitSpan exitSpan) { + this.producer = producer; + this.exitSpan = exitSpan; + } + + public AsyncExitSpan getExitSpan() { + return exitSpan; + } + + @Override + public void sendRequest(RequestChannel channel, HttpContext context) throws IOException, HttpException { + producer.sendRequest((request, entityDetails, requestContext) -> { + if (exitSpan.claimCreation()) { + try { + startExitSpan(request); + } catch (Throwable ignored) { + // Never let tracing instrumentation break the user's HTTP request. + } + } + + channel.sendRequest(request, entityDetails, requestContext); + }, context); + } + + private void startExitSpan(HttpRequest request) { + String operationName = request.getRequestUri(); + String remotePeer = exitSpan.getTarget().toHostString(); + + ContextCarrier contextCarrier = new ContextCarrier(); + AbstractSpan span = ContextManager.createExitSpan( + operationName, + contextCarrier, + remotePeer + ); + + boolean nested = ContextManager.activeSpan().isExit(); + + if (!nested) { + span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT); + Tags.URL.set(span, request.getRequestUri()); + SpanLayer.asHttp(span); + } + + CarrierItem next = contextCarrier.items(); + while (next.hasNext()) { + request.setHeader(next.getHeadKey(), next.getHeadValue()); + next = next.next(); + } + + if (!nested) { + span.prepareForAsync(); + } + + ContextManager.stopSpan(span); + + if (!nested) { + exitSpan.start(span); + } + } + + @Override + public boolean isRepeatable() { + return producer.isRepeatable(); + } + + @Override + public void produce(DataStreamChannel channel) throws IOException { + producer.produce(channel); + } + + @Override + public int available() { + return producer.available(); + } + + @Override + public void failed(Exception cause) { + producer.failed(cause); + } + + @Override + public void releaseResources() { + producer.releaseResources(); + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapper.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapper.java index 2a41cfc849..cb5b0c361d 100644 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapper.java +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapper.java @@ -17,75 +17,53 @@ package org.apache.skywalking.apm.plugin.httpclient.v5.wrapper; -import org.apache.hc.core5.concurrent.FutureCallback; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.List; import org.apache.hc.core5.http.EntityDetails; -import org.apache.hc.core5.http.Header; import org.apache.hc.core5.http.HttpException; import org.apache.hc.core5.http.HttpResponse; import org.apache.hc.core5.http.nio.AsyncResponseConsumer; import org.apache.hc.core5.http.nio.CapacityChannel; import org.apache.hc.core5.http.protocol.HttpContext; -import org.apache.skywalking.apm.agent.core.context.ContextManager; -import org.apache.skywalking.apm.agent.core.context.tag.Tags; -import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; -import org.apache.skywalking.apm.plugin.httpclient.v5.OwnedSpans; - -import java.io.IOException; -import java.nio.ByteBuffer; -import java.util.List; +import org.apache.hc.core5.http.Header; +import org.apache.hc.core5.concurrent.FutureCallback; +import org.apache.skywalking.apm.plugin.httpclient.v5.AsyncExitSpan; public class AsyncResponseConsumerWrapper implements AsyncResponseConsumer { - private AsyncResponseConsumer consumer; + private final AsyncResponseConsumer consumer; + private final AsyncExitSpan exitSpan; - public AsyncResponseConsumerWrapper(AsyncResponseConsumer consumer) { + public AsyncResponseConsumerWrapper( + AsyncResponseConsumer consumer, AsyncExitSpan exitSpan) { this.consumer = consumer; + this.exitSpan = exitSpan; } @Override - public void consumeResponse(HttpResponse response, EntityDetails entityDetails, HttpContext context, + public void consumeResponse( + HttpResponse response, + EntityDetails entityDetails, + HttpContext context, FutureCallback resultCallback) throws HttpException, IOException { - if (ContextManager.isActive()) { - int statusCode = response.getCode(); - AbstractSpan span = ContextManager.activeSpan(); - Tags.HTTP_RESPONSE_STATUS_CODE.set(span, statusCode); - if (statusCode >= 400) { - span.errorOccurred(); - } - ContextManager.stopSpan(); - // consumeResponse runs on the I/O thread that created the local span. Finish it here, otherwise it - // would stay on this thread's span stack when the FutureCallback runs on another thread. - AbstractSpan localSpan = OwnedSpans.activeOwnedSpan(context); - if (localSpan != null) { - ContextManager.stopSpan(localSpan); - } + + exitSpan.onResponse(response.getCode()); + + if (entityDetails == null) { + exitSpan.finish(); } + consumer.consumeResponse(response, entityDetails, context, resultCallback); } @Override - public void informationResponse(HttpResponse response, HttpContext context) throws HttpException, IOException { - if (ContextManager.isActive()) { - int statusCode = response.getCode(); - AbstractSpan span = ContextManager.activeSpan(); - Tags.HTTP_RESPONSE_STATUS_CODE.set(span, statusCode); - if (statusCode >= 400) { - span.errorOccurred(); - } - ContextManager.stopSpan(); - } + public void informationResponse( + HttpResponse response, + HttpContext context) throws HttpException, IOException { consumer.informationResponse(response, context); } - @Override - public void failed(Exception cause) { - if (ContextManager.isActive()) { - ContextManager.activeSpan().errorOccurred().log(cause); - ContextManager.stopSpan(); - } - consumer.failed(cause); - } - @Override public void updateCapacity(CapacityChannel capacityChannel) throws IOException { consumer.updateCapacity(capacityChannel); @@ -98,11 +76,19 @@ public void consume(ByteBuffer src) throws IOException { @Override public void streamEnd(List trailers) throws HttpException, IOException { + exitSpan.finish(); consumer.streamEnd(trailers); } + @Override + public void failed(Exception cause) { + exitSpan.fail(cause); + consumer.failed(cause); + } + @Override public void releaseResources() { + exitSpan.abort(); consumer.releaseResources(); } -} +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java index 5052131a1e..eb63792c88 100644 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java @@ -18,29 +18,22 @@ package org.apache.skywalking.apm.plugin.httpclient.v5.wrapper; import org.apache.hc.core5.concurrent.FutureCallback; -import org.apache.hc.core5.http.protocol.HttpContext; -import org.apache.skywalking.apm.agent.core.context.ContextManager; -import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; -import org.apache.skywalking.apm.plugin.httpclient.v5.OwnedSpans; +import org.apache.skywalking.apm.plugin.httpclient.v5.AsyncExitSpan; public class FutureCallbackWrapper implements FutureCallback { private final FutureCallback callback; - private final HttpContext context; + private final AsyncExitSpan exitSpan; - public FutureCallbackWrapper(FutureCallback callback, HttpContext context) { + public FutureCallbackWrapper(FutureCallback callback, AsyncExitSpan exitSpan) { this.callback = callback; - this.context = context; + this.exitSpan = exitSpan; } @Override public void completed(T o) { - // The callback may run on the caller thread (e.g. HttpAsyncClients.classic), whose active span - // does not belong to this request. Only finish the span created for this request. - AbstractSpan span = OwnedSpans.activeOwnedSpan(context); - if (span != null) { - ContextManager.stopSpan(span); - } + exitSpan.finish(); + if (callback != null) { callback.completed(o); } @@ -48,11 +41,8 @@ public void completed(T o) { @Override public void failed(Exception e) { - AbstractSpan span = OwnedSpans.activeOwnedSpan(context); - if (span != null) { - span.errorOccurred().log(e); - ContextManager.stopSpan(span); - } + exitSpan.fail(e); + if (callback != null) { callback.failed(e); } @@ -60,13 +50,10 @@ public void failed(Exception e) { @Override public void cancelled() { - AbstractSpan span = OwnedSpans.activeOwnedSpan(context); - if (span != null) { - span.errorOccurred(); - ContextManager.stopSpan(span); - } + exitSpan.abort(); + if (callback != null) { callback.cancelled(); } } -} +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/resources/skywalking-plugin.def index dc6622a88f..63c6348953 100644 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/resources/skywalking-plugin.def +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/resources/skywalking-plugin.def @@ -17,4 +17,3 @@ httpclient-5.x=org.apache.skywalking.apm.plugin.httpclient.v5.define.MinimalHttpClientInstrumentation httpclient-5.x=org.apache.skywalking.apm.plugin.httpclient.v5.define.InternalHttpClientInstrumentation httpclient-5.x=org.apache.skywalking.apm.plugin.httpclient.v5.define.HttpAsyncClientInstrumentation -httpclient-5.x=org.apache.skywalking.apm.plugin.httpclient.v5.define.IOSessionImplInstrumentation diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/FutureCallbackWrapperTest.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/FutureCallbackWrapperTest.java index 7bb6dc6046..386a3cf1f7 100644 --- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/FutureCallbackWrapperTest.java +++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/FutureCallbackWrapperTest.java @@ -19,23 +19,15 @@ package org.apache.skywalking.apm.plugin.httpclient.v5; import org.apache.hc.core5.concurrent.FutureCallback; -import org.apache.hc.core5.http.HttpResponse; -import org.apache.hc.core5.http.nio.AsyncResponseConsumer; -import org.apache.hc.core5.http.protocol.BasicHttpContext; -import org.apache.hc.core5.http.protocol.HttpContext; +import org.apache.hc.core5.http.HttpHost; import org.apache.skywalking.apm.agent.core.boot.ServiceManager; 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.AbstractSpan; -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.test.helper.SegmentHelper; -import org.apache.skywalking.apm.agent.test.helper.SpanHelper; 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.TracingSegmentRunner; -import org.apache.skywalking.apm.plugin.httpclient.v5.wrapper.AsyncResponseConsumerWrapper; import org.apache.skywalking.apm.plugin.httpclient.v5.wrapper.FutureCallbackWrapper; import org.junit.Before; import org.junit.Rule; @@ -45,12 +37,9 @@ import org.mockito.junit.MockitoJUnit; import org.mockito.junit.MockitoRule; -import java.util.List; - import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.MatcherAssert.assertThat; import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; @RunWith(TracingSegmentRunner.class) public class FutureCallbackWrapperTest { @@ -60,126 +49,80 @@ public class FutureCallbackWrapperTest { @Rule public AgentServiceRule agentServiceRule = new AgentServiceRule(); + @Rule public MockitoRule rule = MockitoJUnit.rule(); @Mock private FutureCallback delegate; - @Mock - private AsyncResponseConsumer consumer; - @Mock - private HttpResponse response; - - private HttpContext httpContext; - @Before public void setUp() { ServiceManager.INSTANCE.boot(); - httpContext = new BasicHttpContext(); } - @Test - public void completedOnIoThreadStopsOwnedSpan() { - AbstractSpan owned = ContextManager.createLocalSpan("httpasyncclient/local"); - httpContext.setAttribute(Constants.SKYWALKING_LOCAL_SPAN, owned); - - new FutureCallbackWrapper<>(delegate, httpContext).completed("ok"); + private AsyncExitSpan createStartedExitSpan() { + AsyncExitSpan exitSpan = new AsyncExitSpan( + new HttpHost("http", "127.0.0.1", 8080)); - assertThat(ContextManager.isActive(), is(false)); - assertThat(segmentStorage.getTraceSegments().size(), is(1)); - verify(delegate).completed("ok"); - } - - @Test - public void failedOnIoThreadMarksOwnedSpanAsError() { - AbstractSpan owned = ContextManager.createLocalSpan("httpasyncclient/local"); - httpContext.setAttribute(Constants.SKYWALKING_LOCAL_SPAN, owned); - Exception cause = new RuntimeException("boom"); + AbstractSpan requestSpan = ContextManager.createExitSpan( + "/hello", + new ContextCarrier(), + "127.0.0.1:8080"); - new FutureCallbackWrapper<>(delegate, httpContext).failed(cause); + exitSpan.start(requestSpan); + requestSpan.prepareForAsync(); + ContextManager.stopSpan(requestSpan); - assertThat(ContextManager.isActive(), is(false)); - assertThat(segmentStorage.getTraceSegments().size(), is(1)); - List spans = SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0)); - assertThat(SpanHelper.getErrorOccurred(spans.get(0)), is(true)); - verify(delegate).failed(cause); + return exitSpan; } @Test - public void completedOnCallerThreadKeepsCallerSpanActive() throws Exception { - // The request span is created on an "I/O thread", as IOSessionImplPollInterceptor does. - Thread ioThread = new Thread(() -> { - AbstractSpan owned = ContextManager.createLocalSpan("httpasyncclient/local"); - httpContext.setAttribute(Constants.SKYWALKING_LOCAL_SPAN, owned); - }); - ioThread.start(); - ioThread.join(); + public void completedKeepsCallerSpanActive() { + AbstractSpan callerSpan = ContextManager.createEntrySpan("/business", null); + AsyncExitSpan exitSpan = createStartedExitSpan(); - // The callback then runs on the caller/business thread, which has its own active Entry span. - AbstractSpan entry = ContextManager.createEntrySpan("/business", null); - - new FutureCallbackWrapper<>(delegate, httpContext).completed("ok"); + new FutureCallbackWrapper<>(delegate, exitSpan).completed("ok"); assertThat(ContextManager.isActive(), is(true)); - assertThat(ContextManager.activeSpan() == entry, is(true)); + assertThat(ContextManager.activeSpan() == callerSpan, is(true)); verify(delegate).completed("ok"); - ContextManager.stopSpan(entry); + ContextManager.stopSpan(callerSpan); + assertThat(segmentStorage.getTraceSegments().size(), is(1)); - TraceSegment segment = segmentStorage.getTraceSegments().get(0); - assertThat(SegmentHelper.getSpans(segment).size(), is(1)); } - /** - * Full lifecycle of the reported scenario: spans are created and finished on the I/O thread, the FutureCallback - * then runs on the business thread which owns an Entry span that must survive. - */ @Test - public void ioThreadSegmentIsFinishedAndCallerEntrySpanSurvives() throws Exception { - when(response.getCode()).thenReturn(200); - Throwable[] ioError = new Throwable[1]; - Thread ioThread = new Thread(() -> { - try { - // what IOSessionImplPollInterceptor does - AbstractSpan local = ContextManager.createLocalSpan("httpasyncclient/local"); - httpContext.setAttribute(Constants.SKYWALKING_LOCAL_SPAN, local); - ContextManager.createExitSpan("/hello", new ContextCarrier(), "127.0.0.1:8080"); - // response headers arrive on the I/O thread - new AsyncResponseConsumerWrapper<>(consumer).consumeResponse(response, null, httpContext, null); - } catch (Throwable t) { - ioError[0] = t; - } - }); - ioThread.start(); - ioThread.join(); - assertThat(ioError[0] == null, is(true)); - - // local + exit span are both finished, so the I/O thread's segment is complete - assertThat(segmentStorage.getTraceSegments().size(), is(1)); - assertThat(SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0)).size(), is(2)); + public void failedKeepsCallerSpanActive() { + AbstractSpan callerSpan = ContextManager.createEntrySpan("/business", null); + AsyncExitSpan exitSpan = createStartedExitSpan(); + Exception cause = new RuntimeException("boom"); - // entity read to EOF on the business thread triggers the callback there - AbstractSpan entry = ContextManager.createEntrySpan("/business", null); - new FutureCallbackWrapper<>(delegate, httpContext).completed("body"); + new FutureCallbackWrapper<>(delegate, exitSpan).failed(cause); assertThat(ContextManager.isActive(), is(true)); - assertThat(ContextManager.activeSpan() == entry, is(true)); - ContextManager.stopSpan(entry); + assertThat(ContextManager.activeSpan() == callerSpan, is(true)); + verify(delegate).failed(cause); - assertThat(segmentStorage.getTraceSegments().size(), is(2)); - assertThat(SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(1)).size(), is(1)); - verify(delegate).completed("body"); + ContextManager.stopSpan(callerSpan); + + assertThat(segmentStorage.getTraceSegments().size(), is(1)); } @Test - public void nullContextNeverStopsForeignSpan() { - AbstractSpan entry = ContextManager.createEntrySpan("/business", null); + public void cancelledKeepsCallerSpanActive() { + AbstractSpan callerSpan = ContextManager.createEntrySpan("/business", null); + AsyncExitSpan exitSpan = createStartedExitSpan(); - new FutureCallbackWrapper<>(delegate, null).completed("ok"); + new FutureCallbackWrapper<>(delegate, exitSpan).cancelled(); assertThat(ContextManager.isActive(), is(true)); - ContextManager.stopSpan(entry); - verify(delegate).completed("ok"); + assertThat(ContextManager.activeSpan() == callerSpan, is(true)); + verify(delegate).cancelled(); + + ContextManager.stopSpan(callerSpan); + + assertThat(segmentStorage.getTraceSegments().size(), is(1)); } -} +} \ No newline at end of file