From fdeb675c0e911ff8a67f789bb7333dc9e3974bfd Mon Sep 17 00:00:00 2001 From: Sergii Karpenko Date: Tue, 14 Aug 2018 19:02:18 +0300 Subject: [PATCH] Set span in context (#1056) * tests on trace context * improve performance : excessive check removed (#1053) * Fixed checkstyle * set span in trace context * explicit imports * deprecated fixes #1030 --- .../annotation/SleuthAdvisorConfig.java | 8 +- .../instrument/reactor/ReactorSleuth.java | 8 + .../instrument/reactor/SpanSubscriber.java | 2 + .../reactor/SpanSubscriptionProvider.java | 2 +- .../TraceReactorAutoConfiguration.java | 39 +-- .../SleuthSpanCreatorAspectFluxTests.java | 74 +++++- .../SleuthSpanCreatorAspectMonoTests.java | 158 +++++++++++- .../SleuthSpanCreatorAspectWebFluxTests.java | 238 ++++++++++++++++++ .../reactor/SpanSubscriberTests.java | 65 +++-- 9 files changed, 529 insertions(+), 65 deletions(-) create mode 100644 spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectWebFluxTests.java diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/SleuthAdvisorConfig.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/SleuthAdvisorConfig.java index f4c2fb0e2..dbc2cbde8 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/SleuthAdvisorConfig.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/SleuthAdvisorConfig.java @@ -259,12 +259,16 @@ class SleuthInterceptor implements IntroductionInterceptor, BeanFactoryAware { if(publisher instanceof Mono){ return startSpan.flatMap(spanStarted -> ((Mono)publisher) .doOnError(onFailureReactor(log, hasLog, spanStarted)) - .doOnTerminate(afterReactor(startNewSpan, log, hasLog, spanStarted))); + .doOnTerminate(afterReactor(startNewSpan, log, hasLog, spanStarted))) + //put span in context so it can be used by ScopePassingSpanSubscriber + .subscriberContext(context -> context.put(Span.class, span)); } else if(publisher instanceof Flux){ return startSpan.flatMapMany(spanStarted -> ((Flux)publisher) .doOnError(onFailureReactor(log, hasLog, spanStarted)) - .doOnTerminate(afterReactor(startNewSpan, log, hasLog, spanStarted))); + .doOnTerminate(afterReactor(startNewSpan, log, hasLog, spanStarted))) + //put span in context so it can be used by ScopePassingSpanSubscriber + .subscriberContext(context -> context.put(Span.class, span)); } else { throw new IllegalArgumentException("Unexpected type of publisher: "+publisher.getClass()); diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java index be63caba3..46dd82dfd 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java @@ -48,14 +48,22 @@ public abstract class ReactorSleuth { * reactor.core.publisher.Hooks#onEachOperator(Function)} or {@link * reactor.core.publisher.Hooks#onLastOperator(Function)}. * + * @deprecated use {@link ReactorSleuth#scopePassingSpanOperator} instead * @param beanFactory * @param an arbitrary type that is left unchanged by the span operator * * @return a new lazy span operator pointcut */ @SuppressWarnings("unchecked") + @Deprecated public static Function, ? extends Publisher> spanOperator( BeanFactory beanFactory) { + + if(log.isWarnEnabled()){ + log.warn("spanOperator method will be deleted in the next major release. " + + "Use scopePassingSpanOperator() method instead"); + } + return sourcePub -> { // TODO: Remove this once Reactor 3.1.8 is released //do the checks directly on actual original Publisher 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 7e86873a4..cd3a5a445 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 @@ -31,10 +31,12 @@ import java.util.concurrent.atomic.AtomicBoolean; /** * A trace representation of the {@link Subscriber} * + * @deprecated use {@link ScopePassingSpanSubscriber} instead * @author Stephane Maldini * @author Marcin Grzejszczak * @since 2.0.0 */ +@Deprecated final class SpanSubscriber extends AtomicBoolean implements SpanSubscription { private static final Logger log = Loggers.getLogger( diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java index a10319246..3fa591754 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java @@ -38,7 +38,7 @@ class SpanSubscriptionProvider implements Supplier> { final Subscriber subscriber; final Context context; final String name; - private Tracing tracing; + private volatile Tracing tracing; SpanSubscriptionProvider(BeanFactory beanFactory, Subscriber subscriber, 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 da1e8bece..4a1d7e032 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,15 +16,7 @@ package org.springframework.cloud.sleuth.instrument.reactor; -import javax.annotation.PreDestroy; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.function.Supplier; - import brave.Tracing; -import reactor.core.publisher.Hooks; -import reactor.core.publisher.Mono; -import reactor.core.scheduler.Schedulers; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; @@ -34,15 +26,21 @@ import org.springframework.boot.autoconfigure.AutoConfigureAfter; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnNotWebApplication; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication; import org.springframework.cloud.sleuth.instrument.async.TraceableScheduledExecutorService; import org.springframework.cloud.sleuth.instrument.web.TraceWebFluxAutoConfiguration; import org.springframework.context.ApplicationListener; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.event.ContextRefreshedEvent; +import reactor.core.publisher.Hooks; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +import javax.annotation.PreDestroy; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Supplier; /** * {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration Auto-configuration} @@ -64,19 +62,8 @@ public class TraceReactorAutoConfiguration { static final String SLEUTH_TRACE_REACTOR_KEY = TraceReactorConfiguration.class.getName(); - @Bean - @ConditionalOnNotWebApplication LastOperatorWrapper spanOperator() { - return beanFactory -> Hooks.onLastOperator(SLEUTH_TRACE_REACTOR_KEY, ReactorSleuth.spanOperator(beanFactory)); - } - - @Bean - @ConditionalOnWebApplication LastOperatorWrapper noOpLastOperatorWrapper() { - return beanFactory -> { }; - } - @PreDestroy public void cleanupHooks() { - Hooks.resetOnLastOperator(SLEUTH_TRACE_REACTOR_KEY); Hooks.resetOnEachOperator(SLEUTH_TRACE_REACTOR_KEY); Schedulers.resetFactory(); } @@ -94,10 +81,6 @@ public class TraceReactorAutoConfiguration { } } -interface LastOperatorWrapper { - void wrapLastOperator(BeanFactory beanFactory); -} - class HookRegisteringBeanDefinitionRegistryPostProcessor implements BeanDefinitionRegistryPostProcessor { @@ -107,12 +90,10 @@ class HookRegisteringBeanDefinitionRegistryPostProcessor implements @Override public void postProcessBeanFactory( ConfigurableListableBeanFactory beanFactory) throws BeansException { - LastOperatorWrapper wrapper = beanFactory.getBean(LastOperatorWrapper.class); - setupHooks(wrapper, beanFactory); + setupHooks(beanFactory); } - void setupHooks(LastOperatorWrapper wrapper, BeanFactory beanFactory) { - wrapper.wrapLastOperator(beanFactory); + void setupHooks(BeanFactory beanFactory) { Hooks.onEachOperator( TraceReactorAutoConfiguration.TraceReactorConfiguration.SLEUTH_TRACE_REACTOR_KEY, ReactorSleuth.scopePassingSpanOperator(beanFactory)); diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectFluxTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectFluxTests.java index 7231e91a0..9a45dc238 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectFluxTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectFluxTests.java @@ -18,9 +18,11 @@ package org.springframework.cloud.sleuth.annotation; import brave.Span; import brave.Tracer; +import brave.Tracing; +import brave.propagation.TraceContext; import brave.sampler.Sampler; +import org.apache.commons.lang3.StringUtils; import org.junit.Before; -import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; @@ -70,6 +72,7 @@ public class SleuthSpanCreatorAspectFluxTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("test-method"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -82,6 +85,7 @@ public class SleuthSpanCreatorAspectFluxTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("test-method2"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -94,6 +98,7 @@ public class SleuthSpanCreatorAspectFluxTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("custom-name-on-test-method3"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -106,6 +111,7 @@ public class SleuthSpanCreatorAspectFluxTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("custom-name-on-test-method4"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -121,6 +127,7 @@ public class SleuthSpanCreatorAspectFluxTests { then(spans.get(0).name()).isEqualTo("custom-name-on-test-method5"); then(spans.get(0).tags()).containsEntry("testTag", "test"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -134,6 +141,7 @@ public class SleuthSpanCreatorAspectFluxTests { then(spans.get(0).name()).isEqualTo("custom-name-on-test-method6"); then(spans.get(0).tags()).containsEntry("testTag6", "test"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -146,6 +154,7 @@ public class SleuthSpanCreatorAspectFluxTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("custom-name-on-test-method8"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -161,6 +170,7 @@ public class SleuthSpanCreatorAspectFluxTests { .containsEntry("class", "TestBean") .containsEntry("method", "testMethod9"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -184,6 +194,7 @@ public class SleuthSpanCreatorAspectFluxTests { .stream().map(Annotation::value).collect(Collectors.toList())) .contains("customTest.before", "customTest.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -200,6 +211,7 @@ public class SleuthSpanCreatorAspectFluxTests { .stream().map(Annotation::value).collect(Collectors.toList())) .contains("customTest.before", "customTest.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -223,6 +235,7 @@ public class SleuthSpanCreatorAspectFluxTests { .stream().map(Annotation::value).collect(Collectors.toList())) .contains("customTest.before", "customTest.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -249,6 +262,7 @@ public class SleuthSpanCreatorAspectFluxTests { .stream().map(Annotation::value).collect(Collectors.toList())) .contains("customTest.before", "customTest.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -269,6 +283,7 @@ public class SleuthSpanCreatorAspectFluxTests { .containsEntry("testTag12", "test") .containsEntry("error", "test exception 12"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -298,6 +313,7 @@ public class SleuthSpanCreatorAspectFluxTests { .contains("testMethod13.before", "testMethod13.afterFailure", "testMethod13.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -307,6 +323,35 @@ public class SleuthSpanCreatorAspectFluxTests { List spans = new ArrayList<>(this.reporter.getSpans()); then(spans).isEmpty(); + then(this.tracer.currentSpan()).isNull(); + } + + @Test + public void shouldReturnNewSpanFromTraceContext() { + Flux flux = this.testBean.newSpanInTraceContext(); + Long newSpanId = flux.blockFirst(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("span-in-trace-context"); + then(spans.get(0).id()).isEqualTo(toHexString(newSpanId)); + then(this.tracer.currentSpan()).isNull(); + } + + @Test + public void shouldReturnNewSpanFromSubscriberContext() { + Flux flux = this.testBean.newSpanInSubscriberContext(); + Long newSpanId = flux.blockFirst(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("span-in-subscriber-context"); + then(spans.get(0).id()).isEqualTo(toHexString(newSpanId)); + then(this.tracer.currentSpan()).isNull(); + } + + private static String toHexString(long value){ + return StringUtils.leftPad(Long.toHexString(value), 16, '0'); } private void verifyNoSpansUntilFluxComplete(Flux flux) { @@ -377,6 +422,12 @@ public class SleuthSpanCreatorAspectFluxTests { @ContinueSpan Flux testMethod14(String param); + @NewSpan(name = "spanInTraceContext") + Flux newSpanInTraceContext(); + + @NewSpan(name = "spanInSubscriberContext") + Flux newSpanInSubscriberContext(); + void proceed(); void reset(); @@ -387,12 +438,18 @@ public class SleuthSpanCreatorAspectFluxTests { public static final String TEST_STRING1 = "Test String 1"; public static final String TEST_STRING2 = "Test String 2"; + private final Tracer tracer; + private AtomicReference> proceed = new AtomicReference<>(new CompletableFuture<>()); private Flux testFlux = Flux.defer(() -> Flux.just(TEST_STRING1, TEST_STRING2)) .delayUntil(s -> Mono.fromFuture(proceed.get())) .doOnNext(s -> proceed.set(new CompletableFuture<>())); + public TestBean(Tracer tracer) { + this.tracer = tracer; + } + @Override public void reset(){ proceed.set(new CompletableFuture<>()); @@ -483,6 +540,17 @@ public class SleuthSpanCreatorAspectFluxTests { public Flux testMethod14(String param) { return Flux.just(TEST_STRING1, TEST_STRING2); } + + @Override + public Flux newSpanInTraceContext() { + return Flux.defer(() -> Flux.just(tracer.currentSpan().context().spanId())); + } + + @Override + public Flux newSpanInSubscriberContext() { + return Mono.subscriberContext() + .flatMapMany(context -> Flux.just(tracer.currentSpan().context().spanId())); + } } @Configuration @@ -490,8 +558,8 @@ public class SleuthSpanCreatorAspectFluxTests { protected static class TestConfiguration { @Bean - public TestBeanInterface testBean() { - return new TestBean(); + public TestBeanInterface testBean(Tracer tracer) { + return new TestBean(tracer); } @Bean Reporter spanReporter() { diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectMonoTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectMonoTests.java index 530a8554b..bc90b9f63 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectMonoTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectMonoTests.java @@ -18,9 +18,11 @@ package org.springframework.cloud.sleuth.annotation; import brave.Span; import brave.Tracer; +import brave.Tracing; +import brave.propagation.TraceContext; import brave.sampler.Sampler; +import org.apache.commons.lang3.StringUtils; import org.junit.Before; -import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; @@ -29,6 +31,7 @@ import org.springframework.boot.test.context.SpringBootTest; import org.springframework.cloud.sleuth.util.ArrayListSpanReporter; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.data.util.Pair; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import reactor.core.publisher.Mono; import zipkin2.Annotation; @@ -47,6 +50,7 @@ import static reactor.core.publisher.Mono.just; public class SleuthSpanCreatorAspectMonoTests { @Autowired TestBeanInterface testBean; + @Autowired TestBeanOuter testBeanOuter; @Autowired Tracer tracer; @Autowired ArrayListSpanReporter reporter; @@ -82,6 +86,7 @@ public class SleuthSpanCreatorAspectMonoTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("test-method2"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -97,6 +102,7 @@ public class SleuthSpanCreatorAspectMonoTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("custom-name-on-test-method3"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -111,6 +117,7 @@ public class SleuthSpanCreatorAspectMonoTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("custom-name-on-test-method4"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -128,6 +135,7 @@ public class SleuthSpanCreatorAspectMonoTests { then(spans.get(0).name()).isEqualTo("custom-name-on-test-method5"); then(spans.get(0).tags()).containsEntry("testTag", "test"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -143,6 +151,7 @@ public class SleuthSpanCreatorAspectMonoTests { then(spans.get(0).name()).isEqualTo("custom-name-on-test-method6"); then(spans.get(0).tags()).containsEntry("testTag6", "test"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -157,6 +166,7 @@ public class SleuthSpanCreatorAspectMonoTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("custom-name-on-test-method8"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -174,6 +184,7 @@ public class SleuthSpanCreatorAspectMonoTests { .containsEntry("class", "TestBean") .containsEntry("method", "testMethod9"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -199,6 +210,7 @@ public class SleuthSpanCreatorAspectMonoTests { .stream().map(Annotation::value).collect(Collectors.toList())) .contains("customTest.before", "customTest.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -214,6 +226,7 @@ public class SleuthSpanCreatorAspectMonoTests { .stream().map(Annotation::value).collect(Collectors.toList())) .contains("customTest.before", "customTest.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -239,6 +252,7 @@ public class SleuthSpanCreatorAspectMonoTests { .stream().map(Annotation::value).collect(Collectors.toList())) .contains("customTest.before", "customTest.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -267,6 +281,7 @@ public class SleuthSpanCreatorAspectMonoTests { .stream().map(Annotation::value).collect(Collectors.toList())) .contains("customTest.before", "customTest.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -287,6 +302,7 @@ public class SleuthSpanCreatorAspectMonoTests { .containsEntry("testTag12", "test") .containsEntry("error", "test exception 12"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -316,6 +332,7 @@ public class SleuthSpanCreatorAspectMonoTests { .contains("testMethod13.before", "testMethod13.afterFailure", "testMethod13.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -325,8 +342,87 @@ public class SleuthSpanCreatorAspectMonoTests { List spans = new ArrayList<>(this.reporter.getSpans()); then(spans).isEmpty(); + then(this.tracer.currentSpan()).isNull(); } - + + @Test + public void shouldReturnNewSpanFromTraceContext() { + Mono mono = this.testBean.newSpanInTraceContext(); + + then(this.reporter.getSpans()).isEmpty(); + + Long newSpanId = mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("span-in-trace-context"); + then(spans.get(0).id()).isEqualTo(toHexString(newSpanId)); + then(this.tracer.currentSpan()).isNull(); + } + + @Test + public void shouldReturnNewSpanFromTraceContextOuter() { + Mono, Long>> mono = this.testBeanOuter.outerNewSpanInTraceContext(); + + then(this.reporter.getSpans()).isEmpty(); + + Pair, Long> pair = mono.block(); + Long outerSpanIdBefore = pair.getFirst().getFirst(); + Long outerSpanIdAfter = pair.getFirst().getSecond(); + Long innerSpanId = pair.getSecond(); + + then(outerSpanIdBefore).isEqualTo(outerSpanIdAfter).isNotEqualTo(innerSpanId); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(2); + then(spans.get(0).name()).isEqualTo("outer-span-in-trace-context"); + then(spans.get(0).id()).isEqualTo(toHexString(outerSpanIdBefore)); + then(spans.get(1).name()).isEqualTo("span-in-trace-context"); + then(spans.get(1).id()).isEqualTo(toHexString(innerSpanId)); + then(this.tracer.currentSpan()).isNull(); + } + + @Test + public void shouldReturnNewSpanFromSubscriberContext() { + Mono mono = this.testBean.newSpanInSubscriberContext(); + + then(this.reporter.getSpans()).isEmpty(); + + Long newSpanId = mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("span-in-subscriber-context"); + then(spans.get(0).id()).isEqualTo(toHexString(newSpanId)); + then(this.tracer.currentSpan()).isNull(); + } + + @Test + public void shouldReturnNewSpanFromSubscriberContextOuter() { + Mono, Long>> mono = this.testBeanOuter.outerNewSpanInSubscriberContext(); + + then(this.reporter.getSpans()).isEmpty(); + + Pair, Long> pair = mono.block(); + Long outerSpanIdBefore = pair.getFirst().getFirst(); + Long outerSpanIdAfter = pair.getFirst().getSecond(); + Long innerSpanId = pair.getSecond(); + + then(outerSpanIdBefore).isEqualTo(outerSpanIdAfter).isNotEqualTo(innerSpanId); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(2); + then(spans.get(0).name()).isEqualTo("outer-span-in-subscriber-context"); + then(spans.get(0).id()).isEqualTo(toHexString(outerSpanIdBefore)); + then(spans.get(1).name()).isEqualTo("span-in-subscriber-context"); + then(spans.get(1).id()).isEqualTo(toHexString(innerSpanId)); + then(this.tracer.currentSpan()).isNull(); + } + + private static String toHexString(long value){ + return StringUtils.leftPad(Long.toHexString(value), 16, '0'); + } + protected interface TestBeanInterface { // tag::annotated_method[] @@ -375,6 +471,12 @@ public class SleuthSpanCreatorAspectMonoTests { @ContinueSpan(log = "testMethod13") Mono testMethod13(); + + @NewSpan(name = "spanInTraceContext") + Mono newSpanInTraceContext(); + + @NewSpan(name = "spanInSubscriberContext") + Mono newSpanInSubscriberContext(); } protected static class TestBean implements TestBeanInterface { @@ -382,6 +484,12 @@ public class SleuthSpanCreatorAspectMonoTests { public static final String TEST_STRING = "Test String"; public static final Mono TEST_MONO = Mono.defer(() -> just(TEST_STRING)); + private final Tracer tracer; + + public TestBean(Tracer tracer) { + this.tracer = tracer; + } + @Override public Mono testMethod() { return TEST_MONO; @@ -458,6 +566,43 @@ public class SleuthSpanCreatorAspectMonoTests { public Mono testMethod13() { return Mono.defer(() -> Mono.error(new RuntimeException("test exception 13"))); } + + @Override + public Mono newSpanInTraceContext() { + return Mono.defer(() -> Mono.just(tracer.currentSpan().context().spanId())); + } + + @Override + public Mono newSpanInSubscriberContext() { + return Mono.subscriberContext() + .flatMap(context -> Mono.just(tracer.currentSpan().context().spanId())); + } + } + + protected static class TestBeanOuter { + + private final Tracer tracer; + private final TestBeanInterface testBeanInterface; + + public TestBeanOuter(Tracer tracer, TestBeanInterface testBeanInterface) { + this.tracer = tracer; + this.testBeanInterface = testBeanInterface; + } + + @NewSpan(name = "outerSpanInTraceContext") + public Mono, Long>> outerNewSpanInTraceContext() { + return Mono.defer(() -> Mono.just(tracer.currentSpan().context().spanId()) + .zipWith(testBeanInterface.newSpanInTraceContext()) + .map(pair -> Pair.of(Pair.of(pair.getT1(), tracer.currentSpan().context().spanId()), pair.getT2()))); + } + + @NewSpan(name = "outerSpanInSubscriberContext") + public Mono, Long>> outerNewSpanInSubscriberContext() { + return Mono.subscriberContext() + .flatMap(context -> Mono.just(tracer.currentSpan().context().spanId()) + .zipWith(testBeanInterface.newSpanInSubscriberContext()) + .map(pair -> Pair.of(Pair.of(pair.getT1(), tracer.currentSpan().context().spanId()), pair.getT2()))); + } } @Configuration @@ -465,8 +610,13 @@ public class SleuthSpanCreatorAspectMonoTests { protected static class TestConfiguration { @Bean - public TestBeanInterface testBean() { - return new TestBean(); + public TestBeanInterface testBean(Tracer tracer) { + return new TestBean(tracer); + } + + @Bean + public TestBeanOuter testBeanOuter(Tracer tracer, TestBeanInterface testBean) { + return new TestBeanOuter(tracer, testBean); } @Bean Reporter spanReporter() { diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectWebFluxTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectWebFluxTests.java new file mode 100644 index 000000000..b09f50fb9 --- /dev/null +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectWebFluxTests.java @@ -0,0 +1,238 @@ +/* + * Copyright 2013-2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.annotation; + +import brave.Tracer; +import brave.sampler.Sampler; +import org.apache.commons.lang3.StringUtils; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.web.server.LocalServerPort; +import org.springframework.cloud.sleuth.util.ArrayListSpanReporter; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.reactive.function.client.WebClient; +import reactor.core.publisher.Mono; +import zipkin2.Span; +import zipkin2.reporter.Reporter; + +import java.util.List; + +import static org.assertj.core.api.BDDAssertions.then; + +@RunWith(SpringJUnit4ClassRunner.class) +@SpringBootTest(classes = { + SleuthSpanCreatorAspectWebFluxTests.TestEndpoint.class, + SleuthSpanCreatorAspectWebFluxTests.TestConfiguration.class} + , webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) +@EnableAutoConfiguration +public class SleuthSpanCreatorAspectWebFluxTests { + + @Autowired Tracer tracer; + @Autowired ArrayListSpanReporter reporter; + + @Before + public void setup() { + this.reporter.clear(); + } + + @LocalServerPort + private int port; + + @Test + public void shouldReturnSpanFromWebFluxTraceContext() { + + Mono mono = WebClient.create().get().uri("http://localhost:"+port+"/test/ping") + .retrieve().bodyToMono(Long.class); + + then(this.reporter.getSpans()).isEmpty(); + + Long newSpanId = mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).kind()).isEqualTo(Span.Kind.SERVER); + then(spans.get(0).name()).isEqualTo("get /test/ping"); + then(spans.get(0).id()).isEqualTo(toHexString(newSpanId)); + then(this.tracer.currentSpan()).isNull(); + } + + @Test + public void shouldReturnSpanFromWebFluxSubscriptionContext() { + + Mono mono = WebClient.create().get().uri("http://localhost:"+port+"/test/pingFromContext") + .retrieve().bodyToMono(Long.class); + + then(this.reporter.getSpans()).isEmpty(); + + Long newSpanId = mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).kind()).isEqualTo(Span.Kind.SERVER); + then(spans.get(0).name()).isEqualTo("get /test/pingfromcontext"); + then(spans.get(0).id()).isEqualTo(toHexString(newSpanId)); + then(this.tracer.currentSpan()).isNull(); + } + + @Test + public void shouldContinueSpanInWebFlux() { + + Mono mono = WebClient.create().get().uri("http://localhost:"+port+"/test/continueSpan") + .retrieve().bodyToMono(Long.class); + + then(this.reporter.getSpans()).isEmpty(); + + Long newSpanId = mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).kind()).isEqualTo(Span.Kind.SERVER); + then(spans.get(0).name()).isEqualTo("get /test/continuespan"); + then(spans.get(0).id()).isEqualTo(toHexString(newSpanId)); + then(this.tracer.currentSpan()).isNull(); + } + + @Test + public void shouldCreateNewSpanInWebFlux() { + + Mono mono = WebClient.create().get().uri("http://localhost:"+port+"/test/newSpan1") + .retrieve().bodyToMono(Long.class); + + then(this.reporter.getSpans()).isEmpty(); + + Long newSpanId = mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(2); + then(spans.get(0).name()).isEqualTo("new-span-in-trace-context"); + then(spans.get(0).id()).isEqualTo(toHexString(newSpanId)); + then(spans.get(1).kind()).isEqualTo(Span.Kind.SERVER); + then(spans.get(1).name()).isEqualTo("get /test/newspan1"); + then(this.tracer.currentSpan()).isNull(); + } + + @Test + public void shouldCreateNewSpanInWebFluxInSubscriberContext() { + + Mono mono = WebClient.create().get().uri("http://localhost:"+port+"/test/newSpan2") + .retrieve().bodyToMono(Long.class); + + then(this.reporter.getSpans()).isEmpty(); + + Long newSpanId = mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(2); + then(spans.get(0).name()).isEqualTo("new-span-in-subscriber-context"); + then(spans.get(0).id()).isEqualTo(toHexString(newSpanId)); + then(spans.get(1).kind()).isEqualTo(Span.Kind.SERVER); + then(spans.get(1).name()).isEqualTo("get /test/newspan2"); + then(this.tracer.currentSpan()).isNull(); + } + + + private static String toHexString(long value){ + return StringUtils.leftPad(Long.toHexString(value), 16, '0'); + } + + + @Configuration + @EnableAutoConfiguration + protected static class TestConfiguration { + + @Bean + public TestBean testBean(Tracer tracer) { + return new TestBean(tracer); + } + + @Bean Reporter spanReporter() { + return new ArrayListSpanReporter(); + } + + @Bean Sampler alwaysSampler() { + return Sampler.ALWAYS_SAMPLE; + } + } + + @RestController + @RequestMapping("/test") + static class TestEndpoint { + + @Autowired Tracer tracer; + @Autowired TestBean testBean; + + @GetMapping("/ping") + public Mono ping() { + return Mono.just(tracer.currentSpan().context().spanId()); + } + + @GetMapping("/pingFromContext") + public Mono pingFromContext() { + return Mono.subscriberContext() + .flatMap(context -> Mono.just(tracer.currentSpan().context().spanId())); + } + + @GetMapping("/continueSpan") + public Mono continueSpan() { + return testBean.continueSpanInTraceContext(); + } + + @GetMapping("/newSpan1") + public Mono newSpan1() { + return testBean.newSpanInTraceContext(); + } + + @GetMapping("/newSpan2") + public Mono newSpan2() { + return testBean.newSpanInSubscriberContext(); + } + } + + static class TestBean { + + private final Tracer tracer; + + public TestBean(Tracer tracer) { + this.tracer = tracer; + } + + @ContinueSpan + public Mono continueSpanInTraceContext() { + return Mono.defer(() -> Mono.just(tracer.currentSpan().context().spanId())); + } + + @NewSpan(name = "newSpanInTraceContext") + public Mono newSpanInTraceContext() { + return Mono.defer(() -> Mono.just(tracer.currentSpan().context().spanId())); + } + + @NewSpan(name = "newSpanInSubscriberContext") + public Mono newSpanInSubscriberContext() { + return Mono.subscriberContext() + .flatMap(context -> Mono.defer(() -> Mono.just(tracer.currentSpan().context().spanId()))); + } + } +} 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 ca89c6b60..d51270554 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 @@ -20,13 +20,12 @@ import java.util.concurrent.atomic.AtomicReference; import brave.Span; import brave.Tracer; -import brave.Tracing; import brave.sampler.Sampler; -import org.junit.BeforeClass; import reactor.core.publisher.BaseSubscriber; import reactor.core.publisher.Flux; import reactor.core.publisher.Hooks; import reactor.core.publisher.Mono; +import reactor.core.publisher.Operators; import reactor.core.scheduler.Schedulers; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -65,8 +64,7 @@ public class SpanSubscriberTests { .map( d -> d + 1) .map( d -> d + 1) .map( (d) -> { - spanInOperation.set( - SpanSubscriberTests.this.tracer.currentSpan()); + spanInOperation.set(this.tracer.currentSpan()); return d + 1; }) .map( d -> d + 1) @@ -76,8 +74,8 @@ public class SpanSubscriberTests { } then(this.tracer.currentSpan()).isNull(); - then(spanInOperation.get().context().traceId()) - .isEqualTo(span.context().traceId()); + then(spanInOperation.get().context().spanId()) + .isEqualTo(span.context().spanId()); } @Test public void should_support_reactor_fusion_optimization() { @@ -88,7 +86,7 @@ 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(SpanSubscriberTests.this.tracer.currentSpan()); + spanInOperation.set(this.tracer.currentSpan()); return d + 1; }).map(d -> d + 1).subscribe(System.out::println); } finally { @@ -96,7 +94,7 @@ public class SpanSubscriberTests { } then(this.tracer.currentSpan()).isNull(); - then(spanInOperation.get().context().traceId()).isEqualTo(span.context().traceId()); + then(spanInOperation.get().context().spanId()).isEqualTo(span.context().spanId()); } @Test public void should_not_trace_scalar_flows() { @@ -112,7 +110,7 @@ public class SpanSubscriberTests { }); then(this.tracer.currentSpan()).isNotNull(); - then(spanInOperation.get()).isNotInstanceOf(SpanSubscriber.class); + then(spanInOperation.get()).isInstanceOf(ScopePassingSpanSubscriber.class); Mono.error(new Exception()) .subscribe(new BaseSubscriber() { @@ -127,7 +125,7 @@ public class SpanSubscriberTests { }); then(this.tracer.currentSpan()).isNotNull(); - then(spanInOperation.get()).isNotInstanceOf(SpanSubscriber.class); + then(spanInOperation.get()).isInstanceOf(ScopePassingSpanSubscriber.class); Mono.empty() .subscribe(new BaseSubscriber() { @@ -138,7 +136,7 @@ public class SpanSubscriberTests { }); then(this.tracer.currentSpan()).isNotNull(); - then(spanInOperation.get()).isNotInstanceOf(SpanSubscriber.class); + then(spanInOperation.get()).isEqualTo(Operators.emptySubscription()); } finally { span.finish(); } @@ -156,12 +154,12 @@ public class SpanSubscriberTests { 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(SpanSubscriberTests.this.tracer.currentSpan()); + spanInOperation.set(this.tracer.currentSpan()); return d + 1; }).map(d -> d + 1).blockLast(); Awaitility.await().untilAsserted(() -> { - then(spanInOperation.get().context().traceId()).isEqualTo(span.context().traceId()); + then(spanInOperation.get().context().spanId()).isEqualTo(span.context().spanId()); }); then(this.tracer.currentSpan()).isEqualTo(span); } finally { @@ -173,13 +171,13 @@ 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(SpanSubscriberTests.this.tracer.currentSpan()); + 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 - then(spanInOperation.get().context().traceId()).isEqualTo(foo2.context().traceId()); + then(spanInOperation.get().context().spanId()).isEqualTo(foo2.context().spanId()); } finally { foo2.finish(); } @@ -192,15 +190,15 @@ public class SpanSubscriberTests { Span parentSpan = this.tracer.nextSpan().name("foo").start(); log.info("Hello"); try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(parentSpan)) { - final Long traceId = Mono.fromCallable(tracer::currentSpan) - .map(span -> span.context().traceId()) + final Long spanId = Mono.fromCallable(tracer::currentSpan) + .map(span -> span.context().spanId()) .block(); - then(traceId).isNotNull(); + then(spanId).isNotNull(); - final Long secondTraceId = Mono.fromCallable(tracer::currentSpan) - .map(span -> span.context().traceId()) + final Long secondSpanId = Mono.fromCallable(tracer::currentSpan) + .map(span -> span.context().spanId()) .block(); - then(secondTraceId).isEqualTo(traceId); // different trace ids here + then(secondSpanId).isEqualTo(spanId); // different trace ids here } } @@ -212,17 +210,17 @@ public class SpanSubscriberTests { try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) { Mono.fromCallable(tracer::currentSpan) - .map(span -> span.context().traceId()) + .map(span -> span.context().spanId()) .doOnNext(spanInOperation::set) .zipWith( Mono.fromCallable(tracer::currentSpan) - .map(span -> span.context().traceId()) + .map(span -> span.context().spanId()) .doOnNext(spanInZipOperation::set)) .block(); } - then(spanInZipOperation).hasValue(initSpan.context().traceId()); // ok here - then(spanInOperation).hasValue(initSpan.context().traceId()); // Expecting to have value: <1L> but did not. + then(spanInZipOperation).hasValue(initSpan.context().spanId()); // ok here + then(spanInOperation).hasValue(initSpan.context().spanId()); // Expecting to have value: <1L> but did not. } // #646 @@ -237,9 +235,24 @@ public class SpanSubscriberTests { } } + //#1030 + @Test + public void checkTraceIdFromSubscriberContext() { + Span initSpan = this.tracer.nextSpan().name("foo").start(); + final AtomicReference spanInSubscriberContext = new AtomicReference<>(); + + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) { + Mono.subscriberContext() + .map(context -> tracer.currentSpan().context().spanId()) + .doOnNext(spanInSubscriberContext::set) + .block(); + } + + then(spanInSubscriberContext).hasValue(initSpan.context().spanId()); // ok here + } + @AfterClass public static void cleanup() { - Hooks.resetOnLastOperator(); Hooks.resetOnEachOperator(); Schedulers.resetFactory(); }