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

Stephane help (#1126)

Browse Source 
* WIP : Optimize Reactor instrumentation
- Use onLastOperator instead of onEachOperator
- Do not instrument scalar publishers
- Revisit ReactorSleuthMethodInvocationProcessor to reduce ops overhead

* Optimize Reactor instrumentation
- Use only one operator for the MethodInvocationProcessor
- Warning Behavior Change: @NewSpan will defer Span creation/context

* Optimize Reactor Optimization
- Reduce WebFilter instrumentation to 1 operator
- Reduce WebClient instrumentation to 2 operators (1 resp, 1 body)

* Late Formatting commit - because not enough coffee
This commit is contained in:
Marcin Grzejszczak
2018-11-06 23:06:02 +01:00
committed by GitHub
parent 3c25fc1a11
commit d3feae1ce4
12 changed files with 497 additions and 361 deletions
Download Patch File Download Diff File Expand all files Collapse all files

139
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/ReactorSleuthMethodInvocationProcessor.java
View File

@@ -28,7 +28,9 @@ import org.reactivestreams.Subscription;
import reactor.core.CoreSubscriber;
import reactor.core.Scannable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxOperator;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoOperator;
import reactor.util.annotation.Nullable;
import reactor.util.context.Context;
@@ -85,43 +87,37 @@ class ReactorSleuthMethodInvocationProcessor
Publisher<?> publisher = (Publisher) invocation.proceed();
if (publisher instanceof Mono) {
return new MonoSpan((Mono<Object>) publisher,
this,
newSpan,
span,
invocation,
return new MonoSpan((Mono<Object>) publisher, this, newSpan, span, invocation,
log);
}
else if (publisher instanceof Flux) {
return new FluxSpan((Flux<Object>) publisher,
this,
newSpan,
span,
invocation,
return new FluxSpan((Flux<Object>) publisher, this, newSpan, span, invocation,
log);
}
else {
throw new IllegalArgumentException("Unexpected type of publisher: " + publisher.getClass());
throw new IllegalArgumentException(
"Unexpected type of publisher: " + publisher.getClass());
}
}
private static final class FluxSpan extends Flux<Object> implements Scannable {
private static final class FluxSpan extends FluxOperator<Object, Object> {
final Span span;
final MethodInvocation invocation;
final String log;
final boolean hasLog;
final Flux<Object> source;
final Span span;
final MethodInvocation invocation;
final String log;
final boolean hasLog;
final ReactorSleuthMethodInvocationProcessor processor;
final NewSpan newSpan;
FluxSpan(Flux<Object> source,
ReactorSleuthMethodInvocationProcessor processor,
NewSpan newSpan,
@Nullable Span span,
MethodInvocation invocation,
final NewSpan newSpan;
FluxSpan(Flux<Object> source, ReactorSleuthMethodInvocationProcessor processor,
NewSpan newSpan, @Nullable Span span, MethodInvocation invocation,
String log) {
this.source = source;
super(source);
this.span = span;
this.newSpan = newSpan;
this.invocation = invocation;
@@ -143,44 +139,31 @@ class ReactorSleuthMethodInvocationProcessor
span = this.span;
}
try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
this.source.subscribe(new SpanSubscriber(actual,
this.processor,
this.invocation,
this.span == null,
span,
this.log,
this.hasLog));
this.source.subscribe(new SpanSubscriber(actual, this.processor,
this.invocation, this.span == null, span, this.log, this.hasLog));
}
}
@Nullable
public Object scanUnsafe(Attr key) {
if (key == Attr.PARENT) {
return this.source;
}
else {
return null;
}
}
}
private static final class MonoSpan extends Mono<Object> implements Scannable {
private static final class MonoSpan extends MonoOperator<Object, Object> {
final Span span;
final MethodInvocation invocation;
final String log;
final boolean hasLog;
final Mono<Object> source;
final Span span;
final MethodInvocation invocation;
final String log;
final boolean hasLog;
final ReactorSleuthMethodInvocationProcessor processor;
final NewSpan newSpan;
MonoSpan(Mono<Object> source,
ReactorSleuthMethodInvocationProcessor processor,
NewSpan newSpan,
@Nullable Span span,
MethodInvocation invocation,
final NewSpan newSpan;
MonoSpan(Mono<Object> source, ReactorSleuthMethodInvocationProcessor processor,
NewSpan newSpan, @Nullable Span span, MethodInvocation invocation,
String log) {
this.source = source;
super(source);
this.processor = processor;
this.newSpan = newSpan;
this.span = span;
@@ -202,48 +185,37 @@ class ReactorSleuthMethodInvocationProcessor
span = this.span;
}
try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
this.source.subscribe(new SpanSubscriber(actual,
this.processor,
this.invocation,
this.span == null,
span,
this.log,
this.hasLog));
this.source.subscribe(new SpanSubscriber(actual, this.processor,
this.invocation, this.span == null, span, this.log, this.hasLog));
}
}
@Nullable
public Object scanUnsafe(Attr key) {
if (key == Attr.PARENT) {
return this.source;
}
else {
return null;
}
}
}
private static final class SpanSubscriber implements CoreSubscriber<Object>,
Subscription,
Scannable {
private static final class SpanSubscriber
implements CoreSubscriber<Object>, Subscription, Scannable {
final CoreSubscriber<? super Object> actual;
final boolean isNewSpan;
final Span span;
final String log;
final boolean hasLog;
final CurrentTraceContext currentTraceContext;
final CoreSubscriber<? super Object> actual;
final boolean isNewSpan;
final Span span;
final String log;
final boolean hasLog;
final CurrentTraceContext currentTraceContext;
final ReactorSleuthMethodInvocationProcessor processor;
final Context context;
final Context context;
Subscription parent;
SpanSubscriber(CoreSubscriber<? super Object> actual,
ReactorSleuthMethodInvocationProcessor processor,
MethodInvocation invocation,
boolean isNewSpan,
Span span,
String log,
MethodInvocation invocation, boolean isNewSpan, Span span, String log,
boolean hasLog) {
this.actual = actual;
this.isNewSpan = isNewSpan;
@@ -332,6 +304,7 @@ class ReactorSleuthMethodInvocationProcessor
}
return null;
}
}
private boolean isReactorReturnType(Class<?> returnType) {

39
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java
View File

@@ -64,14 +64,12 @@ public abstract class ReactorSleuth {
log.trace("Scope passing operator [" + beanFactory + "]");
}
//Adapt if lazy bean factory
BooleanSupplier isActive =
beanFactory instanceof ConfigurableApplicationContext ?
((ConfigurableApplicationContext) beanFactory)::isActive :
() -> true;
// Adapt if lazy bean factory
BooleanSupplier isActive = beanFactory instanceof ConfigurableApplicationContext
? ((ConfigurableApplicationContext) beanFactory)::isActive : () -> true;
return Operators.liftPublisher((p, sub) -> {
//if Flux/Mono #just, #empty, #error
// if Flux/Mono #just, #empty, #error
if (p instanceof Fuseable.ScalarCallable) {
return sub;
}
@@ -81,40 +79,43 @@ public abstract class ReactorSleuth {
if (log.isTraceEnabled()) {
log.trace("Spring Context [" + beanFactory
+ "] already refreshed. Creating a scope "
+ "passing span subscriber with Reactor Context "
+ "[" + sub.currentContext() + "] and name ["
+ scannable.name() + "]");
+ "passing span subscriber with Reactor Context " + "["
+ sub.currentContext() + "] and name [" + scannable.name()
+ "]");
}
return scopePassingSpanSubscription(beanFactory.getBean(Tracing.class), sub);
return scopePassingSpanSubscription(beanFactory.getBean(Tracing.class),
sub);
}
if (log.isTraceEnabled()) {
log.trace("Spring Context [" + beanFactory
+ "] is not yet refreshed, falling back to lazy span subscriber. Reactor Context is ["
+ sub.currentContext() + "] and name is ["
+ scannable.name() + "]");
+ sub.currentContext() + "] and name is [" + scannable.name()
+ "]");
}
return new LazySpanSubscriber<>(lazyScopePassingSpanSubscription(beanFactory, scannable, sub));
return new LazySpanSubscriber<>(
lazyScopePassingSpanSubscription(beanFactory, scannable, sub));
});
}
static <T> SpanSubscriptionProvider<T> lazyScopePassingSpanSubscription(
BeanFactory beanFactory, Scannable scannable, CoreSubscriber<? super T> sub) {
return new SpanSubscriptionProvider<>(beanFactory, sub, sub.currentContext(), scannable.name());
return new SpanSubscriptionProvider<>(beanFactory, sub, sub.currentContext(),
scannable.name());
}
static <T> CoreSubscriber<? super T> scopePassingSpanSubscription(
Tracing tracing, CoreSubscriber<? super T> sub) {
static <T> CoreSubscriber<? super T> scopePassingSpanSubscription(Tracing tracing,
CoreSubscriber<? super T> sub) {
Context context = sub.currentContext();
Span root = context.hasKey(Span.class) ? context.get(Span.class) : tracing.tracer().currentSpan();
Span root = context.hasKey(Span.class) ? context.get(Span.class)
: tracing.tracer().currentSpan();
if (root != null) {
return new ScopePassingSpanSubscriber<>(sub, context, tracing, root);
}
else {
return sub; //no need to trace
return sub; // no need to trace
}
}

4
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java
View File

@@ -123,8 +123,10 @@ final class ScopePassingSpanSubscriber<T> implements SpanSubscription<T>, Scanna
public Object scanUnsafe(Attr key) {
if (key == Attr.PARENT) {
return this.s;
} else {
}
else {
return key == Attr.ACTUAL ? this.subscriber : null;
}
}
}

6
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java
View File

@@ -64,8 +64,10 @@ final class SpanSubscriptionProvider<T> implements Supplier<SpanSubscription<T>>
}
SpanSubscription<T> newCoreSubscriber(Tracing tracing) {
Span root = context.hasKey(Span.class) ? context.get(Span.class) : tracing.tracer().currentSpan();
return new ScopePassingSpanSubscriber<>(this.subscriber, this.context, tracing, root);
Span root = context.hasKey(Span.class) ? context.get(Span.class)
: tracing.tracer().currentSpan();
return new ScopePassingSpanSubscriber<>(this.subscriber, this.context, tracing,
root);
}
private Tracing tracing() {

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

@@ -25,6 +25,13 @@ import brave.propagation.TraceContext;
import brave.propagation.TraceContextOrSamplingFlags;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Subscription;
import reactor.core.CoreSubscriber;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoOperator;
import reactor.util.annotation.Nullable;
import reactor.util.context.Context;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.Ordered;
import org.springframework.http.HttpHeaders;
@@ -36,8 +43,6 @@ import org.springframework.web.reactive.HandlerMapping;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebFilter;
import org.springframework.web.server.WebFilterChain;
import reactor.core.publisher.Mono;
import reactor.util.context.Context;
/**
* A {@link WebFilter} that creates / continues / closes and detaches spans for a reactive
@@ -139,134 +144,181 @@ public final class TraceWebFilter implements WebFilter, Ordered {
if (log.isDebugEnabled()) {
log.debug("Received a request to uri [" + uri + "]");
}
Span spanFromAttribute = getSpanFromAttribute(exchange);
final String CONTEXT_ERROR = "sleuth.webfilter.context.error";
return chain.filter(exchange)
.compose(f -> f.then(Mono.subscriberContext()).onErrorResume(
t -> Mono.subscriberContext().map(c -> c.put(CONTEXT_ERROR, t)))
.flatMap(c -> {
// reactivate span from context
Span span = spanFromContext(c);
Mono<Void> continuation;
Throwable t = null;
if (c.hasKey(CONTEXT_ERROR)) {
t = c.get(CONTEXT_ERROR);
continuation = Mono.error(t);
}
else {
continuation = Mono.empty();
}
String httpRoute = null;
Object attribute = exchange.getAttribute(
HandlerMapping.BEST_MATCHING_HANDLER_ATTRIBUTE);
if (attribute instanceof HandlerMethod) {
HandlerMethod handlerMethod = (HandlerMethod) attribute;
addClassMethodTag(handlerMethod, span);
addClassNameTag(handlerMethod, span);
Object pattern = exchange.getAttribute(
HandlerMapping.BEST_MATCHING_PATTERN_ATTRIBUTE);
httpRoute = pattern != null ? pattern.toString() : "";
}
addResponseTagsForSpanWithoutParent(exchange,
exchange.getResponse(), span);
DecoratedServerHttpResponse delegate = new DecoratedServerHttpResponse(
exchange.getResponse(),
exchange.getRequest().getMethodValue(), httpRoute);
handler().handleSend(delegate, t, span);
if (log.isDebugEnabled()) {
log.debug("Handled send of " + span);
}
return continuation;
}).subscriberContext(c -> {
Span span;
if (c.hasKey(Span.class)) {
Span parent = c.get(Span.class);
span = tracer().nextSpan(TraceContextOrSamplingFlags
.create(parent.context())).start();
if (log.isDebugEnabled()) {
log.debug("Found span in reactor context" + span);
}
}
else {
if (spanFromAttribute != null) {
span = spanFromAttribute;
if (log.isDebugEnabled()) {
log.debug("Found span in attribute " + span);
}
}
else {
span = handler().handleReceive(extractor(),
exchange.getRequest().getHeaders(),
exchange.getRequest());
if (log.isDebugEnabled()) {
log.debug("Handled receive of span " + span);
}
}
exchange.getAttributes().put(TRACE_REQUEST_ATTR, span);
}
return c.put(Span.class, span);
}));
return new MonoWebFilterTrace(chain.filter(exchange), exchange, this);
}
private Span spanFromContext(Context c) {
if (c.hasKey(Span.class)) {
Span span = c.get(Span.class);
if (log.isDebugEnabled()) {
log.debug("Found span in context " + span);
private static class MonoWebFilterTrace extends MonoOperator<Void, Void> {
final ServerWebExchange exchange;
final Tracer tracer;
final Span attrSpan;
final HttpServerHandler<ServerHttpRequest, ServerHttpResponse> handler;
final TraceContext.Extractor<HttpHeaders> extractor;
MonoWebFilterTrace(Mono<? extends Void> source, ServerWebExchange exchange,
TraceWebFilter parent) {
super(source);
this.tracer = parent.tracer();
this.extractor = parent.extractor();
this.handler = parent.handler();
this.exchange = exchange;
this.attrSpan = exchange.getAttribute(TRACE_REQUEST_ATTR);
}
@Override
public void subscribe(CoreSubscriber<? super Void> subscriber) {
Context context = subscriber.currentContext();
this.source.subscribe(new WebFilterTraceSubscriber(subscriber, context,
findOrCreateSpan(context), this));
}
static final class WebFilterTraceSubscriber implements CoreSubscriber<Void> {
final CoreSubscriber<? super Void> actual;
final Context context;
final Span span;
final ServerWebExchange exchange;
final HttpServerHandler<ServerHttpRequest, ServerHttpResponse> handler;
WebFilterTraceSubscriber(CoreSubscriber<? super Void> actual, Context context,
Span span, MonoWebFilterTrace parent) {
this.actual = actual;
this.span = span;
this.context = context.put(Span.class, span);
this.exchange = parent.exchange;
this.handler = parent.handler;
}
@Override
public void onSubscribe(Subscription subscription) {
this.actual.onSubscribe(subscription);
}
@Override
public void onNext(Void aVoid) {
// IGNORE
}
@Override
public void onError(Throwable t) {
terminateSpan(t);
this.actual.onError(t);
}
@Override
public void onComplete() {
terminateSpan(null);
this.actual.onComplete();
}
@Override
public Context currentContext() {
return this.context;
}
private void terminateSpan(@Nullable Throwable t) {
String httpRoute = null;
Object attribute = this.exchange
.getAttribute(HandlerMapping.BEST_MATCHING_HANDLER_ATTRIBUTE);
if (attribute instanceof HandlerMethod) {
HandlerMethod handlerMethod = (HandlerMethod) attribute;
addClassMethodTag(handlerMethod, this.span);
addClassNameTag(handlerMethod, this.span);
Object pattern = this.exchange
.getAttribute(HandlerMapping.BEST_MATCHING_PATTERN_ATTRIBUTE);
httpRoute = pattern != null ? pattern.toString() : "";
}
addResponseTagsForSpanWithoutParent(this.exchange,
this.exchange.getResponse(), this.span);
DecoratedServerHttpResponse delegate = new DecoratedServerHttpResponse(
this.exchange.getResponse(),
this.exchange.getRequest().getMethodValue(), httpRoute);
this.handler.handleSend(delegate, t, this.span);
if (log.isDebugEnabled()) {
log.debug("Handled send of " + this.span);
}
}
private void addClassMethodTag(Object handler, Span span) {
if (handler instanceof HandlerMethod) {
String methodName = ((HandlerMethod) handler).getMethod().getName();
span.tag(MVC_CONTROLLER_METHOD_KEY, methodName);
if (log.isDebugEnabled()) {
log.debug("Adding a method tag with value [" + methodName
+ "] to a span " + span);
}
}
}
private void addClassNameTag(Object handler, Span span) {
String className;
if (handler instanceof HandlerMethod) {
className = ((HandlerMethod) handler).getBeanType().getSimpleName();
}
else {
className = handler.getClass().getSimpleName();
}
if (log.isDebugEnabled()) {
log.debug("Adding a class tag with value [" + className
+ "] to a span " + span);
}
span.tag(MVC_CONTROLLER_CLASS_KEY, className);
}
private void addResponseTagsForSpanWithoutParent(ServerWebExchange exchange,
ServerHttpResponse response, Span span) {
if (spanWithoutParent(exchange) && response.getStatusCode() != null
&& span != null) {
span.tag(STATUS_CODE_KEY,
String.valueOf(response.getStatusCode().value()));
}
}
private boolean spanWithoutParent(ServerWebExchange exchange) {
return exchange.getAttribute(TRACE_SPAN_WITHOUT_PARENT) != null;
}
}
private Span findOrCreateSpan(Context c) {
Span span;
if (c.hasKey(Span.class)) {
Span parent = c.get(Span.class);
span = tracer
.nextSpan(TraceContextOrSamplingFlags.create(parent.context()))
.start();
if (log.isDebugEnabled()) {
log.debug("Found span in reactor context" + span);
}
}
else {
if (this.attrSpan != null) {
span = this.attrSpan;
if (log.isDebugEnabled()) {
log.debug("Found span in attribute " + span);
}
}
else {
span = this.handler.handleReceive(this.extractor,
this.exchange.getRequest().getHeaders(),
this.exchange.getRequest());
if (log.isDebugEnabled()) {
log.debug("Handled receive of span " + span);
}
}
this.exchange.getAttributes().put(TRACE_REQUEST_ATTR, span);
}
return span;
}
Span span = defaultSpan();
if (log.isDebugEnabled()) {
log.debug("No span found in context. Creating a new one " + span);
}
return span;
}
private Span defaultSpan() {
return tracer().nextSpan().start();
}
private void addResponseTagsForSpanWithoutParent(ServerWebExchange exchange,
ServerHttpResponse response, Span span) {
if (spanWithoutParent(exchange) && response.getStatusCode() != null
&& span != null) {
span.tag(STATUS_CODE_KEY, String.valueOf(response.getStatusCode().value()));
}
}
private Span getSpanFromAttribute(ServerWebExchange exchange) {
return exchange.getAttribute(TRACE_REQUEST_ATTR);
}
private boolean spanWithoutParent(ServerWebExchange exchange) {
return exchange.getAttribute(TRACE_SPAN_WITHOUT_PARENT) != null;
}
private void addClassMethodTag(Object handler, Span span) {
if (handler instanceof HandlerMethod) {
String methodName = ((HandlerMethod) handler).getMethod().getName();
span.tag(MVC_CONTROLLER_METHOD_KEY, methodName);
if (log.isDebugEnabled()) {
log.debug("Adding a method tag with value [" + methodName + "] to a span "
+ span);
}
}
}
private void addClassNameTag(Object handler, Span span) {
String className;
if (handler instanceof HandlerMethod) {
className = ((HandlerMethod) handler).getBeanType().getSimpleName();
}
else {
className = handler.getClass().getSimpleName();
}
if (log.isDebugEnabled()) {
log.debug("Adding a class tag with value [" + className + "] to a span "
+ span);
}
span.tag(MVC_CONTROLLER_CLASS_KEY, className);
}
@Override

262
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;
@@ -47,7 +56,7 @@ import org.springframework.web.reactive.function.client.WebClient;
* @author Marcin Grzejszczak
* @since 2.0.0
*/
class TraceWebClientBeanPostProcessor implements BeanPostProcessor {
final class TraceWebClientBeanPostProcessor implements BeanPostProcessor {
private final BeanFactory beanFactory;
@@ -87,7 +96,7 @@ class TraceWebClientBeanPostProcessor implements BeanPostProcessor {
}
class TraceExchangeFilterFunction implements ExchangeFilterFunction {
final class TraceExchangeFilterFunction implements ExchangeFilterFunction {
private static final Log log = LogFactory.getLog(TraceExchangeFilterFunction.class);
@@ -117,6 +126,8 @@ class TraceExchangeFilterFunction implements ExchangeFilterFunction {
final BeanFactory beanFactory;
final Function<? super Publisher<DataBuffer>, ? extends Publisher<DataBuffer>> scopePassingTransformer;
Tracer tracer;
HttpTracing httpTracing;
@@ -127,86 +138,185 @@ class TraceExchangeFilterFunction implements ExchangeFilterFunction {
TraceExchangeFilterFunction(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.scopePassingTransformer = ReactorSleuth
.scopePassingSpanOperator(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;
return new MonoWebClientTrace(next, request, this);
}
private void handleReceive(Span clientSpan, Tracer.SpanInScope ws,
ClientResponse clientResponse, Throwable throwable) {
handler().handleReceive(clientResponse, throwable, clientSpan);
ws.close();
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;
MonoWebClientTrace(ExchangeFunction next, ClientRequest request,
TraceExchangeFilterFunction parent) {
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;
}
@Override
public void subscribe(CoreSubscriber<? super ClientResponse> subscriber) {
final ClientRequest.Builder builder = ClientRequest.from(this.request);
Context context = subscriber.currentContext();
this.next.exchange(builder.build()).subscribe(new WebClientTracerSubscriber(
subscriber, context, findOrCreateSpan(builder), this));
}
private Span findOrCreateSpan(ClientRequest.Builder builder) {
if (log.isDebugEnabled()) {
log.debug("Instrumenting WebClient call");
}
Span clientSpan = this.handler.handleSend(this.injector, builder,
this.request, this.tracer.nextSpan());
if (log.isDebugEnabled()) {
log.debug("Handled send of " + clientSpan);
}
return clientSpan;
}
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(subscription);
}
@Override
public void onNext(ClientResponse response) {
done = true;
try {
// decorate response body
this.actual.onNext(ClientResponse.from(response)
.body(response.bodyToFlux(DataBuffer.class)
.transform(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 (!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 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 [" + span
+ "]");
}
handleReceive(span, 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 ["
+ span + "]");
}
throwable = new RestClientException("Status code of the response is ["
+ clientResponse.statusCode().value()
+ "] and the reason is ["
+ clientResponse.statusCode().getReasonPhrase() + "]");
}
handleReceive(span, ws, clientResponse, throwable);
}
}
}
@SuppressWarnings("unchecked")

9
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java
View File

@@ -46,9 +46,8 @@ public class ScopePassingSpanSubscriberTests {
@Test
public void should_set_empty_context_when_context_is_null() {
ScopePassingSpanSubscriber<?> subscriber = new ScopePassingSpanSubscriber<>(null
, null,
this.tracing, null);
ScopePassingSpanSubscriber<?> subscriber = new ScopePassingSpanSubscriber<>(null,
null, this.tracing, null);
then(subscriber.currentContext().isEmpty()).isTrue();
}
@@ -58,8 +57,8 @@ public class ScopePassingSpanSubscriberTests {
Span span = this.tracing.tracer().nextSpan();
try (Tracer.SpanInScope ws = this.tracing.tracer()
.withSpanInScope(span.start())) {
CoreSubscriber<?> subscriber =
ReactorSleuth.scopePassingSpanSubscription(tracing, new BaseSubscriber<Object>() {
CoreSubscriber<?> subscriber = ReactorSleuth
.scopePassingSpanSubscription(tracing, new BaseSubscriber<Object>() {
});
then(subscriber.currentContext().get(Span.class)).isEqualTo(span);

74
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java
View File

@@ -88,10 +88,10 @@ public class SpanSubscriberTests {
try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) {
Mono.just(1).flatMap(d -> Flux.just(d + 1).collectList().map(p -> p.get(0)))
.map(d -> d + 1).map((d) -> {
spanInOperation.set(this.tracer.currentSpan());
return d + 1;
}).map(d -> d + 1).subscribe(System.out::println);
.map(d -> d + 1).map((d) -> {
spanInOperation.set(this.tracer.currentSpan());
return d + 1;
}).map(d -> d + 1).subscribe(System.out::println);
}
finally {
span.finish();
@@ -106,13 +106,13 @@ public class SpanSubscriberTests {
Span span = this.tracer.nextSpan().name("foo").start();
log.info("Hello");
//Disable global hooks for local hook testing
// Disable global hooks for local hook testing
Hooks.resetOnLastOperator();
try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) {
Function<? super Publisher<Integer>, ? extends Publisher<Integer>> transformer =
ReactorSleuth.scopePassingSpanOperator(factory);
Function<? super Publisher<Integer>, ? extends Publisher<Integer>> transformer = ReactorSleuth
.scopePassingSpanOperator(factory);
Subscriber<Object> assertNoSpanSubscriber = new CoreSubscriber<Object>() {
@Override
@@ -159,26 +159,20 @@ public class SpanSubscriberTests {
}
};
transformer.apply(Mono.just(1).hide())
.subscribe(assertSpanSubscriber);
transformer.apply(Mono.just(1).hide()).subscribe(assertSpanSubscriber);
transformer.apply(Mono.just(1))
.subscribe(assertNoSpanSubscriber);
transformer.apply(Mono.just(1)).subscribe(assertNoSpanSubscriber);
transformer.apply(Mono.<Integer>error(new Exception()).hide())
.subscribe(assertSpanSubscriber);
.subscribe(assertSpanSubscriber);
transformer.apply(Mono.error(new Exception()))
.subscribe(assertNoSpanSubscriber);
.subscribe(assertNoSpanSubscriber);
transformer.apply(Mono.<Integer>empty().hide())
.subscribe(assertSpanSubscriber);
transformer.apply(Mono.empty())
.subscribe(assertNoSpanSubscriber);
.subscribe(assertSpanSubscriber);
transformer.apply(Mono.empty()).subscribe(assertNoSpanSubscriber);
}
finally {
@@ -198,12 +192,12 @@ public class SpanSubscriberTests {
try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) {
Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.1")
.map(d -> d + 1).map(d -> d + 1)
.publishOn(Schedulers.newSingle("secondThread")).log("reactor.2")
.map((d) -> {
spanInOperation.set(this.tracer.currentSpan());
return d + 1;
}).map(d -> d + 1).blockLast();
.map(d -> d + 1).map(d -> d + 1)
.publishOn(Schedulers.newSingle("secondThread")).log("reactor.2")
.map((d) -> {
spanInOperation.set(this.tracer.currentSpan());
return d + 1;
}).map(d -> d + 1).blockLast();
Awaitility.await().untilAsserted(() -> {
then(spanInOperation.get().context().traceId())
@@ -220,10 +214,10 @@ public class SpanSubscriberTests {
try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(foo2)) {
Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.")
.map(d -> d + 1).map(d -> d + 1).map((d) -> {
spanInOperation.set(this.tracer.currentSpan());
return d + 1;
}).map(d -> d + 1).blockLast();
.map(d -> d + 1).map(d -> d + 1).map((d) -> {
spanInOperation.set(this.tracer.currentSpan());
return d + 1;
}).map(d -> d + 1).blockLast();
then(this.tracer.currentSpan()).isEqualTo(foo2);
// parent cause there's an async span in the meantime
@@ -243,11 +237,11 @@ public class SpanSubscriberTests {
log.info("Hello");
try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(parentSpan)) {
final Long spanId = Mono.fromCallable(tracer::currentSpan)
.map(span -> span.context().spanId()).block();
.map(span -> span.context().spanId()).block();
then(spanId).isNotNull();
final Long secondSpanId = Mono.fromCallable(tracer::currentSpan)
.map(span -> span.context().spanId()).block();
.map(span -> span.context().spanId()).block();
then(secondSpanId).isEqualTo(spanId); // different trace ids here
}
}
@@ -260,11 +254,11 @@ public class SpanSubscriberTests {
try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) {
Mono.fromCallable(tracer::currentSpan).map(span -> span.context().spanId())
.doOnNext(spanInOperation::set)
.zipWith(Mono.fromCallable(tracer::currentSpan)
.map(span -> span.context().spanId())
.doOnNext(spanInZipOperation::set))
.block();
.doOnNext(spanInOperation::set)
.zipWith(Mono.fromCallable(tracer::currentSpan)
.map(span -> span.context().spanId())
.doOnNext(spanInZipOperation::set))
.block();
}
then(spanInZipOperation).hasValue(initSpan.context().spanId()); // ok here
@@ -283,8 +277,8 @@ public class SpanSubscriberTests {
try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) {
Mono.just("value1")
.flatMap(request -> Mono.just("value2").then(Mono.just("foo")))
.map(a -> "qwe").block();
.flatMap(request -> Mono.just("value2").then(Mono.just("foo")))
.map(a -> "qwe").block();
}
}
@@ -296,8 +290,8 @@ public class SpanSubscriberTests {
try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) {
Mono.subscriberContext()
.map(context -> tracer.currentSpan().context().spanId())
.doOnNext(spanInSubscriberContext::set).block();
.map(context -> tracer.currentSpan().context().spanId())
.doOnNext(spanInSubscriberContext::set).block();
}
then(spanInSubscriberContext).hasValue(initSpan.context().spanId()); // ok here

11
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/FlatMapTests.java
View File

@@ -116,14 +116,15 @@ public class FlatMapTests {
thenSpanInFooHasSameTraceId(secondTraceId, config);
LOGGER.info("Span in Foo has same trace id");
// and
List<String> requestUri = Arrays
.stream(capture.toString().split("\n"))
List<String> requestUri = Arrays.stream(capture.toString().split("\n"))
.filter(s -> s.contains("Received a request to uri"))
.map(s -> s.split(",")[1]).collect(Collectors.toList());
LOGGER.info("TracingFilter should not have any trace when receiving a request " + requestUri);
LOGGER.info(
"TracingFilter should not have any trace when receiving a request "
+ requestUri);
then(requestUri).as(
"TracingFilter should not have any trace when receiving a request")
.containsOnly("");
"TracingFilter should not have any trace when receiving a request")
.containsOnly("");
// and #866
then(factoryUser.wasSchedulerWrapped).isTrue();
LOGGER.info("Factory was wrapped");

3
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFluxTests.java
View File

@@ -552,7 +552,8 @@ class TestBean {
log.info("New Span in Subscriber Context");
return Mono.subscriberContext()
.doOnSuccess(context -> log.info("New Span in deferred Trace Context"))
.flatMap(context -> Mono.defer(() -> Mono.just(tracer.currentSpan().context().spanId())));
.flatMap(context -> Mono
.defer(() -> Mono.just(tracer.currentSpan().context().spanId())));
}
}

8
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebAsyncClientAutoConfigurationTests.java
View File

@@ -90,10 +90,10 @@ public class TraceWebAsyncClientAutoConfigurationTests {
}
Awaitility.await().untilAsserted(() -> {
then(this.accumulator
.getSpans().stream().filter(span -> Span.Kind.CLIENT == span.kind())
.findFirst().get()).matches(
span -> span.duration() >= TimeUnit.MILLISECONDS.toMicros(100));
then(this.accumulator.getSpans().stream()
.filter(span -> Span.Kind.CLIENT == span.kind()).findFirst().get())
.matches(span -> span.duration() >= TimeUnit.MILLISECONDS
.toMicros(100));
then(this.tracer.tracer().currentSpan()).isNull();
});
}

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

@@ -296,7 +296,8 @@ public class WebClientTests {
Awaitility.await().untilAsserted(() -> {
then(this.tracer.currentSpan()).isNull();
System.out.println("Collected span " + this.reporter.getSpans());
then(this.reporter.getSpans()).isNotEmpty().extracting("traceId", String.class)
then(this.reporter.getSpans()).isNotEmpty()
.extracting("traceId", String.class)
.containsOnly(span.context().traceIdString());
then(this.reporter.getSpans()).extracting("kind.name").contains("CLIENT");
});