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

Reverts impl to the refactored one

Browse Source 
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 bee61a9988.
This commit is contained in:
Marcin Grzejszczak
2019-09-19 09:57:06 +02:00
parent 8afa3e8d37
commit 75dcc87bd5
2 changed files with 242 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<ClientRequest.Builder, String> SETTER = new Propagation.Setter<ClientRequest.Builder, String>() {
@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<? super Publisher<DataBuffer>, ? extends Publisher<DataBuffer>> 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<ClientResponse> filter(ClientRequest request, ExchangeFunction next) {
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;
}
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<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(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<ClientRequest, ClientResponse> {

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

@@ -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<zipkin2.Span> 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<String, String> home(@RequestHeader HttpHeaders headers) {
Map<String, String> map = new HashMap<>();