From 7b11141cd7cce706c831018714d90502d192bb4c Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Fri, 21 Jul 2017 11:31:43 +0200 Subject: [PATCH] fix #646 : Filter fused flows from tracing --- .../instrument/reactor/SpanSubscriber.java | 37 +++++----- .../TraceReactorAutoConfiguration.java | 14 +++- .../reactor/SpanSubscriberTests.java | 67 +++++++++++++++++++ 3 files changed, 93 insertions(+), 25 deletions(-) diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java index f49df0725..eaa976613 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java @@ -18,8 +18,7 @@ import reactor.util.context.Context; * @author Marcin Grzejszczak * @since 1.3.0 */ -class SpanSubscriber extends AtomicBoolean - implements Subscription, CoreSubscriber { +class SpanSubscriber extends AtomicBoolean implements Subscription, CoreSubscriber { private static final Logger log = Loggers.getLogger(SpanSubscriber.class); @@ -62,30 +61,29 @@ class SpanSubscriber extends AtomicBoolean } @Override public void request(long n) { + if (log.isTraceEnabled()) { + log.trace("Request"); + } this.tracer.continueSpan(this.span); if (log.isTraceEnabled()) { log.trace("Request - continued"); } this.s.request(n); - Span localSpan = this.span; - Span rootSpan = this.rootSpan; // We're in the main thread so we don't want to pollute it with wrong spans // that's why we need to detach the current one and continue with its parent - if (log.isTraceEnabled()) { - log.trace("Will detach spans. Root span is " + rootSpan + " and stored span is " + localSpan); - } - while (localSpan != null) { - if (rootSpan != null) { - if (localSpan.getSpanId() != rootSpan.getSpanId() && - !isRootParentSpan(localSpan)) { - localSpan = continueDetachedSpan(localSpan); + Span localRootSpan = this.span; + while (localRootSpan != null) { + if (this.rootSpan != null) { + if (localRootSpan.getSpanId() != this.rootSpan.getSpanId() && + !isRootParentSpan(localRootSpan)) { + localRootSpan = continueDetachedSpan(localRootSpan); } else { - localSpan = null; + localRootSpan = null; } - } else if (!isRootParentSpan(localSpan)) { - localSpan = continueDetachedSpan(localSpan); + } else if (!isRootParentSpan(localRootSpan)) { + localRootSpan = continueDetachedSpan(localRootSpan); } else { - localSpan = null; + localRootSpan = null; } } if (log.isTraceEnabled()) { @@ -103,12 +101,7 @@ class SpanSubscriber extends AtomicBoolean log.trace("Will detach span {}", localRootSpan); } Span detachedSpan = this.tracer.detach(localRootSpan); - Span continuedSpan = this.tracer.continueSpan(detachedSpan); - if (log.isTraceEnabled()) { - log.trace("Now current span is " + continuedSpan + ". Root span is " + this.rootSpan + - " and stored span for the subscriber is " + this.span); - } - return continuedSpan; + return this.tracer.continueSpan(detachedSpan); } @Override public void cancel() { 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..302b524bb 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,7 @@ 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 +45,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, @@ -58,4 +66,4 @@ public class TraceReactorAutoConfiguration { }); } } -} +} \ No newline at end of file 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..a27fbdb68 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 @@ -10,6 +10,7 @@ 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 +22,10 @@ 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; @@ -62,6 +65,70 @@ public class SpanSubscriberTests { 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().getTraceId()).isEqualTo(span.getTraceId()); + 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(); + } + @Test public void should_pass_tracing_info_when_using_reactor_async() {