From 75dcc87bd5eaae2b0248de2bd7336138003ba109 Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Thu, 19 Sep 2019 09:57:06 +0200 Subject: [PATCH 1/2] Reverts impl to the refactored one Revert "Not using httpStatus() method for custom status codes" This reverts commit b59277f0 Revert "Came back to previous impl for WebClient instrumentation; fixes gh-1442" This reverts commit bee61a998853f7e3733ed539c89543a1f7a7cfed. --- .../TraceWebClientBeanPostProcessor.java | 295 +++++++++++++----- .../client/integration/WebClientTests.java | 34 +- 2 files changed, 242 insertions(+), 87 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 123ecaf95..7362707f1 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 @@ -19,20 +19,29 @@ package org.springframework.cloud.sleuth.instrument.web.client; import java.util.Collections; import java.util.List; import java.util.function.Consumer; +import java.util.function.Function; import brave.Span; import brave.Tracer; +import brave.Tracing; import brave.http.HttpClientHandler; import brave.http.HttpTracing; import brave.propagation.Propagation; import brave.propagation.TraceContext; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscription; +import reactor.core.CoreSubscriber; import reactor.core.publisher.Mono; +import reactor.util.annotation.Nullable; +import reactor.util.context.Context; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth; +import org.springframework.core.io.buffer.DataBuffer; import org.springframework.web.client.RestClientException; import org.springframework.web.reactive.function.client.ClientRequest; import org.springframework.web.reactive.function.client.ClientResponse; @@ -90,9 +99,6 @@ final class TraceWebClientBeanPostProcessor implements BeanPostProcessor { final class TraceExchangeFilterFunction implements ExchangeFilterFunction { private static final Log log = LogFactory.getLog(TraceExchangeFilterFunction.class); - - private static final String CLIENT_SPAN_KEY = "sleuth.webclient.clientSpan"; - static final Propagation.Setter SETTER = new Propagation.Setter() { @Override public void put(ClientRequest.Builder carrier, String key, String value) { @@ -111,12 +117,14 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { } }; - public static ExchangeFilterFunction create(BeanFactory beanFactory) { - return new TraceExchangeFilterFunction(beanFactory); - } + private static final String CLIENT_SPAN_KEY = "sleuth.webclient.clientSpan"; + + private static final String CANCELLED_SUBSCRIPTION_ERROR = "CANCELLED"; final BeanFactory beanFactory; + final Function, ? extends Publisher> scopePassingTransformer; + Tracer tracer; HttpTracing httpTracing; @@ -127,86 +135,27 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { TraceExchangeFilterFunction(BeanFactory beanFactory) { this.beanFactory = beanFactory; + this.scopePassingTransformer = ReactorSleuth + .scopePassingSpanOperator(beanFactory); + } + + public static ExchangeFilterFunction create(BeanFactory beanFactory) { + return new TraceExchangeFilterFunction(beanFactory); } @Override public Mono filter(ClientRequest request, ExchangeFunction next) { - final ClientRequest.Builder builder = ClientRequest.from(request); - Mono exchange = Mono.defer(() -> next.exchange(builder.build())) - .cast(Object.class).onErrorResume(Mono::just) - .zipWith(Mono.subscriberContext()).flatMap(anyAndContext -> { - if (log.isDebugEnabled()) { - log.debug("Wrapping the context [" + anyAndContext + "]"); - } - Object any = anyAndContext.getT1(); - Span clientSpan = anyAndContext.getT2().get(CLIENT_SPAN_KEY); - Mono continuation; - final Tracer.SpanInScope ws = tracer().withSpanInScope(clientSpan); - if (any instanceof Throwable) { - continuation = Mono.error((Throwable) any); - } - else { - continuation = Mono.just((ClientResponse) any); - } - return continuation - .doAfterSuccessOrError((clientResponse, throwable1) -> { - Throwable throwable = throwable1; - if (clientResponse == null - || clientResponse.statusCode() == null) { - if (log.isDebugEnabled()) { - log.debug( - "No response was returned. Will close the span [" - + clientSpan + "]"); - } - handleReceive(clientSpan, ws, clientResponse, - throwable); - return; - } - boolean error = clientResponse.statusCode() - .is4xxClientError() - || clientResponse.statusCode().is5xxServerError(); - if (error) { - if (log.isDebugEnabled()) { - log.debug( - "Non positive status code was returned from the call. Will close the span [" - + clientSpan + "]"); - } - throwable = new RestClientException( - "Status code of the response is [" - + clientResponse.statusCode().value() - + "] and the reason is [" - + clientResponse.statusCode() - .getReasonPhrase() - + "]"); - } - handleReceive(clientSpan, ws, clientResponse, throwable); - }); - }).subscriberContext(c -> { - if (log.isDebugEnabled()) { - log.debug("Instrumenting WebClient call"); - } - Span parent = c.getOrDefault(Span.class, null); - Span clientSpan = handler().handleSend(injector(), builder, request, - tracer().nextSpan()); - if (log.isDebugEnabled()) { - log.debug("Handled send of " + clientSpan); - } - if (parent == null) { - c = c.put(Span.class, clientSpan); - if (log.isDebugEnabled()) { - log.debug("Reactor Context got injected with the client span " - + clientSpan); - } - } - return c.put(CLIENT_SPAN_KEY, clientSpan); - }); - return exchange; - } + ClientRequest.Builder builder = ClientRequest.from(request); + if (log.isDebugEnabled()) { + log.debug("Instrumenting WebClient call"); + } + Span span = handler().handleSend(injector(), builder, request, + tracer().nextSpan()); + if (log.isDebugEnabled()) { + log.debug("Handled send of " + span); + } - private void handleReceive(Span clientSpan, Tracer.SpanInScope ws, - ClientResponse clientResponse, Throwable throwable) { - handler().handleReceive(clientResponse, throwable, clientSpan); - ws.close(); + return new MonoWebClientTrace(next, builder.build(), this, span); } @SuppressWarnings("unchecked") @@ -241,6 +190,190 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { return this.injector; } + private static final class MonoWebClientTrace extends Mono { + + final ExchangeFunction next; + + final ClientRequest request; + + final Tracer tracer; + + final HttpClientHandler handler; + + final TraceContext.Injector injector; + + final Tracing tracing; + + final Function, ? extends Publisher> scopePassingTransformer; + + private final Span span; + + MonoWebClientTrace(ExchangeFunction next, ClientRequest request, + TraceExchangeFilterFunction parent, Span span) { + this.next = next; + this.request = request; + this.tracer = parent.tracer(); + this.handler = parent.handler(); + this.injector = parent.injector(); + this.tracing = parent.httpTracing().tracing(); + this.scopePassingTransformer = parent.scopePassingTransformer; + this.span = span; + } + + @Override + public void subscribe(CoreSubscriber subscriber) { + + Context context = subscriber.currentContext(); + + this.next.exchange(request).subscribe( + new WebClientTracerSubscriber(subscriber, context, span, this)); + } + + static final class WebClientTracerSubscriber + implements CoreSubscriber { + + final CoreSubscriber actual; + + final Context context; + + final Span span; + + final Tracer.SpanInScope ws; + + final HttpClientHandler handler; + + final Function, ? extends Publisher> scopePassingTransformer; + + final Tracing tracing; + + boolean done; + + WebClientTracerSubscriber(CoreSubscriber actual, + Context context, Span span, MonoWebClientTrace parent) { + this.actual = actual; + this.span = span; + this.handler = parent.handler; + this.tracing = parent.tracing; + this.scopePassingTransformer = parent.scopePassingTransformer; + + if (!context.hasKey(Span.class)) { + context = context.put(Span.class, span); + if (log.isDebugEnabled()) { + log.debug("Reactor Context got injected with the client span " + + span); + } + } + + this.context = context.put(CLIENT_SPAN_KEY, span); + this.ws = parent.tracer.withSpanInScope(span); + + } + + @Override + public void onSubscribe(Subscription subscription) { + this.actual.onSubscribe(new Subscription() { + @Override + public void request(long n) { + subscription.request(n); + } + + @Override + public void cancel() { + terminateSpanOnCancel(); + subscription.cancel(); + } + }); + } + + @Override + public void onNext(ClientResponse response) { + this.done = true; + try { + // decorate response body + this.actual + .onNext(ClientResponse.from(response) + .body(response.bodyToFlux(DataBuffer.class) + .transform(this.scopePassingTransformer)) + .build()); + } + finally { + terminateSpan(response, null); + } + } + + @Override + public void onError(Throwable t) { + try { + this.actual.onError(t); + } + finally { + terminateSpan(null, t); + } + } + + @Override + public void onComplete() { + try { + this.actual.onComplete(); + } + finally { + if (!this.done) { + terminateSpan(null, null); + } + } + } + + @Override + public Context currentContext() { + return this.context; + } + + void handleReceive(Span clientSpan, Tracer.SpanInScope ws, + ClientResponse clientResponse, Throwable throwable) { + this.handler.handleReceive(clientResponse, throwable, clientSpan); + ws.close(); + } + + void terminateSpanOnCancel() { + if (log.isDebugEnabled()) { + log.debug("Subscription was cancelled. Will close the span [" + + this.span + "]"); + } + + this.span.tag("error", CANCELLED_SUBSCRIPTION_ERROR); + handleReceive(this.span, this.ws, null, null); + } + + void terminateSpan(@Nullable ClientResponse clientResponse, + @Nullable Throwable throwable) { + if (clientResponse == null || clientResponse.statusCode() == null) { + if (log.isDebugEnabled()) { + log.debug("No response was returned. Will close the span [" + + this.span + "]"); + } + handleReceive(this.span, this.ws, clientResponse, throwable); + return; + } + boolean error = clientResponse.statusCode().is4xxClientError() + || clientResponse.statusCode().is5xxServerError(); + if (error) { + if (log.isDebugEnabled()) { + log.debug( + "Non positive status code was returned from the call. Will close the span [" + + this.span + "]"); + } + throwable = new RestClientException("Status code of the response is [" + + clientResponse.statusCode().value() + + "] and the reason is [" + + clientResponse.statusCode().getReasonPhrase() + "]"); + } + handleReceive(this.span, this.ws, clientResponse, throwable); + } + + } + + } + static final class HttpAdapter extends brave.http.HttpClientAdapter { 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 1d4ae877d..917d6f19c 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 @@ -16,6 +16,7 @@ package org.springframework.cloud.sleuth.instrument.web.client.integration; +import java.time.Duration; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -52,6 +53,7 @@ import org.awaitility.Awaitility; import org.junit.After; import org.junit.Before; import org.junit.ClassRule; +import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; @@ -375,6 +377,32 @@ public class WebClientTests { .contains("CLIENT"); } + @Test + @Ignore("Flakey on CI") + public void shouldReportTraceForCancelledRequestViaWebClient() { + Span span = this.tracer.nextSpan().name("foo").start(); + + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { + this.webClient.get().uri("http://localhost:" + this.port + "/noresponse") + .retrieve().bodyToMono(String.class).timeout(Duration.ofMillis(0)) + .block(); + } + catch (Exception e) { + + } + finally { + span.finish(); + } + + Awaitility.await().untilAsserted(() -> { + System.out.println("Found spans " + this.reporter.getSpans()); + final Optional clientSpan = this.reporter.getSpans().stream() + .filter(s -> s.kind() == zipkin2.Span.Kind.CLIENT).findFirst(); + then(clientSpan).isPresent(); + then(clientSpan.get().tags()).containsEntry("error", "CANCELLED"); + }); + } + @Test @SuppressWarnings("unchecked") public void shouldNotBreakWhenCustomStatusCodeIsSetViaWebClient() { @@ -651,12 +679,6 @@ public class WebClientTests { return traceId; } - @RequestMapping(value = "/customstatuscode", method = RequestMethod.GET) - public ResponseEntity customStatusCode() { - this.span = this.tracer.currentSpan(); - return ResponseEntity.status(499).build(); - } - @RequestMapping("/") public Map home(@RequestHeader HttpHeaders headers) { Map map = new HashMap<>(); From 32aef9f1548f19ed7e7a7247e5bcf91c9219f29a Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Thu, 19 Sep 2019 11:42:08 +0200 Subject: [PATCH 2/2] Improved performance of ReactorSleuth; fixes gh-1449 --- .../sleuth/instrument/reactor/ReactorSleuth.java | 12 ++++++++---- .../reactor/ScopePassingSpanSubscriberTests.java | 11 ++++++++++- 2 files changed, 18 insertions(+), 5 deletions(-) diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java index 666f73bb0..e6dbaa3c1 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java @@ -16,6 +16,8 @@ package org.springframework.cloud.sleuth.instrument.reactor; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.function.BooleanSupplier; import java.util.function.Function; @@ -84,8 +86,7 @@ public abstract class ReactorSleuth { + "]"); } - return scopePassingSpanSubscription(beanFactory.getBean(Tracing.class), - sub); + return scopePassingSpanSubscription(beanFactory, sub); } if (log.isTraceEnabled()) { log.trace("Spring Context [" + beanFactory @@ -104,9 +105,12 @@ public abstract class ReactorSleuth { scannable.name()); } - static CoreSubscriber scopePassingSpanSubscription(Tracing tracing, - CoreSubscriber sub) { + private static Map CACHE = new ConcurrentHashMap<>(); + static CoreSubscriber scopePassingSpanSubscription( + BeanFactory beanFactory, CoreSubscriber sub) { + Tracing tracing = CACHE.computeIfAbsent(beanFactory, + beanFactory1 -> beanFactory1.getBean(Tracing.class)); Context context = sub.currentContext(); Span root = context.hasKey(Span.class) ? context.get(Span.class) diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java index f37571e97..fd110e04a 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java @@ -21,11 +21,14 @@ import brave.Tracer; import brave.Tracing; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.BDDMockito; import org.mockito.junit.MockitoJUnitRunner; import reactor.core.CoreSubscriber; import reactor.core.publisher.BaseSubscriber; import reactor.util.context.Context; +import org.springframework.beans.factory.BeanFactory; + import static org.assertj.core.api.BDDAssertions.then; /** @@ -58,7 +61,7 @@ public class ScopePassingSpanSubscriberTests { try (Tracer.SpanInScope ws = this.tracing.tracer() .withSpanInScope(span.start())) { CoreSubscriber subscriber = ReactorSleuth.scopePassingSpanSubscription( - this.tracing, new BaseSubscriber() { + beanFactory(), new BaseSubscriber() { }); then(subscriber.currentContext().get(Span.class)).isEqualTo(span); @@ -66,4 +69,10 @@ public class ScopePassingSpanSubscriberTests { } + private BeanFactory beanFactory() { + BeanFactory beanFactory = BDDMockito.mock(BeanFactory.class); + BDDMockito.given(beanFactory.getBean(Tracing.class)).willReturn(this.tracing); + return beanFactory; + } + }