From b94efb90a15dbd071b3a0a4ce3396c1325296553 Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Thu, 20 Jul 2017 10:38:09 +0200 Subject: [PATCH] fix #646 : Filter fused flows from tracing --- .../TraceReactorAutoConfiguration.java | 13 ++- .../reactor/SpanSubscriberTests.java | 79 ++++++++++++++++--- 2 files changed, 80 insertions(+), 12 deletions(-) diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java index 5890d2bd3..123f769ac 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java @@ -16,6 +16,8 @@ import org.springframework.cloud.sleuth.Tracer; import org.springframework.cloud.sleuth.instrument.async.TraceableScheduledExecutorService; import org.springframework.cloud.sleuth.instrument.web.TraceWebFluxAutoConfiguration; import org.springframework.context.annotation.Configuration; + +import reactor.core.Fuseable; import reactor.core.publisher.Hooks; import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; @@ -44,8 +46,15 @@ public class TraceReactorAutoConfiguration { @PostConstruct public void setupHooks() { - Hooks.onNewSubscriber((pub, sub) -> - new SpanSubscriber(sub, sub.currentContext(), this.tracer, pub.toString())); + Hooks.onNewSubscriber((pub, sub) -> { + //do not trace fused flows or simple just/error/empty + if(pub instanceof Fuseable && sub instanceof Fuseable.QueueSubscription + || pub instanceof Fuseable.ScalarCallable){ + return sub; + } + return new SpanSubscriber(sub, sub.currentContext(), this.tracer, pub + .toString()); + }); Schedulers.setFactory(new Schedulers.Factory() { @Override public ScheduledExecutorService decorateScheduledExecutorService( String schedulerType, diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java index eac363666..c1d742e35 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java @@ -5,11 +5,11 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.awaitility.Awaitility; -import org.junit.AfterClass; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.reactivestreams.Publisher; +import org.reactivestreams.Subscription; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; @@ -21,8 +21,9 @@ import org.springframework.cloud.sleuth.util.ExceptionUtils; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.test.context.junit4.SpringRunner; +import reactor.core.publisher.BaseSubscriber; import reactor.core.publisher.Flux; -import reactor.core.publisher.Hooks; +import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; import static org.assertj.core.api.BDDAssertions.then; @@ -58,7 +59,71 @@ public class SpanSubscriberTests { .subscribe(System.out::println); then(this.tracer.getCurrentSpan()).isNull(); - then(spanInOperation.get().getTraceId()).isEqualTo(span.getTraceId()); + then(spanInOperation.get().getParents().get(0)).isEqualTo(span.getSpanId()); + then(ExceptionUtils.getLastException()).isNull(); + } + + @Test public void should_support_reactor_fusion_optimization() { + Span span = this.tracer.createSpan("foo"); + final AtomicReference spanInOperation = new AtomicReference<>(); + log.info("Hello"); + + Mono.just(1) + .flatMap( d -> Flux.just(d + 1).collectList().map(p -> p.get(0))) + .map( d -> d + 1) + .map( (d) -> { + spanInOperation.set(SpanSubscriberTests.this.tracer.getCurrentSpan()); + return d + 1; + }) + .map( d -> d + 1) + .subscribe(System.out::println); + + then(this.tracer.getCurrentSpan()).isNull(); + then(spanInOperation.get().getParents().get(0)).isEqualTo(span.getSpanId()); + then(ExceptionUtils.getLastException()).isNull(); + } + + @Test public void should_not_trace_scalar_flows() { + Span span = this.tracer.createSpan("foo"); + final AtomicReference spanInOperation = new AtomicReference<>(); + log.info("Hello"); + + Mono.just(1) + .subscribe(new BaseSubscriber() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + spanInOperation.set(subscription); + } + }); + + then(this.tracer.getCurrentSpan()).isNotNull(); + then(spanInOperation.get()).isNotInstanceOf(SpanSubscriber.class); + + Mono.error(new Exception()) + .subscribe(new BaseSubscriber() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + spanInOperation.set(subscription); + } + + @Override + protected void hookOnError(Throwable throwable) { + } + }); + + then(this.tracer.getCurrentSpan()).isNotNull(); + then(spanInOperation.get()).isNotInstanceOf(SpanSubscriber.class); + + Mono.empty() + .subscribe(new BaseSubscriber() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + spanInOperation.set(subscription); + } + }); + + then(this.tracer.getCurrentSpan()).isNotNull(); + then(spanInOperation.get()).isNotInstanceOf(SpanSubscriber.class); then(ExceptionUtils.getLastException()).isNull(); } @@ -107,16 +172,10 @@ public class SpanSubscriberTests { then(this.tracer.getCurrentSpan()).isEqualTo(foo2); then(ExceptionUtils.getLastException()).isNull(); // parent cause there's an async span in the meantime - then(spanInOperation.get().getTraceId()).isEqualTo(foo2.getTraceId()); + then(spanInOperation.get().getSavedSpan().getParents().get(0)).isEqualTo(foo2.getSpanId()); tracer.close(foo2); } - @AfterClass - public static void cleanup() { - Hooks.resetOnNewSubscriber(); - Schedulers.resetFactory(); - } - @EnableAutoConfiguration @Configuration static class Config {