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

fix #646 : Filter fused flows from tracing

Browse Source
This commit is contained in:
Marcin Grzejszczak
2017-07-21 11:31:43 +02:00
parent 2ebba99c4e
commit 7b11141cd7
3 changed files with 93 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,8 +18,7 @@ import reactor.util.context.Context;
* @author Marcin Grzejszczak
* @since 1.3.0
*/
class SpanSubscriber extends AtomicBoolean
implements Subscription, CoreSubscriber<Object> {
class SpanSubscriber extends AtomicBoolean implements Subscription, CoreSubscriber<Object> {
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() {

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

@@ -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 {
});
}
}
}
}

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

@@ -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<Span> 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<Subscription> spanInOperation = new AtomicReference<>();
log.info("Hello");
Mono.just(1)
.subscribe(new BaseSubscriber<Integer>() {
@Override
protected void hookOnSubscribe(Subscription subscription) {
spanInOperation.set(subscription);
}
});
then(this.tracer.getCurrentSpan()).isNotNull();
then(spanInOperation.get()).isNotInstanceOf(SpanSubscriber.class);
Mono.<Integer>error(new Exception())
.subscribe(new BaseSubscriber<Integer>() {
@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.<Integer>empty()
.subscribe(new BaseSubscriber<Integer>() {
@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() {