From 3c0932bb321dff7f8876ab63e8d5b92887903d3e Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Thu, 6 Feb 2020 15:02:12 +0800 Subject: [PATCH] Fixes null bugs on WebClient canceled request (#1548) --- .../TraceWebClientBeanPostProcessor.java | 36 +++++++++++-------- .../client/integration/WebClientTests.java | 23 ++++++++++-- 2 files changed, 42 insertions(+), 17 deletions(-) diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java index 558f384cf..e50f31fc0 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java @@ -18,6 +18,7 @@ package org.springframework.cloud.sleuth.instrument.web.client; import java.util.Collections; import java.util.List; +import java.util.concurrent.CancellationException; import java.util.function.Consumer; import java.util.function.Function; @@ -131,9 +132,14 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { } }; - private static final String CLIENT_SPAN_KEY = "sleuth.webclient.clientSpan"; + static final String CLIENT_SPAN_KEY = "sleuth.webclient.clientSpan"; - private static final String CANCELLED_SUBSCRIPTION_ERROR = "CANCELLED"; + static final Exception CANCELLED_ERROR = new CancellationException("CANCELLED") { + @Override + public Throwable fillInStackTrace() { + return this; // stack trace doesn't add value here + } + }; final ConfigurableApplicationContext springContext; @@ -170,6 +176,8 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { } MonoWebClientTrace trace = new MonoWebClientTrace(next, wrapper.buildRequest(), this, span); + // TODO: investigate why this commit leaks a scope: + // 8f5bcdabd7af23df443e771432eb85597f3b3076 tracer().withSpanInScope(parentSpan); return trace; } @@ -356,13 +364,14 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { return this.context; } - void handleReceive(Span clientSpan, ClientResponse clientResponse, - Throwable throwable) { + void handleReceive(Span clientSpan, @Nullable ClientResponse res, + @Nullable Throwable error) { if (log.isTraceEnabled()) { log.trace("Handling receive"); } - this.handler.handleReceive(new HttpClientResponse(clientResponse), - throwable, clientSpan); + HttpClientResponse response = res != null ? new HttpClientResponse(res) + : null; + this.handler.handleReceive(response, error, clientSpan); if (log.isTraceEnabled()) { log.trace("Closed scope"); } @@ -374,32 +383,31 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { + this.span + "]"); } - this.span.tag("error", CANCELLED_SUBSCRIPTION_ERROR); - handleReceive(this.span, null, null); + handleReceive(this.span, null, CANCELLED_ERROR); } void terminateSpan(@Nullable ClientResponse clientResponse, - @Nullable Throwable throwable) { + @Nullable Throwable error) { if (clientResponse == null) { if (log.isDebugEnabled()) { log.debug("No response was returned. Will close the span [" + this.span + "]"); } - handleReceive(this.span, clientResponse, throwable); + handleReceive(this.span, null, error); return; } int statusCode = clientResponse.rawStatusCode(); - boolean error = statusCode >= 400; - if (error) { + boolean isHttpError = statusCode >= 400; + if (isHttpError) { if (log.isDebugEnabled()) { log.debug( "Non positive status code was returned from the call. Will close the span [" + this.span + "]"); } - throwable = new RestClientException( + error = new RestClientException( "Status code of the response is [" + statusCode + "]"); } - handleReceive(this.span, clientResponse, throwable); + handleReceive(this.span, clientResponse, error); } } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java index 26962aa66..3c47698aa 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java @@ -40,6 +40,7 @@ import com.netflix.loadbalancer.ILoadBalancer; import com.netflix.loadbalancer.Server; import junitparams.JUnitParamsRunner; import junitparams.Parameters; +import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.http.HttpResponse; import org.apache.http.client.methods.HttpGet; @@ -55,6 +56,8 @@ import org.junit.ClassRule; import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; +import org.reactivestreams.Subscription; +import reactor.core.publisher.BaseSubscriber; import zipkin2.Annotation; import zipkin2.reporter.Reporter; @@ -112,8 +115,7 @@ public class WebClientTests { static final String SAMPLED_NAME = "X-B3-Sampled"; static final String PARENT_ID_NAME = "X-B3-ParentSpanId"; - private static final org.apache.commons.logging.Log log = LogFactory - .getLog(WebClientTests.class); + private static final Log log = LogFactory.getLog(WebClientTests.class); @Rule public final SpringMethodRule springMethodRule = new SpringMethodRule(); @@ -351,7 +353,7 @@ public class WebClientTests { @Test @SuppressWarnings("unchecked") - public void shouldWorkWhenCustomStatusCodeIsReturned() throws InterruptedException { + public void shouldWorkWhenCustomStatusCodeIsReturned() { Span span = this.tracer.nextSpan().name("foo").start(); try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { @@ -370,6 +372,21 @@ public class WebClientTests { .contains("CLIENT"); } + @Test + public void shouldTagOnCancel() { + this.webClient.get().uri("http://localhost:" + this.port + "/doNotSkip") + .retrieve().bodyToMono(String.class) + .subscribe(new BaseSubscriber() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + cancel(); + } + }); + + then(this.reporter.getSpans()).isNotEmpty(); + then(this.reporter.getSpans().get(0).tags()).containsEntry("error", "CANCELLED"); + } + @Test public void shouldRespectSkipPattern() { this.webClient.get().uri("http://localhost:" + this.port + "/skip").retrieve()