kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix the trace breaking bugs in WebFlux server and WebClient(#114) 

* Fix the bug that maybe generate multiple traces when invoking HTTP requests by Spring WebFlux server and WebClient.
This commit is contained in:
Zimmem 2022-03-01 20:47:18 +08:00 committed by GitHub
parent bf126a6afb
commit 2bd58bce7a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 70 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.md
View File

@ -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.

80
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
View File

@ -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<ClientResponse> ret1 = (Mono<ClientResponse>) 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<Object> 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();
});
});
}

46
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
View File

@ -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<Void> monoReturn = (Mono<Void>) 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();
}
}
});
}