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

Set span in context (#1056)

Browse Source 
* tests on trace context
* improve performance : excessive check removed (#1053)
* Fixed checkstyle
* set span in trace context
* explicit imports
* deprecated

fixes #1030
This commit is contained in:
Sergii Karpenko
2018-08-14 19:02:18 +03:00
committed by Marcin Grzejszczak
parent 786ad2c7c5
commit fdeb675c0e
9 changed files with 529 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/SleuthAdvisorConfig.java
View File

@@ -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());

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

@@ -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 <T> an arbitrary type that is left unchanged by the span operator
*
* @return a new lazy span operator pointcut
*/
@SuppressWarnings("unchecked")
@Deprecated
public static <T> Function<? super Publisher<T>, ? extends Publisher<T>> 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

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

@@ -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<T> extends AtomicBoolean implements SpanSubscription<T> {
private static final Logger log = Loggers.getLogger(

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

@@ -38,7 +38,7 @@ class SpanSubscriptionProvider<T> implements Supplier<SpanSubscription<T>> {
final Subscriber<? super T> subscriber;
final Context context;
final String name;
private Tracing tracing;
private volatile Tracing tracing;
SpanSubscriptionProvider(BeanFactory beanFactory,
Subscriber<? super T> subscriber,

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

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

74
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectFluxTests.java
View File

@@ -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<zipkin2.Span> spans = new ArrayList<>(this.reporter.getSpans());
then(spans).isEmpty();
then(this.tracer.currentSpan()).isNull();
}
@Test
public void shouldReturnNewSpanFromTraceContext() {
Flux<Long> flux = this.testBean.newSpanInTraceContext();
Long newSpanId = flux.blockFirst();
List<zipkin2.Span> 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<Long> flux = this.testBean.newSpanInSubscriberContext();
Long newSpanId = flux.blockFirst();
List<zipkin2.Span> 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<String> flux) {
@@ -377,6 +422,12 @@ public class SleuthSpanCreatorAspectFluxTests {
@ContinueSpan
Flux<String> testMethod14(String param);
@NewSpan(name = "spanInTraceContext")
Flux<Long> newSpanInTraceContext();
@NewSpan(name = "spanInSubscriberContext")
Flux<Long> 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<CompletableFuture<Void>> proceed
= new AtomicReference<>(new CompletableFuture<>());
private Flux<String> 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<String> testMethod14(String param) {
return Flux.just(TEST_STRING1, TEST_STRING2);
}
@Override
public Flux<Long> newSpanInTraceContext() {
return Flux.defer(() -> Flux.just(tracer.currentSpan().context().spanId()));
}
@Override
public Flux<Long> 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<zipkin2.Span> spanReporter() {

158
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectMonoTests.java
View File

@@ -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<zipkin2.Span> spans = new ArrayList<>(this.reporter.getSpans());
then(spans).isEmpty();
then(this.tracer.currentSpan()).isNull();
}
@Test
public void shouldReturnNewSpanFromTraceContext() {
Mono<Long> mono = this.testBean.newSpanInTraceContext();
then(this.reporter.getSpans()).isEmpty();
Long newSpanId = mono.block();
List<zipkin2.Span> 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<Pair<Pair<Long, Long>, Long>> mono = this.testBeanOuter.outerNewSpanInTraceContext();
then(this.reporter.getSpans()).isEmpty();
Pair<Pair<Long, Long>, 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<zipkin2.Span> 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<Long> mono = this.testBean.newSpanInSubscriberContext();
then(this.reporter.getSpans()).isEmpty();
Long newSpanId = mono.block();
List<zipkin2.Span> 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<Pair<Pair<Long, Long>, Long>> mono = this.testBeanOuter.outerNewSpanInSubscriberContext();
then(this.reporter.getSpans()).isEmpty();
Pair<Pair<Long, Long>, 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<zipkin2.Span> 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<String> testMethod13();
@NewSpan(name = "spanInTraceContext")
Mono<Long> newSpanInTraceContext();
@NewSpan(name = "spanInSubscriberContext")
Mono<Long> 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<String> TEST_MONO = Mono.defer(() -> just(TEST_STRING));
private final Tracer tracer;
public TestBean(Tracer tracer) {
this.tracer = tracer;
}
@Override
public Mono<String> testMethod() {
return TEST_MONO;
@@ -458,6 +566,43 @@ public class SleuthSpanCreatorAspectMonoTests {
public Mono<String> testMethod13() {
return Mono.defer(() -> Mono.error(new RuntimeException("test exception 13")));
}
@Override
public Mono<Long> newSpanInTraceContext() {
return Mono.defer(() -> Mono.just(tracer.currentSpan().context().spanId()));
}
@Override
public Mono<Long> 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<Pair<Pair<Long, Long>, 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<Pair<Pair<Long, Long>, 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<zipkin2.Span> spanReporter() {

238
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectWebFluxTests.java Normal file
View File

@@ -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<Long> mono = WebClient.create().get().uri("http://localhost:"+port+"/test/ping")
.retrieve().bodyToMono(Long.class);
then(this.reporter.getSpans()).isEmpty();
Long newSpanId = mono.block();
List<zipkin2.Span> 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<Long> mono = WebClient.create().get().uri("http://localhost:"+port+"/test/pingFromContext")
.retrieve().bodyToMono(Long.class);
then(this.reporter.getSpans()).isEmpty();
Long newSpanId = mono.block();
List<zipkin2.Span> 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<Long> mono = WebClient.create().get().uri("http://localhost:"+port+"/test/continueSpan")
.retrieve().bodyToMono(Long.class);
then(this.reporter.getSpans()).isEmpty();
Long newSpanId = mono.block();
List<zipkin2.Span> 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<Long> mono = WebClient.create().get().uri("http://localhost:"+port+"/test/newSpan1")
.retrieve().bodyToMono(Long.class);
then(this.reporter.getSpans()).isEmpty();
Long newSpanId = mono.block();
List<zipkin2.Span> 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<Long> mono = WebClient.create().get().uri("http://localhost:"+port+"/test/newSpan2")
.retrieve().bodyToMono(Long.class);
then(this.reporter.getSpans()).isEmpty();
Long newSpanId = mono.block();
List<zipkin2.Span> 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<zipkin2.Span> 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<Long> ping() {
return Mono.just(tracer.currentSpan().context().spanId());
}
@GetMapping("/pingFromContext")
public Mono<Long> pingFromContext() {
return Mono.subscriberContext()
.flatMap(context -> Mono.just(tracer.currentSpan().context().spanId()));
}
@GetMapping("/continueSpan")
public Mono<Long> continueSpan() {
return testBean.continueSpanInTraceContext();
}
@GetMapping("/newSpan1")
public Mono<Long> newSpan1() {
return testBean.newSpanInTraceContext();
}
@GetMapping("/newSpan2")
public Mono<Long> newSpan2() {
return testBean.newSpanInSubscriberContext();
}
}
static class TestBean {
private final Tracer tracer;
public TestBean(Tracer tracer) {
this.tracer = tracer;
}
@ContinueSpan
public Mono<Long> continueSpanInTraceContext() {
return Mono.defer(() -> Mono.just(tracer.currentSpan().context().spanId()));
}
@NewSpan(name = "newSpanInTraceContext")
public Mono<Long> newSpanInTraceContext() {
return Mono.defer(() -> Mono.just(tracer.currentSpan().context().spanId()));
}
@NewSpan(name = "newSpanInSubscriberContext")
public Mono<Long> newSpanInSubscriberContext() {
return Mono.subscriberContext()
.flatMap(context -> Mono.defer(() -> Mono.just(tracer.currentSpan().context().spanId())));
}
}
}

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

@@ -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.<Integer>error(new Exception())
.subscribe(new BaseSubscriber<Integer>() {
@@ -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.<Integer>empty()
.subscribe(new BaseSubscriber<Integer>() {
@@ -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 <AtomicReference[null]> to have value: <1L> but did not.
then(spanInZipOperation).hasValue(initSpan.context().spanId()); // ok here
then(spanInOperation).hasValue(initSpan.context().spanId()); // Expecting <AtomicReference[null]> 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<Long> 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();
}