From 7428609ff7e401c70640f646ce64c5e7d59d7319 Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Fri, 2 Mar 2018 00:54:36 +0100 Subject: [PATCH] Added an aspect around Netty's WebClient#request method fixes gh-806 --- .../main/asciidoc/spring-cloud-sleuth.adoc | 9 + .../sleuth/instrument/web/TraceWebAspect.java | 20 +- .../TraceWebClientAutoConfiguration.java | 177 +++++++++++++++++- .../client/integration/WebClientTests.java | 34 +++- 4 files changed, 216 insertions(+), 24 deletions(-) diff --git a/docs/src/main/asciidoc/spring-cloud-sleuth.adoc b/docs/src/main/asciidoc/spring-cloud-sleuth.adoc index 665ceac4c..5cd9eb560 100644 --- a/docs/src/main/asciidoc/spring-cloud-sleuth.adoc +++ b/docs/src/main/asciidoc/spring-cloud-sleuth.adoc @@ -1023,6 +1023,15 @@ tracing context gets injected to the sent requests. To block these features, set `spring.sleuth.web.client.enabled` to `false`. +==== Netty `HttpClient` + +We instrument the Netty's `HttpClient`. + +To block this feature, set `spring.sleuth.web.client.enabled` to `false`. + +IMPORTANT: You have to register `HttpClient` as a bean so that the instrumentation happens. +If you create a `HttpClient` instance with a `new` keyword, the instrumentation does NOT work. + === Feign By default, Spring Cloud Sleuth provides integration with Feign through `TraceFeignClientAutoConfiguration`. diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebAspect.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebAspect.java index 32f889293..fbb89bda7 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebAspect.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebAspect.java @@ -16,8 +16,6 @@ package org.springframework.cloud.sleuth.instrument.web; -import javax.servlet.http.HttpServletRequest; -import javax.servlet.http.HttpServletResponse; import java.lang.reflect.Field; import java.util.concurrent.Callable; @@ -32,8 +30,6 @@ import org.springframework.cloud.sleuth.SpanNamer; import org.springframework.cloud.sleuth.instrument.async.TraceCallable; import org.springframework.web.context.request.async.WebAsyncTask; -import brave.Span; - /** * Aspect that adds tracing to *

@@ -73,14 +69,12 @@ public class TraceWebAspect { private final Tracer tracer; private final SpanNamer spanNamer; - //private final TraceKeys traceKeys; private final ErrorParser errorParser; - public TraceWebAspect(Tracer tracer, SpanNamer spanNamer, //TraceKeys traceKeys, + public TraceWebAspect(Tracer tracer, SpanNamer spanNamer, ErrorParser errorParser) { this.tracer = tracer; this.spanNamer = spanNamer; - //this.traceKeys = traceKeys; this.errorParser = errorParser; } @@ -99,9 +93,6 @@ public class TraceWebAspect { @Pointcut("execution(public org.springframework.web.context.request.async.WebAsyncTask *(..))") private void anyPublicMethodReturningWebAsyncTask() { } // NOSONAR - @Pointcut("execution(public * org.springframework.web.servlet.HandlerExceptionResolver.resolveException(..)) && args(request, response, handler, ex)") - private void anyHandlerExceptionResolver(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) { } // NOSONAR - @Pointcut("(anyRestControllerAnnotated() || anyControllerAnnotated()) && anyPublicMethodReturningWebAsyncTask()") private void anyControllerOrRestControllerWithPublicWebAsyncTaskMethod() { } // NOSONAR @@ -140,13 +131,4 @@ public class TraceWebAspect { return webAsyncTask; } - @Around("anyHandlerExceptionResolver(request, response, handler, ex)") - public Object markRequestForSpanClosing(ProceedingJoinPoint pjp, - HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Throwable { - Span currentSpan = this.tracer.currentSpan(); - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(currentSpan)){ - return pjp.proceed(); - } - } - } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java index f2d21cfdc..72c19ab33 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java @@ -19,13 +19,31 @@ package org.springframework.cloud.sleuth.instrument.web.client; import java.io.IOException; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; +import brave.Span; +import brave.Tracer; +import brave.http.HttpClientHandler; import brave.http.HttpTracing; import brave.httpasyncclient.TracingHttpAsyncClientBuilder; import brave.httpclient.TracingHttpClientBuilder; +import brave.propagation.Propagation; +import brave.propagation.TraceContext; +import brave.propagation.TraceContextOrSamplingFlags; import brave.spring.web.TracingClientHttpRequestInterceptor; +import io.netty.handler.codec.http.HttpHeaders; +import io.netty.handler.codec.http.HttpMethod; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.http.impl.client.HttpClientBuilder; import org.apache.http.impl.nio.client.HttpAsyncClientBuilder; +import org.aspectj.lang.ProceedingJoinPoint; +import org.aspectj.lang.annotation.Around; +import org.aspectj.lang.annotation.Aspect; +import org.aspectj.lang.annotation.Pointcut; +import org.reactivestreams.Publisher; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.ListableBeanFactory; @@ -48,6 +66,10 @@ import org.springframework.http.client.ClientHttpRequestInterceptor; import org.springframework.http.client.ClientHttpResponse; import org.springframework.web.client.RestTemplate; import org.springframework.web.reactive.function.client.WebClient; +import reactor.core.publisher.Mono; +import reactor.ipc.netty.http.client.HttpClient; +import reactor.ipc.netty.http.client.HttpClientRequest; +import reactor.ipc.netty.http.client.HttpClientResponse; /** * {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration @@ -117,6 +139,15 @@ public class TraceWebClientAutoConfiguration { return new TraceWebClientBeanPostProcessor(beanFactory); } } + + @Configuration + @ConditionalOnClass(HttpClient.class) + static class NettyConfiguration { + @Bean + public NettyAspect traceNetyAspect(HttpTracing httpTracing) { + return new NettyAspect(httpTracing); + } + } } class RestTemplateInterceptorInjector { @@ -209,4 +240,148 @@ class LazyTracingClientHttpRequestInterceptor implements ClientHttpRequestInterc } return this.interceptor; } -} \ No newline at end of file +} + +@Aspect +class NettyAspect { + + private final TracingHttpClientInstrumentation instrumentation; + + NettyAspect(HttpTracing httpTracing) { + this.instrumentation = TracingHttpClientInstrumentation.create(httpTracing); + } + + @Pointcut("execution(public * reactor.ipc.netty.http.client.HttpClient.request(..)) && args(method, url, handler)") + private void anyHttpClientRequestSending(HttpMethod method, + String url, Function> handler) { } // NOSONAR + + @Around("anyHttpClientRequestSending(method, url, handler)") + public Object wrapHttpClientRequestSending(ProceedingJoinPoint pjp, + HttpMethod method, + String url, Function> handler) throws Throwable { + return this.instrumentation.wrapHttpClientRequestSending(pjp, method, url, handler); + } +} + +class TracingHttpClientInstrumentation { + private static final Log log = LogFactory.getLog(TracingHttpClientInstrumentation.class); + + static final Propagation.Setter SETTER = new Propagation.Setter() { + @Override public void put(HttpHeaders carrier, String key, String value) { + if (!carrier.contains(key)) { + carrier.add(key, value); + } + } + + @Override public String toString() { + return "HttpHeaders::add"; + } + }; + + static final Propagation.Getter GETTER = new Propagation.Getter() { + @Override public String get(HttpHeaders carrier, String key) { + return carrier.get(key); + } + + @Override public String toString() { + return "HttpHeaders::get"; + } + }; + + static TracingHttpClientInstrumentation create(HttpTracing httpTracing) { + return new TracingHttpClientInstrumentation(httpTracing); + } + + final Tracer tracer; + final HttpClientHandler handler; + final TraceContext.Injector injector; + final HttpTracing httpTracing; + + TracingHttpClientInstrumentation(HttpTracing httpTracing) { + this.tracer = httpTracing.tracing().tracer(); + this.handler = HttpClientHandler.create(httpTracing, new HttpAdapter()); + this.injector = httpTracing.tracing().propagation().injector(SETTER); + this.httpTracing = httpTracing; + } + + Object wrapHttpClientRequestSending(ProceedingJoinPoint pjp, + HttpMethod method, + String url, Function> handler) throws Throwable { + // add headers and set CS + final Span currentSpan = this.tracer.currentSpan(); + final AtomicReference span = new AtomicReference<>(); + final AtomicBoolean requestAlreadyInstrumented = new AtomicBoolean(); + Function> combinedFunction = + req -> { + try (Tracer.SpanInScope spanInScope = this.tracer.withSpanInScope(currentSpan)) { + io.netty.handler.codec.http.HttpHeaders headers = req + .requestHeaders(); + TraceContextOrSamplingFlags flags = this.httpTracing.tracing() + .propagation().extractor(GETTER).extract(headers); + if (flags != TraceContextOrSamplingFlags.EMPTY) { + requestAlreadyInstrumented.set(true); + if (log.isDebugEnabled()) { + log.debug("Request already instrumented. Skipping"); + } + return handle(handler, req); + } + span.set(this.handler.handleSend(this.injector, headers, req)); + try (Tracer.SpanInScope clientInScope = this.tracer.withSpanInScope(span.get())) { + if (log.isDebugEnabled()) { + log.debug("Created a new client span for Netty client"); + } + return handle(handler, req); + } + } + }; + // run + Mono responseMono = + (Mono) pjp.proceed(new Object[] { method , url, combinedFunction }); + // get response + return responseMono.doOnSuccessOrError((httpClientResponse, throwable) -> { + if (requestAlreadyInstrumented.get()) { + if (log.isDebugEnabled()) { + log.debug("Request already instrumented. Skipping"); + return; + } + } + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span.get())) { + // status codes and CR + this.handler.handleReceive(httpClientResponse, throwable, span.get()); + if (log.isDebugEnabled()) { + log.debug("Setting client sent spans"); + } + } + }); + } + + private Publisher handle( + Function> handler, + HttpClientRequest req) { + if (handler != null) { + return handler.apply(req); + } + return req; + } + + static final class HttpAdapter + extends brave.http.HttpClientAdapter { + + @Override public String method(HttpClientRequest request) { + return request.method().name(); + } + + @Override public String url(HttpClientRequest request) { + return request.uri(); + } + + @Override public String requestHeader(HttpClientRequest request, String name) { + Object result = request.requestHeaders().get(name); + return result != null ? result.toString() : ""; + } + + @Override public Integer statusCode(HttpClientResponse response) { + return response.status().code(); + } + } +} 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 60bd17ddf..c16c155ec 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 @@ -87,6 +87,8 @@ import org.springframework.web.client.RestTemplate; import org.springframework.web.reactive.function.client.WebClient; import reactor.core.publisher.Hooks; import reactor.core.scheduler.Schedulers; +import reactor.ipc.netty.http.client.HttpClient; +import reactor.ipc.netty.http.client.HttpClientResponse; import zipkin2.Annotation; import zipkin2.reporter.Reporter; @@ -116,10 +118,9 @@ public class WebClientTests { @Autowired @LoadBalanced RestTemplate template; @Autowired WebClient webClient; @Autowired WebClient.Builder webClientBuilder; - // #845 - @Autowired HttpClientBuilder httpClientBuilder; - // #845 - @Autowired HttpAsyncClientBuilder httpAsyncClientBuilder; + @Autowired HttpClientBuilder httpClientBuilder; // #845 + @Autowired HttpClient nettyHttpClient; + @Autowired HttpAsyncClientBuilder httpAsyncClientBuilder; // #845 @Autowired ArrayListSpanReporter reporter; @Autowired Tracer tracer; @Autowired TestErrorController testErrorController; @@ -237,6 +238,27 @@ public class WebClientTests { then(this.reporter.getSpans()).isNotEmpty(); } + @Test + @SuppressWarnings("unchecked") + public void shouldAttachTraceIdWhenCallingAnotherServiceForNettyHttpClient() throws Exception { + Span span = this.tracer.nextSpan().name("foo").start(); + + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { + HttpClientResponse response = this.nettyHttpClient + .get("http://localhost:" + port).block(); + + then(response).isNotNull(); + } finally { + span.finish(); + } + + then(this.tracer.currentSpan()).isNull(); + then(this.reporter.getSpans()).isNotEmpty(); + then(this.reporter.getSpans()) + .extracting("traceId", String.class) + .containsOnly(span.context().traceIdString()); + } + @Test @SuppressWarnings("unchecked") public void shouldAttachTraceIdWhenCallingAnotherServiceForHttpClient() throws Exception { @@ -465,6 +487,10 @@ public class WebClientTests { RestTemplateCustomizer myRestTemplateCustomizer() { return new MyRestTemplateCustomizer(); } + + @Bean HttpClient reactorHttpClient() { + return HttpClient.create(); + } } static class MyRestTemplateCustomizer implements RestTemplateCustomizer {