SpringCloud/spring-cloud-sleuth
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added an aspect around Netty's WebClient#request method

Browse Source 
fixes gh-806
This commit is contained in:
Marcin Grzejszczak
2018-03-02 00:54:36 +01:00
parent 0e50456ab0
commit 7428609ff7
4 changed files with 216 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
docs/src/main/asciidoc/spring-cloud-sleuth.adoc
View File

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

20
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebAspect.java
View File

@@ -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
* <p/>
@@ -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();
}
}
}

177
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java
View File

@@ -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;
}
}
}
@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<? super HttpClientRequest, ? extends Publisher<Void>> handler) { } // NOSONAR
@Around("anyHttpClientRequestSending(method, url, handler)")
public Object wrapHttpClientRequestSending(ProceedingJoinPoint pjp,
HttpMethod method,
String url, Function<? super HttpClientRequest, ? extends Publisher<Void>> 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<HttpHeaders, String> SETTER = new Propagation.Setter<HttpHeaders, String>() {
@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<HttpHeaders, String> GETTER = new Propagation.Getter<HttpHeaders, String>() {
@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<HttpClientRequest, HttpClientResponse> handler;
final TraceContext.Injector<HttpHeaders> 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<? super HttpClientRequest, ? extends Publisher<Void>> handler) throws Throwable {
// add headers and set CS
final Span currentSpan = this.tracer.currentSpan();
final AtomicReference<Span> span = new AtomicReference<>();
final AtomicBoolean requestAlreadyInstrumented = new AtomicBoolean();
Function<HttpClientRequest, Publisher<Void>> 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<HttpClientResponse> responseMono =
(Mono<HttpClientResponse>) 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<Void> handle(
Function<? super HttpClientRequest, ? extends Publisher<Void>> handler,
HttpClientRequest req) {
if (handler != null) {
return handler.apply(req);
}
return req;
}
static final class HttpAdapter
extends brave.http.HttpClientAdapter<HttpClientRequest, HttpClientResponse> {
@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();
}
}
}

34
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java
View File

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