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

Came back to previous impl for WebClient instrumentation; fixes gh-1442

Browse Source
This commit is contained in:
Marcin Grzejszczak
2019-09-18 11:52:54 +02:00
parent 6da8b91bcd
commit ae0b4fe615
2 changed files with 82 additions and 330 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -19,43 +19,25 @@ 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.Flux;
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.ParameterizedTypeReference;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseCookie;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.util.MultiValueMap;
import org.springframework.web.client.RestClientException;
import org.springframework.web.reactive.function.BodyExtractor;
import org.springframework.web.reactive.function.client.ClientRequest;
import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
import org.springframework.web.reactive.function.client.ExchangeFunction;
import org.springframework.web.reactive.function.client.ExchangeStrategies;
import org.springframework.web.reactive.function.client.WebClient;
/**
@@ -108,6 +90,9 @@ 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<ClientRequest.Builder, String> SETTER = new Propagation.Setter<ClientRequest.Builder, String>() {
@Override
public void put(ClientRequest.Builder carrier, String key, String value) {
@@ -126,14 +111,12 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction {
}
};
private static final String CLIENT_SPAN_KEY = "sleuth.webclient.clientSpan";
private static final String CANCELLED_SUBSCRIPTION_ERROR = "CANCELLED";
public static ExchangeFilterFunction create(BeanFactory beanFactory) {
return new TraceExchangeFilterFunction(beanFactory);
}
final BeanFactory beanFactory;
final Function<? super Publisher<DataBuffer>, ? extends Publisher<DataBuffer>> scopePassingTransformer;
Tracer tracer;
HttpTracing httpTracing;
@@ -144,27 +127,86 @@ 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<ClientResponse> filter(ClientRequest request, ExchangeFunction next) {
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);
}
final ClientRequest.Builder builder = ClientRequest.from(request);
Mono<ClientResponse> 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<ClientResponse> 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;
}
return new MonoWebClientTrace(next, builder.build(), this, span);
private void handleReceive(Span clientSpan, Tracer.SpanInScope ws,
ClientResponse clientResponse, Throwable throwable) {
handler().handleReceive(clientResponse, throwable, clientSpan);
ws.close();
}
@SuppressWarnings("unchecked")
@@ -199,296 +241,6 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction {
return this.injector;
}
private static final class MonoWebClientTrace extends Mono<ClientResponse> {
final ExchangeFunction next;
final ClientRequest request;
final Tracer tracer;
final HttpClientHandler<ClientRequest, ClientResponse> handler;
final TraceContext.Injector<ClientRequest.Builder> injector;
final Tracing tracing;
final Function<? super Publisher<DataBuffer>, ? extends Publisher<DataBuffer>> 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<? super ClientResponse> subscriber) {
Context context = subscriber.currentContext();
this.next.exchange(request).subscribe(
new WebClientTracerSubscriber(subscriber, context, span, this));
}
static final class WebClientTracerSubscriber
implements CoreSubscriber<ClientResponse> {
final CoreSubscriber<? super ClientResponse> actual;
final Context context;
final Span span;
final Tracer.SpanInScope ws;
final HttpClientHandler<ClientRequest, ClientResponse> handler;
final Function<? super Publisher<DataBuffer>, ? extends Publisher<DataBuffer>> scopePassingTransformer;
final Tracing tracing;
boolean done;
WebClientTracerSubscriber(CoreSubscriber<? super ClientResponse> 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(wrapped(response));
}
finally {
terminateSpan(response, null);
}
}
// TODO: Remove once fixed
// https://github.com/spring-projects/spring-framework/issues/23366
private ClientResponse wrapped(ClientResponse response) {
return new ClientResponse() {
@Override
public HttpStatus statusCode() {
try {
return response.statusCode();
}
catch (IllegalArgumentException ex) {
return null;
}
}
@Override
public int rawStatusCode() {
return response.rawStatusCode();
}
@Override
public Headers headers() {
return response.headers();
}
@Override
public MultiValueMap<String, ResponseCookie> cookies() {
return response.cookies();
}
@Override
public ExchangeStrategies strategies() {
return response.strategies();
}
@Override
public <T> T body(
BodyExtractor<T, ? super ClientHttpResponse> extractor) {
return response.body(extractor);
}
@Override
public <T> Mono<T> bodyToMono(Class<? extends T> elementClass) {
return response.bodyToMono(elementClass);
}
@Override
public <T> Mono<T> bodyToMono(
ParameterizedTypeReference<T> typeReference) {
return response.bodyToMono(typeReference);
}
@Override
public <T> Flux<T> bodyToFlux(Class<? extends T> elementClass) {
return (Flux<T>) response.bodyToFlux(DataBuffer.class)
.transform(scopePassingTransformer);
}
@Override
public <T> Flux<T> bodyToFlux(
ParameterizedTypeReference<T> typeReference) {
return (Flux<T>) response.bodyToFlux(DataBuffer.class)
.transform(scopePassingTransformer);
}
@Override
public <T> Mono<ResponseEntity<T>> toEntity(Class<T> bodyType) {
return response.toEntity(bodyType);
}
@Override
public <T> Mono<ResponseEntity<T>> toEntity(
ParameterizedTypeReference<T> typeReference) {
return response.toEntity(typeReference);
}
@Override
public <T> Mono<ResponseEntity<List<T>>> toEntityList(
Class<T> elementType) {
return response.toEntityList(elementType);
}
@Override
public <T> Mono<ResponseEntity<List<T>>> toEntityList(
ParameterizedTypeReference<T> typeReference) {
return response.toEntityList(typeReference);
}
};
}
@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) {
if (log.isDebugEnabled()) {
log.debug("No response was returned. Will close the span ["
+ this.span + "]");
}
handleReceive(this.span, this.ws, clientResponse, throwable);
return;
}
int statusCode = statusCodeAsInt(clientResponse);
boolean error = isError(statusCode);
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 ["
+ statusCode + "] and the reason is ["
+ reasonPhrase(clientResponse) + "]");
}
handleReceive(this.span, this.ws, clientResponse, throwable);
}
private String reasonPhrase(ClientResponse clientResponse) {
try {
return clientResponse.statusCode().getReasonPhrase();
}
catch (IllegalArgumentException ex) {
return "";
}
}
private boolean isError(int code) {
return code >= 400;
}
private int statusCodeAsInt(ClientResponse response) {
try {
return response.rawStatusCode();
}
catch (Exception dontCare) {
return 0;
}
}
}
}
static final class HttpAdapter
extends brave.http.HttpClientAdapter<ClientRequest, ClientResponse> {

2
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/feign/issues/issue362/Issue362Tests.java
View File

@@ -209,7 +209,7 @@ class CustomConfig {
this.feignComponentAsserter.executedComponents.put(ErrorDecoder.class, true);
if (response.status() == 409) {
return new RetryableException(response.status(), "Article not Ready",
Request.HttpMethod.GET, new Date());
Request.HttpMethod.GET, new Date(), response.request());
}
else {
return super.decode(methodKey, response);