diff --git a/CHANGES.md b/CHANGES.md index 79a40e725d..3f30b4f687 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 `NullPointerException` thrown by the `spring-webflux-5.x-webclient` and `spring-webflux-6.x-webclient` plugins when `DefaultClientRequestBuilder$BodyInserterRequest#writeTo` runs 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/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..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); - 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 fc8ef190d4..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,92 +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); - 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/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 9dec7d109a..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,68 +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 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(); + + 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); @@ -91,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 f606856edf..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,21 +18,22 @@ 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; +import org.apache.skywalking.apm.plugin.httpclient.v5.AsyncExitSpan; public class FutureCallbackWrapper implements FutureCallback { - private FutureCallback callback; + private final FutureCallback callback; + private final AsyncExitSpan exitSpan; - public FutureCallbackWrapper(FutureCallback callback) { + public FutureCallbackWrapper(FutureCallback callback, AsyncExitSpan exitSpan) { this.callback = callback; + this.exitSpan = exitSpan; } @Override public void completed(T o) { - if (ContextManager.isActive()) { - ContextManager.stopSpan(); - } + exitSpan.finish(); + if (callback != null) { callback.completed(o); } @@ -40,10 +41,8 @@ public void completed(T o) { @Override public void failed(Exception e) { - if (ContextManager.isActive()) { - ContextManager.activeSpan().errorOccurred().log(e); - ContextManager.stopSpan(); - } + exitSpan.fail(e); + if (callback != null) { callback.failed(e); } @@ -51,12 +50,10 @@ public void failed(Exception e) { @Override public void cancelled() { - if (ContextManager.isActive()) { - ContextManager.activeSpan().errorOccurred(); - ContextManager.stopSpan(); - } + 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 new file mode 100644 index 0000000000..386a3cf1f7 --- /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,128 @@ +/* + * 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.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.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.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 static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.mockito.Mockito.verify; + +@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; + + @Before + public void setUp() { + ServiceManager.INSTANCE.boot(); + } + + private AsyncExitSpan createStartedExitSpan() { + AsyncExitSpan exitSpan = new AsyncExitSpan( + new HttpHost("http", "127.0.0.1", 8080)); + + AbstractSpan requestSpan = ContextManager.createExitSpan( + "/hello", + new ContextCarrier(), + "127.0.0.1:8080"); + + exitSpan.start(requestSpan); + requestSpan.prepareForAsync(); + ContextManager.stopSpan(requestSpan); + + return exitSpan; + } + + @Test + public void completedKeepsCallerSpanActive() { + AbstractSpan callerSpan = ContextManager.createEntrySpan("/business", null); + AsyncExitSpan exitSpan = createStartedExitSpan(); + + new FutureCallbackWrapper<>(delegate, exitSpan).completed("ok"); + + assertThat(ContextManager.isActive(), is(true)); + assertThat(ContextManager.activeSpan() == callerSpan, is(true)); + verify(delegate).completed("ok"); + + ContextManager.stopSpan(callerSpan); + + assertThat(segmentStorage.getTraceSegments().size(), is(1)); + } + + @Test + public void failedKeepsCallerSpanActive() { + AbstractSpan callerSpan = ContextManager.createEntrySpan("/business", null); + AsyncExitSpan exitSpan = createStartedExitSpan(); + Exception cause = new RuntimeException("boom"); + + new FutureCallbackWrapper<>(delegate, exitSpan).failed(cause); + + assertThat(ContextManager.isActive(), is(true)); + assertThat(ContextManager.activeSpan() == callerSpan, is(true)); + verify(delegate).failed(cause); + + ContextManager.stopSpan(callerSpan); + + assertThat(segmentStorage.getTraceSegments().size(), is(1)); + } + + @Test + public void cancelledKeepsCallerSpanActive() { + AbstractSpan callerSpan = ContextManager.createEntrySpan("/business", null); + AsyncExitSpan exitSpan = createStartedExitSpan(); + + new FutureCallbackWrapper<>(delegate, exitSpan).cancelled(); + + assertThat(ContextManager.isActive(), is(true)); + assertThat(ContextManager.activeSpan() == callerSpan, is(true)); + verify(delegate).cancelled(); + + ContextManager.stopSpan(callerSpan); + + assertThat(segmentStorage.getTraceSegments().size(), is(1)); + } +} \ No newline at end of file