diff --git a/CHANGES.md b/CHANGES.md index 45a5d270c..6e7639cbf 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,3 +1,4 @@ + Changes by Version ================== Release Notes. @@ -9,6 +10,7 @@ Release Notes. * Remove plugin for ServiceComb Java Chassis 0.x * Add Guava EventBus plugin. * Fix Dubbo 3.x plugin's tracing problem. +* Fix the bug that maybe generate multiple trace when invoke http request by spring webflux webclient. * Support Druid Connection pool metrics collecting. * Support HikariCP Connection pool metrics collecting. * Support Dbcp2 Connection pool metrics collecting. diff --git a/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-webflux-5.x-webclient-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v5/webclient/WebFluxWebClientInterceptor.java b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-webflux-5.x-webclient-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v5/webclient/WebFluxWebClientInterceptor.java index dd386d3d2..ce1b37912 100644 --- a/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-webflux-5.x-webclient-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v5/webclient/WebFluxWebClientInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-webflux-5.x-webclient-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v5/webclient/WebFluxWebClientInterceptor.java @@ -20,6 +20,7 @@ package org.apache.skywalking.apm.plugin.spring.webflux.v5.webclient; 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; @@ -34,40 +35,13 @@ import reactor.core.publisher.Mono; import java.lang.reflect.Method; import java.net.URI; +import java.util.Optional; public class WebFluxWebClientInterceptor implements InstanceMethodsAroundInterceptorV2 { @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, MethodInvocationContext context) throws Throwable { - if (allArguments[0] == null) { - //illegal args,can't trace ignore - return; - } - ClientRequest request = (ClientRequest) allArguments[0]; - final ContextCarrier contextCarrier = new ContextCarrier(); - - URI uri = request.url(); - final String requestURIString = getRequestURIString(uri); - final String operationName = requestURIString; - final String remotePeer = getIPAndPort(uri); - AbstractSpan span = ContextManager.createExitSpan(operationName, contextCarrier, remotePeer); - - //set components name - span.setComponent(ComponentsDefine.SPRING_WEBCLIENT); - Tags.URL.set(span, uri.toString()); - Tags.HTTP.METHOD.set(span, request.method().toString()); - SpanLayer.asHttp(span); - - if (request instanceof EnhancedInstance) { - ((EnhancedInstance) request).setSkyWalkingDynamicField(contextCarrier); - } - - //user async interface - span.prepareForAsync(); - ContextManager.stopSpan(); - - context.setContext(span); } @Override @@ -77,19 +51,45 @@ public class WebFluxWebClientInterceptor implements InstanceMethodsAroundInterce return ret; } Mono ret1 = (Mono) ret; - AbstractSpan span = (AbstractSpan) context.getContext(); - return ret1.doOnSuccess(clientResponse -> { - HttpStatus httpStatus = clientResponse.statusCode(); - if (httpStatus != null) { - Tags.HTTP_RESPONSE_STATUS_CODE.set(span, httpStatus.value()); - if (httpStatus.isError()) { - span.errorOccurred(); - } + return Mono.subscriberContext().flatMap(ctx -> { + + ClientRequest request = (ClientRequest) allArguments[0]; + URI uri = request.url(); + final String operationName = getRequestURIString(uri); + final String remotePeer = getIPAndPort(uri); + AbstractSpan span = ContextManager.createExitSpan(operationName, remotePeer); + + // get ContextSnapshot from reactor context, the snapshot is set to reactor context by any other plugin + // such as DispatcherHandlerHandleMethodInterceptor in spring-webflux-5.x-plugin + final Optional optional = ctx.getOrEmpty("SKYWALKING_CONTEXT_SNAPSHOT"); + optional.ifPresent(snapshot -> ContextManager.continued((ContextSnapshot) snapshot)); + + //set components name + span.setComponent(ComponentsDefine.SPRING_WEBCLIENT); + Tags.URL.set(span, uri.toString()); + Tags.HTTP.METHOD.set(span, request.method().toString()); + SpanLayer.asHttp(span); + + final ContextCarrier contextCarrier = new ContextCarrier(); + ContextManager.inject(contextCarrier); + if (request instanceof EnhancedInstance) { + ((EnhancedInstance) request).setSkyWalkingDynamicField(contextCarrier); } - }).doOnError(error -> { - span.log(error); - }).doFinally(s -> { - span.asyncFinish(); + + //user async interface + span.prepareForAsync(); + ContextManager.stopSpan(); + return ret1.doOnSuccess(clientResponse -> { + HttpStatus httpStatus = clientResponse.statusCode(); + if (httpStatus != null) { + Tags.HTTP_RESPONSE_STATUS_CODE.set(span, httpStatus.value()); + if (httpStatus.isError()) { + span.errorOccurred(); + } + } + }).doOnError(span::log).doFinally(s -> { + span.asyncFinish(); + }); }); } diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/spring-webflux-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v5/DispatcherHandlerHandleMethodInterceptor.java b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-webflux-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v5/DispatcherHandlerHandleMethodInterceptor.java index e40584e0a..c86c6ea88 100644 --- a/apm-sniffer/optional-plugins/optional-spring-plugins/spring-webflux-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v5/DispatcherHandlerHandleMethodInterceptor.java +++ b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-webflux-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v5/DispatcherHandlerHandleMethodInterceptor.java @@ -18,8 +18,6 @@ package org.apache.skywalking.apm.plugin.spring.webflux.v5; -import java.lang.reflect.Method; -import java.util.List; 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; @@ -41,6 +39,9 @@ import org.springframework.web.server.adapter.DefaultServerWebExchange; import org.springframework.web.util.pattern.PathPattern; import reactor.core.publisher.Mono; +import java.lang.reflect.Method; +import java.util.List; + public class DispatcherHandlerHandleMethodInterceptor implements InstanceMethodsAroundInterceptor { @Override @@ -74,7 +75,7 @@ public class DispatcherHandlerHandleMethodInterceptor implements InstanceMethods span.prepareForAsync(); ContextManager.stopSpan(span); - exchange.getAttributes().put("SKYWALING_SPAN", span); + exchange.getAttributes().put("SKYWALKING_SPAN", span); } private void maybeSetPattern(AbstractSpan span, ServerWebExchange exchange) { @@ -92,28 +93,37 @@ public class DispatcherHandlerHandleMethodInterceptor implements InstanceMethods @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) throws Throwable { + ServerWebExchange exchange = (ServerWebExchange) allArguments[0]; - AbstractSpan span = (AbstractSpan) exchange.getAttributes().get("SKYWALING_SPAN"); + AbstractSpan span = (AbstractSpan) exchange.getAttributes().get("SKYWALKING_SPAN"); + Mono monoReturn = (Mono) ret; - return ((Mono) ret).doFinally(s -> { + // add skywalking context snapshot to reactor context. + EnhancedInstance instance = getInstance(allArguments[0]); + if (instance != null && instance.getSkyWalkingDynamicField() != null) { + monoReturn = monoReturn.subscriberContext( + c -> c.put("SKYWALKING_CONTEXT_SNAPSHOT", instance.getSkyWalkingDynamicField())); + } - if (span != null) { - maybeSetPattern(span, exchange); - try { + return monoReturn.doFinally(s -> { - HttpStatus httpStatus = exchange.getResponse().getStatusCode(); - // fix webflux-2.0.0-2.1.0 version have bug. httpStatus is null. not support - if (httpStatus != null) { - Tags.HTTP_RESPONSE_STATUS_CODE.set(span, httpStatus.value()); - if (httpStatus.isError()) { - span.errorOccurred(); - } - } - } finally { - span.asyncFinish(); + if (span != null) { + maybeSetPattern(span, exchange); + try { + + HttpStatus httpStatus = exchange.getResponse().getStatusCode(); + // fix webflux-2.0.0-2.1.0 version have bug. httpStatus is null. not support + if (httpStatus != null) { + Tags.HTTP_RESPONSE_STATUS_CODE.set(span, httpStatus.value()); + if (httpStatus.isError()) { + span.errorOccurred(); } } + } finally { + span.asyncFinish(); + } + } }); }