Fix HttpClient 5.x callback span handling - #832
Open
Ayush0612005 wants to merge 3 commits into
Open
Ayush0612005 wants to merge 3 commits into
Ayush0612005 wants to merge 3 commits into
Conversation
wu-sheng
reviewed
Sep 22, 2026
wu-sheng
left a comment
wu-sheng
left a comment
Member
There was a problem hiding this comment.
Thanks for working on this. The diagnosis is right: with HttpAsyncClients.classic(...), ClassicToAsyncResponseConsumer.fireComplete() runs the result callback on the caller thread when the body is read to EOF, so FutureCallbackWrapper pops the caller's span. But the fix introduces a regression and still leaves some cases broken. I checked everything below with unit probes and against a real client: the scenario app with the agent and the mock collector, on httpclient 5.0 and 5.5.2.
Issues in this PR
1. Regression: overlapping requests on one I/O reactor thread leave its context open forever
IOSessionImplPollInterceptor creates the local and exit spans in poll() and leaves both on the reactor thread's ThreadLocal span stack until the response arrives. A reactor thread serves many connections (one thread per CPU by default), so another request is often polled before the first response comes back:
poll(A) [localA, exitA]
poll(B) [localA, exitA, localB, exitB] <- B is pushed into A's segment
response(A) stopSpan() pops exitB <- A's status lands on B's exit span
AsyncResponseConsumerWrapper.consumeResponse still pops whatever span is on top. The new OwnedSpans checks only stop a span when it is the one on top, so once the wrong span has been popped they never match again, and [localA, exitA] is never stopped.
Real client, httpclient 5.0, one reactor thread (IOReactorConfig.setIoThreadCount(1)), 4 overlapping requests answered out of order, 2 rounds:
| Plugin | I/O side |
|---|---|
| before this PR | All 4 requests nested into one segment as a local, exit, local, exit… chain with 4 refs. The structure is wrong, but the segment is reported. |
| this PR | The I/O segment is never reported. Round 2's spans (#9–#15) nest into round 1's never-finished segment, so the downstream entry segments reference a parent segment that never arrives. |
Any interleaving on a reactor thread follows the same pattern, for example HTTP/2 streams. I only tested HTTP/1.1 connections sharing one reactor thread.
2. The "is it the active span" check matches across threads when both contexts are ignored
IgnoredTracerContext.createLocalSpan() and activeSpan() return the same static NOOP_SPAN. So owned == ContextManager.activeSpan() in OwnedSpans is true on any thread whose context is ignored. Both sides are ignored when the OAP is disconnected (keep_tracing=false, the default) or when sampling drops both. The attribute is never removed, so the callback on the caller thread still ends the caller's ignored context. A unit probe showed the caller context closed after completed(), which is the #14097 symptom. The impact is small because no traced data is lost, but the next ContextManager.stopSpan() on that thread throws an NPE, which is logged as an intercept failure.
3. The local span now ends when the response headers arrive
Before, it ended at completed(), so it covered the body transfer. Now no span covers the time spent receiving the body.
4. The stored span is never removed from the HttpContext
When users reuse an HttpClientContext, it holds a reference to the finished span until the next request.
Other cases found (they predate this PR, but matter for the issue)
- Null HttpContext on 5.4+. From 5.4 on,
CloseableHttpAsyncClient#execute(SimpleHttpRequest, FutureCallback)passes anullcontext: 5.0–5.2.3 passHttpClientContext.create(), and 5.4–5.6 passnull. The classic facade does the same forclient.execute(request, handler).HttpAsyncClientDoExecuteInterceptorthen throws an NPE oncontext.setAttribute(...), which is logged, and no snapshot is stored. In the exact repro from #14097, no HTTP span is created at all. The only effect of the plugin isFutureCallbackWrapperpopping the caller's span. So the null-context test in this PR is the case that matters for the issue. - On 5.4+, the request is not in the context when
poll()runs.HttpAsyncMainClientExeccallsclientContext.setRequest(request)insideproduceRequest, which the I/O layer calls afterpoll(). httpcore 5.3 also keeps the request in a field, sogetAttribute(HTTP_REQUEST)returnsnull. The current code reads the request aftercreateLocalSpan, so the resulting NPE leaves that span on the reactor thread. That path is only reachable when the caller passes its own context. So on 5.4+ the async plugin produces no I/O-side spans. informationResponse(1xx) stops the exit span, so the final status code lands on the local span.AsyncResponseConsumerWrapper.failedalso pops whatever span is on top of the current thread.httpclient-5.x-scenarioonly runs 5.0 and 5.1, so none of the 5.4+ behavior, the classic facade, or the interleaving case is covered by CI.
Suggested fix: async spans, finished by reference
Don't keep the spans on the reactor thread's stack; finish them by reference with the async-span API, as asynchttpclient-2.x-plugin does:
HttpAsyncClientDoExecuteInterceptor(caller thread): only when the caller context is active, create a small per-requestAsyncRequestSpansobject holding the snapshot. Store it in the HttpContext and pass it toAsyncResponseConsumerWrapper. If the context isnull, setHttpClientContext.create()as the argument;castOrCreate(5.4+) andadapt(5.0) both keep the same instance. Stop wrapping theFutureCallback.IOSessionImplPollInterceptor(reactor thread):- Create the local and exit spans as before and inject the headers.
- Call
exit.prepareForAsync()thenstopSpan(exit), andlocal.prepareForAsync()thenstopSpan(local). The order matters becauseprepareForAsync()requires the span to be the active one. - Hand both spans to
AsyncRequestSpans. The reactor thread's stack is empty again whenpoll()returns.
AsyncResponseConsumerWrapper:consumeResponse: set the status tag, thenexit.asyncFinish(). If there is no body, alsolocal.asyncFinish().streamEnd:local.asyncFinish()at the last body byte.failed: mark both spans as errors and finish them.releaseResources: fallback, finish whatever is still open.informationResponse: no span work.
- Each span sits in an
AtomicReferenceand is finished at most once (getAndSet(null)), from whichever thread gets there first.asyncFinish()does nothing onNoopSpan, and there are no "is it on top" checks left, so issue 2 goes away too. - Remove
FutureCallbackWrapper. Nothing that runs on the caller thread touches spans any more. - In
poll(), read the request withHttpCoreContext#getRequest(), which has the same signature in 5.0 and 5.3, and read it before creating any span. When it'snull, skip; don't throw aftercreateLocalSpan.
Durations: the exit span runs from the request being sent to the response headers, as before. The local span runs from poll to the last body byte, which matches what completed() gave in the normal async path.
Verified results
| Case | before this PR | this PR | suggested fix |
|---|---|---|---|
| 5.0, 4 overlapping requests on 1 reactor thread, 2 rounds | 1 segment, all 4 nested | segment never reported, round 2 nested into it | one segment per request (local + exit, CrossThread ref to the caller, durations 429/125/328/74 ms for 400/100/300/50 ms server delays, downstream refs correct) |
| 5.0, caller's own sync exit span after the async calls | stays in the caller segment | stays in the caller segment | stays in the caller segment |
5.5.2, HttpAsyncClients.classic(...), sync exit span after reading the body |
not re-run (reported in #14097) | not run | stays in the caller segment: the caller's context survives |
| unit tests: normal flow, overlapping requests, callback on the caller thread, failure after poll, response with no body, double release, null context, untraced caller | 8/8 pass, plus checkstyle |
Cases that still have issues with the suggested fix
- 5.4+: still no I/O-side spans, because the request isn't available in
poll()(see above). Supporting 5.4+ needs a different hook, for example whereproduceRequestruns. That is a separate follow-up, and 5.4+ is outside the documented 5.0/5.1 range. The caller-side fix for #14097 does work on 5.5.2. - An ignored caller snapshot is still continued. The I/O side then starts a new trace, which may be sampled. Only storing the snapshot when
snapshot.isValid()would stop that; I left it out to keep the change focused. - Scenario coverage: it would help to add a single-reactor concurrent endpoint to
httpclient-5.x-scenario(see the probe below), plus a 5.5+ version for the classic facade once the 5.4+ hook exists.
Suggested patch (main code, against main @ caf5d0a, replaces this PR's changes)
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncRequestSpans.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncRequestSpans.java
new file mode 100644
index 0000000000..c196d3856c
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncRequestSpans.java
@@ -0,0 +1,94 @@
+/*
+ * 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 java.util.concurrent.atomic.AtomicReference;
+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;
+
+/**
+ * The tracing state of one async request. It is created on the caller thread when the request is submitted, reaches
+ * the I/O reactor thread through the request's {@code HttpContext}, and is held by the response consumer wrapper.
+ * <p>
+ * The spans are created on the reactor thread when the request is polled, and are stopped there right away in async
+ * mode. The reactor thread serves many connections, so the spans of one request must not stay on its span stack
+ * while other requests are processed. They are finished by {@link AbstractSpan#asyncFinish()} from whichever thread
+ * ends the exchange, each at most once.
+ */
+public class AsyncRequestSpans {
+
+ private final ContextSnapshot snapshot;
+ private final AtomicReference<AbstractSpan> localSpan = new AtomicReference<>();
+ private final AtomicReference<AbstractSpan> exitSpan = new AtomicReference<>();
+
+ public AsyncRequestSpans(ContextSnapshot snapshot) {
+ this.snapshot = snapshot;
+ }
+
+ public ContextSnapshot getSnapshot() {
+ return snapshot;
+ }
+
+ /**
+ * @param localSpan the local span, already stopped on the reactor thread after {@link AbstractSpan#prepareForAsync()}
+ * @param exitSpan the exit span, already stopped on the reactor thread after {@link AbstractSpan#prepareForAsync()}
+ */
+ public void start(AbstractSpan localSpan, AbstractSpan exitSpan) {
+ this.localSpan.set(localSpan);
+ this.exitSpan.set(exitSpan);
+ }
+
+ /**
+ * The response head arrived, the exit span ends here.
+ */
+ public void onResponse(int statusCode) {
+ AbstractSpan span = exitSpan.getAndSet(null);
+ if (span != null) {
+ Tags.HTTP_RESPONSE_STATUS_CODE.set(span, statusCode);
+ if (statusCode >= 400) {
+ span.errorOccurred();
+ }
+ span.asyncFinish();
+ }
+ }
+
+ /**
+ * The whole response has been received, or the exchange was released.
+ */
+ public void onComplete() {
+ finish(exitSpan, null);
+ finish(localSpan, null);
+ }
+
+ public void onFailure(Exception cause) {
+ finish(exitSpan, cause);
+ finish(localSpan, cause);
+ }
+
+ private static void finish(AtomicReference<AbstractSpan> spanRef, Exception cause) {
+ AbstractSpan span = spanRef.getAndSet(null);
+ if (span != null) {
+ if (cause != null) {
+ span.errorOccurred().log(cause);
+ }
+ span.asyncFinish();
+ }
+ }
+}
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..3496565adf 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
@@ -19,5 +19,5 @@ package org.apache.skywalking.apm.plugin.httpclient.v5;
public class Constants {
- public static String SKYWALKING_CONTEXT_SNAPSHOT = "skywalking-context-snapshot";
+ public static final String SKYWALKING_ASYNC_REQUEST_SPANS = "skywalking-async-request-spans";
}
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..278742b3c6 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,7 +18,7 @@
package org.apache.skywalking.apm.plugin.httpclient.v5;
-import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
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;
@@ -26,7 +26,6 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedI
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.AsyncResponseConsumerWrapper;
-import org.apache.skywalking.apm.plugin.httpclient.v5.wrapper.FutureCallbackWrapper;
import java.lang.reflect.Method;
@@ -35,14 +34,19 @@ public class HttpAsyncClientDoExecuteInterceptor implements InstanceMethodsAroun
@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
MethodInterceptResult result) throws Throwable {
- AsyncResponseConsumer consumer = (AsyncResponseConsumer) allArguments[2];
+ if (!ContextManager.isActive()) {
+ return;
+ }
HttpContext context = (HttpContext) allArguments[4];
- FutureCallback callback = (FutureCallback) allArguments[5];
- allArguments[2] = new AsyncResponseConsumerWrapper(consumer);
- allArguments[5] = new FutureCallbackWrapper(callback);
- if (ContextManager.isActive()) {
- context.setAttribute(Constants.SKYWALKING_CONTEXT_SNAPSHOT, ContextManager.capture());
+ if (context == null) {
+ // Since 5.4 the convenience execute methods pass no context, doExecute creates one by
+ // HttpClientContext#castOrCreate, which keeps a given HttpClientContext instance.
+ context = HttpClientContext.create();
+ allArguments[4] = context;
}
+ AsyncRequestSpans spans = new AsyncRequestSpans(ContextManager.capture());
+ context.setAttribute(Constants.SKYWALKING_ASYNC_REQUEST_SPANS, spans);
+ allArguments[2] = new AsyncResponseConsumerWrapper<>((AsyncResponseConsumer<?>) allArguments[2], spans);
}
@Override
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..3afc00f679 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
@@ -18,15 +18,14 @@
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.HttpRequest;
import org.apache.hc.core5.http.nio.command.RequestExecutionCommand;
import org.apache.hc.core5.http.protocol.HttpContext;
+import org.apache.hc.core5.http.protocol.HttpCoreContext;
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;
@@ -54,26 +53,33 @@ public class IOSessionImplPollInterceptor implements InstanceMethodsAroundInterc
return ret;
}
HttpContext httpContext = ((RequestExecutionCommand) command).getContext();
- ContextSnapshot snapshot = (ContextSnapshot) httpContext.getAttribute(Constants.SKYWALKING_CONTEXT_SNAPSHOT);
- if (snapshot == null) {
+ AsyncRequestSpans spans = (AsyncRequestSpans) httpContext.getAttribute(Constants.SKYWALKING_ASYNC_REQUEST_SPANS);
+ if (spans == null) {
return ret;
}
- httpContext.removeAttribute(Constants.SKYWALKING_CONTEXT_SNAPSHOT);
+ httpContext.removeAttribute(Constants.SKYWALKING_ASYNC_REQUEST_SPANS);
+ // Since httpcore 5.3 the request is a field of HttpCoreContext, not an attribute any more.
+ HttpRequest request = httpContext instanceof HttpCoreContext
+ ? ((HttpCoreContext) httpContext).getRequest()
+ : (HttpRequest) httpContext.getAttribute(HttpCoreContext.HTTP_REQUEST);
+ if (request == null) {
+ return ret;
+ }
+ URI uri = request.getUri();
+ String url = uri.toURL().toString();
+
AbstractSpan localSpan = ContextManager.createLocalSpan("httpasyncclient/local");
localSpan.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT);
localSpan.setLayer(SpanLayer.HTTP);
- ContextManager.continued(snapshot);
+ ContextManager.continued(spans.getSnapshot());
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.URL.set(span, url);
Tags.HTTP.METHOD.set(span, request.getMethod());
SpanLayer.asHttp(span);
CarrierItem next = contextCarrier.items();
@@ -81,6 +87,14 @@ public class IOSessionImplPollInterceptor implements InstanceMethodsAroundInterc
next = next.next();
request.setHeader(next.getHeadKey(), next.getHeadValue());
}
+
+ // This reactor thread serves many connections, other requests are polled before this response arrives.
+ // Leave nothing on its span stack, the spans are finished by AsyncRequestSpans.
+ span.prepareForAsync();
+ ContextManager.stopSpan(span);
+ localSpan.prepareForAsync();
+ ContextManager.stopSpan(localSpan);
+ spans.start(localSpan, span);
return ret;
}
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..3db8ea558b 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
@@ -25,57 +25,46 @@ 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.AsyncRequestSpans;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
+/**
+ * Finishes the spans of the request through {@link AsyncRequestSpans}, never through the span stack of the current
+ * thread, because the result may be consumed on a thread other than the one that created the spans, e.g. the caller
+ * thread of {@code HttpAsyncClients.classic(...)}.
+ */
public class AsyncResponseConsumerWrapper<T> implements AsyncResponseConsumer<T> {
- private AsyncResponseConsumer<T> consumer;
+ private final AsyncResponseConsumer<T> consumer;
+ private final AsyncRequestSpans spans;
- public AsyncResponseConsumerWrapper(AsyncResponseConsumer<T> consumer) {
+ public AsyncResponseConsumerWrapper(AsyncResponseConsumer<T> consumer, AsyncRequestSpans spans) {
this.consumer = consumer;
+ this.spans = spans;
}
@Override
public void consumeResponse(HttpResponse response, EntityDetails entityDetails, HttpContext context,
FutureCallback<T> 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();
+ spans.onResponse(response.getCode());
+ if (entityDetails == null) {
+ // no entity, streamEnd will not be called
+ spans.onComplete();
}
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();
- }
consumer.informationResponse(response, context);
}
@Override
public void failed(Exception cause) {
- if (ContextManager.isActive()) {
- ContextManager.activeSpan().errorOccurred().log(cause);
- ContextManager.stopSpan();
- }
+ spans.onFailure(cause);
consumer.failed(cause);
}
@@ -91,11 +80,14 @@ public class AsyncResponseConsumerWrapper<T> implements AsyncResponseConsumer<T>
@Override
public void streamEnd(List<? extends Header> trailers) throws HttpException, IOException {
+ spans.onComplete();
consumer.streamEnd(trailers);
}
@Override
public void releaseResources() {
+ // the exchange is over, e.g. cancelled, finish whatever is still open
+ spans.onComplete();
consumer.releaseResources();
}
}
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
deleted file mode 100644
index f606856edf..0000000000
--- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java
+++ /dev/null
@@ -1,62 +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.wrapper;
-
-import org.apache.hc.core5.concurrent.FutureCallback;
-import org.apache.skywalking.apm.agent.core.context.ContextManager;
-
-public class FutureCallbackWrapper<T> implements FutureCallback<T> {
-
- private FutureCallback<T> callback;
-
- public FutureCallbackWrapper(FutureCallback<T> callback) {
- this.callback = callback;
- }
-
- @Override
- public void completed(T o) {
- if (ContextManager.isActive()) {
- ContextManager.stopSpan();
- }
- if (callback != null) {
- callback.completed(o);
- }
- }
-
- @Override
- public void failed(Exception e) {
- if (ContextManager.isActive()) {
- ContextManager.activeSpan().errorOccurred().log(e);
- ContextManager.stopSpan();
- }
- if (callback != null) {
- callback.failed(e);
- }
- }
-
- @Override
- public void cancelled() {
- if (ContextManager.isActive()) {
- ContextManager.activeSpan().errorOccurred();
- ContextManager.stopSpan();
- }
- if (callback != null) {
- callback.cancelled();
- }
- }
-}Unit tests for the patch (HttpAsyncClientTracingTest)
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientTracingTest.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientTracingTest.java
new file mode 100644
index 0000000000..4832403ec6
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientTracingTest.java
@@ -0,0 +1,310 @@
+/*
+ * 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.concurrent.FutureCallback;
+import org.apache.hc.core5.http.EntityDetails;
+import org.apache.hc.core5.http.HttpResponse;
+import org.apache.hc.core5.http.message.BasicHttpRequest;
+import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
+import org.apache.hc.core5.http.nio.AsyncRequestProducer;
+import org.apache.hc.core5.http.nio.AsyncResponseConsumer;
+import org.apache.hc.core5.http.nio.command.RequestExecutionCommand;
+import org.apache.hc.core5.http.protocol.BasicHttpContext;
+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.agent.core.context.trace.AbstractTracingSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
+import org.apache.skywalking.apm.agent.core.context.util.TagValuePair;
+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.junit.After;
+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.io.IOException;
+import java.net.URI;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.CoreMatchers.sameInstance;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Mockito.when;
+
+/**
+ * Drives the interceptors and the consumer wrapper in the thread model of the async client: the request is
+ * submitted on the caller thread, polled and answered on an I/O reactor thread, which serves many connections.
+ */
+@RunWith(TracingSegmentRunner.class)
+public class HttpAsyncClientTracingTest {
+
+ @SegmentStoragePoint
+ private SegmentStorage segmentStorage;
+
+ @Rule
+ public AgentServiceRule agentServiceRule = new AgentServiceRule();
+ @Rule
+ public MockitoRule rule = MockitoJUnit.rule();
+
+ @Mock
+ private AsyncRequestProducer requestProducer;
+ @Mock
+ private AsyncResponseConsumer<String> consumer;
+ @Mock
+ private FutureCallback<String> callback;
+ @Mock
+ private AsyncClientExchangeHandler exchangeHandler;
+ @Mock
+ private HttpResponse ok;
+ @Mock
+ private HttpResponse serverError;
+ @Mock
+ private EntityDetails entity;
+
+ private final HttpAsyncClientDoExecuteInterceptor doExecuteInterceptor = new HttpAsyncClientDoExecuteInterceptor();
+ private final IOSessionImplPollInterceptor pollInterceptor = new IOSessionImplPollInterceptor();
+ private ExecutorService reactor;
+
+ @Before
+ public void setUp() {
+ when(ok.getCode()).thenReturn(200);
+ when(serverError.getCode()).thenReturn(500);
+ reactor = Executors.newSingleThreadExecutor();
+ }
+
+ @After
+ public void tearDown() {
+ reactor.shutdownNow();
+ }
+
+ @Test
+ public void responseOnReactorThread() throws Throwable {
+ AbstractSpan entry = ContextManager.createEntrySpan("/caller", null);
+ Object[] request = submit(new BasicHttpContext(), "/a");
+ assertThat("the callback is not wrapped", request[5], sameInstance(callback));
+
+ onReactor(() -> {
+ poll(request);
+ assertThat("nothing is left on the reactor thread", ContextManager.isActive(), is(false));
+ wrapper(request).consumeResponse(ok, entity, context(request), null);
+ wrapper(request).streamEnd(null);
+ return null;
+ });
+
+ assertThat(segmentStorage.getTraceSegments().size(), is(1));
+ TraceSegment ioSegment = segmentStorage.getTraceSegments().get(0);
+ assertThat(ioSegment.getRef(), notNullValue());
+ assertExitSpan(ioSegment, "/a", "200", false);
+
+ assertThat(ContextManager.activeSpan(), sameInstance(entry));
+ ContextManager.stopSpan(entry);
+ assertThat(segmentStorage.getTraceSegments().size(), is(2));
+ }
+
+ /**
+ * Two connections on the same reactor thread, request B is polled before the response of A arrives.
+ */
+ @Test
+ public void interleavedRequestsOnOneReactorThread() throws Throwable {
+ AbstractSpan entry = ContextManager.createEntrySpan("/caller", null);
+ Object[] a = submit(new BasicHttpContext(), "/a");
+ Object[] b = submit(new BasicHttpContext(), "/b");
+
+ onReactor(() -> {
+ poll(a);
+ poll(b);
+ wrapper(a).consumeResponse(ok, entity, context(a), null);
+ wrapper(b).consumeResponse(serverError, entity, context(b), null);
+ wrapper(b).streamEnd(null);
+ wrapper(a).streamEnd(null);
+ assertThat(ContextManager.isActive(), is(false));
+ return null;
+ });
+
+ List<TraceSegment> segments = segmentStorage.getTraceSegments();
+ assertThat(segments.size(), is(2));
+ assertExitSpan(segments.get(0), "/b", "500", true);
+ assertExitSpan(segments.get(1), "/a", "200", false);
+ ContextManager.stopSpan(entry);
+ }
+
+ /**
+ * {@code HttpAsyncClients.classic(...)}: the body is read, and the result callback fired, on the caller thread.
+ */
+ @Test
+ public void resultConsumedOnCallerThread() throws Throwable {
+ AbstractSpan entry = ContextManager.createEntrySpan("/caller", null);
+ Object[] request = submit(new BasicHttpContext(), "/a");
+
+ onReactor(() -> {
+ poll(request);
+ wrapper(request).consumeResponse(ok, entity, context(request), null);
+ wrapper(request).streamEnd(null);
+ return null;
+ });
+ assertThat(segmentStorage.getTraceSegments().size(), is(1));
+
+ ((FutureCallback<String>) request[5]).completed("body");
+ assertThat(ContextManager.activeSpan(), sameInstance(entry));
+ ContextManager.stopSpan(entry);
+ assertThat(segmentStorage.getTraceSegments().size(), is(2));
+ }
+
+ @Test
+ public void failureAfterPoll() throws Throwable {
+ ContextManager.createEntrySpan("/caller", null);
+ Object[] request = submit(new BasicHttpContext(), "/a");
+
+ onReactor(() -> {
+ poll(request);
+ return null;
+ });
+ // e.g. a connection reset reported by another thread
+ wrapper(request).failed(new IOException("reset"));
+
+ assertThat(segmentStorage.getTraceSegments().size(), is(1));
+ for (AbstractTracingSpan span : SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0))) {
+ assertThat(SpanHelper.getErrorOccurred(span), is(true));
+ assertThat(SpanHelper.getLogs(span).size(), is(1));
+ }
+ ContextManager.stopSpan();
+ }
+
+ @Test
+ public void responseWithoutEntity() throws Throwable {
+ ContextManager.createEntrySpan("/caller", null);
+ Object[] request = submit(new BasicHttpContext(), "/a");
+
+ onReactor(() -> {
+ poll(request);
+ wrapper(request).consumeResponse(ok, null, context(request), null);
+ return null;
+ });
+
+ assertThat(segmentStorage.getTraceSegments().size(), is(1));
+ ContextManager.stopSpan();
+ }
+
+ @Test
+ public void releasedWithoutResponse() throws Throwable {
+ ContextManager.createEntrySpan("/caller", null);
+ Object[] request = submit(new BasicHttpContext(), "/a");
+
+ onReactor(() -> {
+ poll(request);
+ return null;
+ });
+ wrapper(request).releaseResources();
+ wrapper(request).releaseResources();
+
+ assertThat(segmentStorage.getTraceSegments().size(), is(1));
+ ContextManager.stopSpan();
+ }
+
+ /**
+ * Since 5.4, {@code CloseableHttpAsyncClient#execute(SimpleHttpRequest, FutureCallback)} passes a null context.
+ */
+ @Test
+ public void nullContextIsCreated() throws Throwable {
+ ContextManager.createEntrySpan("/caller", null);
+ Object[] request = submit(null, "/a");
+
+ assertThat(request[4], instanceOf(HttpClientContext.class));
+ assertThat(context(request).getAttribute(Constants.SKYWALKING_ASYNC_REQUEST_SPANS), notNullValue());
+ ContextManager.stopSpan();
+ }
+
+ @Test
+ public void untracedCallerIsNotEnhanced() throws Throwable {
+ Object[] request = submit(null, "/a");
+
+ assertThat(request[2], sameInstance(consumer));
+ assertThat(request[4], nullValue());
+ }
+
+ private Object[] submit(HttpContext context, String path) throws Throwable {
+ Object[] allArguments = new Object[] {null, requestProducer, consumer, null, context, callback};
+ doExecuteInterceptor.beforeMethod(null, null, allArguments, null, null);
+ if (allArguments[4] != null) {
+ context(allArguments).setAttribute(
+ HttpClientContext.HTTP_REQUEST, new BasicHttpRequest("GET", URI.create("http://127.0.0.1:8080" + path)));
+ }
+ return allArguments;
+ }
+
+ private void poll(Object[] request) throws Exception {
+ try {
+ pollInterceptor.afterMethod(
+ null, null, null, null, new RequestExecutionCommand(exchangeHandler, context(request)));
+ } catch (Exception e) {
+ throw e;
+ } catch (Throwable t) {
+ throw new IllegalStateException(t);
+ }
+ }
+
+ private void onReactor(Callable<Void> task) throws Exception {
+ reactor.submit(task).get();
+ }
+
+ private static HttpContext context(Object[] request) {
+ return (HttpContext) request[4];
+ }
+
+ @SuppressWarnings("unchecked")
+ private static AsyncResponseConsumerWrapper<String> wrapper(Object[] request) {
+ return (AsyncResponseConsumerWrapper<String>) request[2];
+ }
+
+ private static void assertExitSpan(TraceSegment segment, String operationName, String statusCode,
+ boolean error) {
+ List<AbstractTracingSpan> spans = SegmentHelper.getSpans(segment);
+ assertThat(spans.size(), is(2));
+ AbstractTracingSpan exit = spans.get(0);
+ assertThat(exit.isExit(), is(true));
+ assertThat(exit.getOperationName(), is(operationName));
+ assertThat(SpanHelper.getErrorOccurred(exit), is(error));
+ String status = null;
+ for (TagValuePair tag : SpanHelper.getTags(exit)) {
+ if (tag.getKey().key().equals("http.status_code")) {
+ status = tag.getValue();
+ }
+ }
+ assertThat(status, is(statusCode));
+ assertThat(spans.get(1).getOperationName(), is("httpasyncclient/local"));
+ }
+}Probe endpoints used for the real-client runs (added to the httpclient-5.x-scenario app)
package org.apache.skywalking.testcase.httpclient5;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import org.apache.hc.client5.http.async.methods.SimpleHttpRequests;
import org.apache.hc.client5.http.async.methods.SimpleHttpResponse;
import org.apache.hc.client5.http.classic.methods.HttpGet;
import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
import org.apache.hc.core5.http.io.entity.EntityUtils;
import org.apache.hc.core5.reactor.IOReactorConfig;
import org.apache.hc.core5.util.Timeout;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/httpclient-5.x/probe")
public class ProbeController {
private static final String BASE = "http://127.0.0.1:8080/httpclient-5.x/probe";
@Autowired
private CloseableHttpClient httpClient;
/** One I/O reactor thread, so concurrent requests on different connections interleave on it. */
private final CloseableHttpAsyncClient singleReactorClient = start(HttpAsyncClients.custom()
.setIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(1).build())
.build());
private final CloseableHttpAsyncClient classicBackend = start(HttpAsyncClients.createDefault());
private static CloseableHttpAsyncClient start(CloseableHttpAsyncClient client) {
client.start();
return client;
}
@GetMapping("/slow")
public String slow(@RequestParam("ms") long ms) throws InterruptedException {
Thread.sleep(ms);
return "slept " + ms;
}
/** Four overlapping requests answered out of order, then a sync call that must stay in the caller's segment. */
@GetMapping("/concurrent")
public String concurrent() throws Exception {
List<Future<SimpleHttpResponse>> futures = new ArrayList<>();
for (long ms : new long[] {400, 100, 300, 50}) {
futures.add(singleReactorClient.execute(SimpleHttpRequests.get(BASE + "/slow?ms=" + ms), null));
}
StringBuilder result = new StringBuilder();
for (Future<SimpleHttpResponse> future : futures) {
result.append(future.get().getBodyText()).append(';');
}
afterward();
return result.toString();
}
/** apache/skywalking#14097: the classic facade reads the body, and fires the callback, on this thread. */
@GetMapping("/classic")
public String classic() throws Exception {
Method factory = HttpAsyncClients.class.getMethod("classic", CloseableHttpAsyncClient.class, Timeout.class);
CloseableHttpClient classic = (CloseableHttpClient) factory.invoke(null, classicBackend, Timeout.ofSeconds(30));
String body = classic.execute(new HttpGet(BASE + "/slow?ms=10"),
response -> EntityUtils.toString(response.getEntity()));
afterward();
return body;
}
/** An exit span made after the async calls, it lands in the caller's segment only if the context survived. */
private void afterward() throws Exception {
httpClient.execute(new HttpGet(BASE + "/slow?ms=1"), response -> EntityUtils.toString(response.getEntity()));
}
}Run with -javaagent, the mock collector, and curl against /httpclient-5.x/probe/concurrent (twice) and /httpclient-5.x/probe/classic (5.5+). Then check receiveData:
- every
/probe/slowentry segment should have a CrossProcess ref to anhttpasyncclient/localsegment that was actually reported; - the caller's entry segment should still contain the sync
/probe/slowexit span.
Contributor
|
I think we can simplify the fix by removing IOSessionImplPollInterceptor instrumentation entirely. The original poll() approach may have assumed that FutureCallback runs on the reactor I/O thread. In that case, creating and closing spans on that thread could appear safe. However, with HttpAsyncClients.classic, FutureCallback can run on the business thread after the response body is consumed to EOF. Therefore, ContextManager.stopSpan() in the callback may close the caller span on the business thread, instead of the HTTP span created on the I/O thread. We could create the HTTP exit span in doExecute, while the caller context is active, then call prepareForAsync() and stop that specific span before returning. The callback wrapper keeps the AbstractSpan reference and calls asyncFinish() in completed / failed, rather than calling parameterless ContextManager.stopSpan(). We should also finish the retained span safely from cancellation or resource-release paths when completed is not invoked. |
Author
|
Thanks for looking at this — I agree that retaining the span by reference and calling asyncFinish() instead of a parameterless stopSpan() in the callback is the right direction, and it's also consistent with the AsyncRequestSpans approach Wu Sheng suggested above. One constraint with moving span creation into doExecute is that at that point we only have the AsyncRequestProducer, not the concrete HttpRequest. The actual request (URI, method, headers) is produced later by produceRequest on the I/O reactor, after poll(). We also need the concrete request to inject the sw8 propagation headers. So I think we can keep the existing I/O-side creation while getting the behavior you're suggesting: create the local + exit spans in IOSessionImplPollInterceptor, immediately detach them with prepareForAsync() / stopSpan(), store them in an AsyncRequestSpans holder, and finish them by reference with asyncFinish() from the response/callback lifecycle, including cancellation and releaseResources(). I'll implement it this way and update the PR. Let me know if I'm missing something about why creating the span in doExecute would still give us access to the concrete request. |
Contributor
|
We can wrap the AsyncRequestProducer argument in doExecute. The wrapper delegates sendRequest, but supplies a wrapped RequestChannel. When the producer eventually calls RequestChannel.sendRequest(HttpRequest, EntityDetails, HttpContext), the wrapper receives the concrete HttpRequest and can inject the sw8 headers before forwarding it. Therefore, we do not need IOSessionImplPollInterceptor just to obtain the request or inject propagation headers. doExecute can create and detach the exit span while the caller context is active; the producer wrapper only updates request metadata and injects the carrier into the actual request. The response/callback lifecycle then finishes the retained span by asyncFinish(). |
Member
|
+1 to @xuzhiguang's direction: wrap the Facts the design relies on (verified in source)
This also fixes 5.4+, where the plugin currently produces no async spans: the Recommended changesRemove: New public class AsyncExitSpan {
private final HttpHost target;
// the doExecute thread, where the caller's context is active; cleared once the span is created or doExecute returns
private volatile Thread creator = Thread.currentThread();
private AbstractSpan span;
public AsyncExitSpan(HttpHost target) {
this.target = target;
}
public HttpHost getTarget() {
return target;
}
/** true only once, only on the doExecute thread, only before doExecute returns */
public boolean claimCreation() {
if (creator != Thread.currentThread()) {
return false;
}
creator = null;
return true;
}
public void callerReturned() {
creator = null;
}
/** the span is already detached from the caller's stack, after prepareForAsync() */
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();
}
}
}
/** the whole response was received */
public synchronized void finish() {
end(false, null);
}
public synchronized void fail(Throwable cause) {
end(true, cause);
}
/** cancelled, or released before the response completed */
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;
}
}The methods are
public void beforeMethod(...) {
if (!ContextManager.isActive()) {
return;
}
AsyncExitSpan exitSpan = new AsyncExitSpan((HttpHost) allArguments[0]);
allArguments[1] = new AsyncRequestProducerWrapper((AsyncRequestProducer) allArguments[1], exitSpan);
allArguments[2] = new AsyncResponseConsumerWrapper<>((AsyncResponseConsumer<?>) allArguments[2], exitSpan);
// wrap even a null callback: it is where the future's end is reported
allArguments[5] = new FutureCallbackWrapper<>((FutureCallback<?>) allArguments[5], exitSpan);
}
public Object afterMethod(...) { // also runs after handleMethodException
if (allArguments[1] instanceof AsyncRequestProducerWrapper) {
((AsyncRequestProducerWrapper) allArguments[1]).getExitSpan().callerReturned();
}
return ret;
}
public void handleMethodException(..., Throwable t) {
if (allArguments[1] instanceof AsyncRequestProducerWrapper) {
((AsyncRequestProducerWrapper) allArguments[1]).getExitSpan().fail(t);
}
}
New @Override
public void sendRequest(RequestChannel channel, HttpContext context) throws HttpException, IOException {
producer.sendRequest((request, entityDetails, ctx) -> {
if (exitSpan.claimCreation()) {
try {
startExitSpan(request);
} catch (Throwable t) {
LOGGER.error(t, "Failed to trace the async HTTP request."); // never fail the user's request
}
}
channel.sendRequest(request, entityDetails, ctx);
}, context);
}
private void startExitSpan(HttpRequest request) throws URISyntaxException {
URI uri = request.getUri();
HttpHost target = exitSpan.getTarget();
// same precedence as InternalAbstractHttpAsyncClient: an explicit target wins over the request's authority
String scheme = target != null ? target.getSchemeName() : uri.getScheme();
String host = target != null ? target.getHostName() : uri.getHost();
int port = target != null ? target.getPort() : uri.getPort();
if (host == null) {
return;
}
if (scheme == null) {
scheme = "http";
}
if (port < 0) {
port = "https".equalsIgnoreCase(scheme) ? 443 : 80;
}
String peer = host + ":" + port;
String path = uri.getPath() == null || uri.getPath().isEmpty() ? "/" : uri.getPath();
String url = scheme + "://" + peer + path + (uri.getRawQuery() == null ? "" : "?" + uri.getRawQuery());
// Inside another plugin's exit span, createExitSpan reuses that span (depth + 1). It must not be turned into an
// async span, so then we only propagate, like the classic client does.
boolean nested = ContextManager.activeSpan().isExit();
// create without the carrier: createExitSpan(op, carrier, peer) injects before returning, and injection throws
// for a reused outer exit span without a peer, which would leave the extra depth unbalanced
AbstractSpan span = ContextManager.createExitSpan(path, peer);
try {
if (!nested) {
span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT);
Tags.URL.set(span, url);
Tags.HTTP.METHOD.set(span, request.getMethod());
SpanLayer.asHttp(span);
}
ContextCarrier carrier = new ContextCarrier();
ContextManager.inject(carrier);
CarrierItem next = carrier.items();
while (next.hasNext()) {
next = next.next();
request.setHeader(next.getHeadKey(), next.getHeadValue());
}
} finally {
// detach BEFORE forwarding: the client may report a failure on this thread before doExecute returns
if (!nested) {
span.prepareForAsync();
}
ContextManager.stopSpan(span);
if (!nested) {
exitSpan.start(span);
}
}
}
consumeResponse: exitSpan.onResponse(response.getCode());
if (entityDetails == null) { exitSpan.finish(); } // no body, streamEnd won't come
consumer.consumeResponse(...);
streamEnd: exitSpan.finish(); // last byte received
consumer.streamEnd(trailers);
failed: exitSpan.fail(cause);
consumer.failed(cause);
releaseResources: exitSpan.abort(); // see below
consumer.releaseResources();
informationResponse: consumer.informationResponse(...); // 1xx: no span work
completed(result): exitSpan.finish(); if (callback != null) callback.completed(result);
failed(e): exitSpan.fail(e); if (callback != null) callback.failed(e);
cancelled(): exitSpan.abort(); if (callback != null) callback.cancelled();It covers the paths the consumer never sees, such as the failure caught in Resulting behaviour
Known limitations to document
Tests to addUnit tests:
Scenario:
A correction to my earlier review: the patch I posted there finished the spans as successful in |
Author
|
Thanks for the detailed review and for verifying the approach against both HttpClient 5.0 and 5.5.2. I agree with this direction. I’ll rework the PR to wrap the AsyncRequestProducer, create and detach the exit span in the caller context, and remove the IOSessionImpl#poll instrumentation. The retained span will then be finished by reference with asyncFinish() from the response consumer and callback lifecycle, including failure, cancellation, and releaseResources() paths. I’ll also add the suggested regression tests, update the scenario expectations, and verify the plugin against the relevant HttpClient versions before requesting another review. Thanks again for catching the reactor-thread interleaving and lifecycle issues in the previous implementation. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fix HttpClient 5.x callback span handling (#14097)
The HttpClient 5.x plugin previously called
ContextManager.stopSpan()fromFutureCallbackWrapperwithout verifying that the active span belonged to theHTTP request.
With
HttpAsyncClients.classic(...), the response body can be consumed on thecaller/business thread and the
FutureCallbackcan therefore execute on thatsame thread. In that case, the callback could incorrectly stop the caller's
active span.
The fix moves request ownership into an
AsyncExitSpanshared by the asyncrequest producer, response consumer, and callback lifecycle.
The request producer wrapper receives the concrete
HttpRequestthroughRequestChannel.sendRequest(...), creates the HTTP exit span in the callercontext, injects the propagation headers into the request, and detaches the
span for asynchronous completion.
The response consumer and callback finish the retained request span by
reference using
asyncFinish(), so they do not stop whichever span happens tobe active on their current thread.
The reactor-thread
poll()instrumentation used by the previous implementationhas been removed so request spans are not left on the I/O reactor thread's
ThreadLocal stack while multiple requests are in flight.
Regression coverage verifies that callback completion, failure, and
cancellation do not incorrectly stop the caller's active span.
CHANGESlog.