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 new file mode 100644 index 000000000..e75a4fc51 --- /dev/null +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java @@ -0,0 +1,52 @@ +package org.springframework.cloud.sleuth.instrument.reactor; + +import java.util.function.Function; +import java.util.function.Predicate; + +import org.reactivestreams.Publisher; +import org.springframework.cloud.sleuth.Tracer; + +import reactor.core.Fuseable; +import reactor.core.Scannable; +import reactor.core.publisher.Operators; + +/** + * Reactive Span pointcuts factories + * + * @author Stephane Maldini + * @since 2.0.0 + */ +public abstract class ReactorSleuth { + + /** + * Return a span operator pointcut given a {@link Tracer}. This can be used in reactor + * via {@link reactor.core.publisher.Flux#transform(Function)}, {@link + * reactor.core.publisher.Mono#transform(Function)}, {@link + * reactor.core.publisher.Hooks#onEachOperator(Function)} or {@link + * reactor.core.publisher.Hooks#onLastOperator(Function)}. + * + * @param tracer the {@link Tracer} instance to use in this span operator + * @param an arbitrary type that is left unchanged by the span operator + * + * @return a new Span operator pointcut + */ + public static Function, ? extends Publisher> spanOperator(Tracer tracer) { + return Operators.lift(POINTCUT_FILTER, ((scannable, sub) -> { + //do not trace fused flows + if(scannable instanceof Fuseable && sub instanceof Fuseable.QueueSubscription){ + return sub; + } + return new SpanSubscriber<>( + sub, + sub.currentContext(), + tracer, + scannable.name()); + })); + } + + private static final Predicate POINTCUT_FILTER = + s -> !(s instanceof Fuseable.ScalarCallable); + + private ReactorSleuth() { + } +} 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 new file mode 100644 index 000000000..3be48c36c --- /dev/null +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java @@ -0,0 +1,177 @@ +package org.springframework.cloud.sleuth.instrument.reactor; + +import java.util.concurrent.atomic.AtomicBoolean; + +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import org.springframework.cloud.sleuth.Span; +import org.springframework.cloud.sleuth.Tracer; +import reactor.core.CoreSubscriber; +import reactor.util.Logger; +import reactor.util.Loggers; +import reactor.util.context.Context; + +/** + * A trace representation of the {@link Subscriber} + * + * @author Stephane Maldini + * @author Marcin Grzejszczak + * @since 2.0.0 + */ +final class SpanSubscriber extends AtomicBoolean implements Subscription, + CoreSubscriber { + + private static final Logger log = Loggers.getLogger(SpanSubscriber.class); + + private final Span span; + private final Span rootSpan; + private final Subscriber subscriber; + private final Context context; + private final Tracer tracer; + private Subscription s; + + SpanSubscriber(Subscriber subscriber, Context ctx, Tracer tracer, + String name) { + this.subscriber = subscriber; + this.tracer = tracer; + Span root = ctx.getOrDefault(Span.class, tracer.getCurrentSpan()); + if (log.isTraceEnabled()) { + log.trace("Span from context [{}]", root); + } + this.rootSpan = root; + if (log.isTraceEnabled()) { + log.trace("Stored context root span [{}]", this.rootSpan); + } + this.span = tracer.createSpan(name, root); + if (log.isTraceEnabled()) { + log.trace("Created span [{}], with name [{}]", this.span, name); + } + this.context = ctx.put(Span.class, this.span); + } + + @Override public void onSubscribe(Subscription subscription) { + if (log.isTraceEnabled()) { + log.trace("On subscribe"); + } + this.s = subscription; + this.tracer.continueSpan(this.span); + if (log.isTraceEnabled()) { + log.trace("On subscribe - span continued"); + } + this.subscriber.onSubscribe(this); + } + + @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); + // 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 + Span localRootSpan = this.span; + while (localRootSpan != null) { + if (this.rootSpan != null) { + if (localRootSpan.getSpanId() != this.rootSpan.getSpanId() && + !isRootParentSpan(localRootSpan)) { + localRootSpan = continueDetachedSpan(localRootSpan); + } else { + localRootSpan = null; + } + } else if (!isRootParentSpan(localRootSpan)) { + localRootSpan = continueDetachedSpan(localRootSpan); + } else { + localRootSpan = null; + } + } + if (log.isTraceEnabled()) { + log.trace("Request after cleaning. Current span [{}]", + this.tracer.getCurrentSpan()); + } + } + + private boolean isRootParentSpan(Span localRootSpan) { + return localRootSpan.getSpanId() == localRootSpan.getTraceId(); + } + + private Span continueDetachedSpan(Span localRootSpan) { + if (log.isTraceEnabled()) { + log.trace("Will detach span {}", localRootSpan); + } + Span detachedSpan = this.tracer.detach(localRootSpan); + return this.tracer.continueSpan(detachedSpan); + } + + @Override public void cancel() { + try { + if (log.isTraceEnabled()) { + log.trace("Cancel"); + } + this.s.cancel(); + } + finally { + cleanup(); + } + } + + @Override public void onNext(T o) { + this.subscriber.onNext(o); + } + + @Override public void onError(Throwable throwable) { + try { + this.subscriber.onError(throwable); + } + finally { + cleanup(); + } + } + + @Override public void onComplete() { + try { + this.subscriber.onComplete(); + } + finally { + cleanup(); + } + } + + void cleanup() { + if (compareAndSet(false, true)) { + if (log.isTraceEnabled()) { + log.trace("Cleaning up"); + } + if (this.tracer.getCurrentSpan() != this.span) { + if (log.isTraceEnabled()) { + log.trace("Detaching span"); + } + this.tracer.detach(this.tracer.getCurrentSpan()); + this.tracer.continueSpan(this.span); + if (log.isTraceEnabled()) { + log.trace("Continuing span"); + } + } + if (log.isTraceEnabled()) { + log.trace("Closing span"); + } + this.tracer.close(this.span); + if (log.isTraceEnabled()) { + log.trace("Span closed"); + } + if (this.rootSpan != null) { + this.tracer.continueSpan(this.rootSpan); + this.tracer.close(this.rootSpan); + if (log.isTraceEnabled()) { + log.trace("Closed root span"); + } + } + } + } + + @Override public Context currentContext() { + return this.context; + } +} \ No newline at end of file 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 new file mode 100644 index 000000000..db88ade78 --- /dev/null +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java @@ -0,0 +1,66 @@ +package org.springframework.cloud.sleuth.instrument.reactor; + +import java.util.concurrent.ScheduledExecutorService; +import java.util.function.Supplier; +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; + +import org.springframework.beans.factory.annotation.Autowired; +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.ConditionalOnNotWebApplication; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.cloud.sleuth.SpanNamer; +import org.springframework.cloud.sleuth.TraceKeys; +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.publisher.Hooks; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +/** + * {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration Auto-configuration} + * to enable tracing of Reactor components via Spring Cloud Sleuth. + * + * @author Stephane Maldini + * @author Marcin Grzejszczak + * @since 2.0.0 + */ +@Configuration +@ConditionalOnProperty(value="spring.sleuth.reactor.enabled", matchIfMissing=true) +@ConditionalOnClass(Mono.class) +@AutoConfigureAfter(TraceWebFluxAutoConfiguration.class) +public class TraceReactorAutoConfiguration { + + @Configuration + @ConditionalOnBean(Tracer.class) + @ConditionalOnNotWebApplication + static class TraceReactorConfiguration { + @Autowired Tracer tracer; + @Autowired TraceKeys traceKeys; + @Autowired SpanNamer spanNamer; + + @PostConstruct + public void setupHooks() { + Hooks.onLastOperator(ReactorSleuth.spanOperator(this.tracer)); + Schedulers.setFactory(new Schedulers.Factory() { + @Override public ScheduledExecutorService decorateExecutorService(String schedulerType, + Supplier actual) { + return new TraceableScheduledExecutorService(actual.get(), + TraceReactorConfiguration.this.tracer, + TraceReactorConfiguration.this.traceKeys, + TraceReactorConfiguration.this.spanNamer); + } + }); + } + + @PreDestroy + public void cleanupHooks() { + Hooks.resetOnLastOperator(); + Schedulers.resetFactory(); + } + } +} \ No newline at end of file diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFluxAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFluxAutoConfiguration.java new file mode 100644 index 000000000..bb5867a27 --- /dev/null +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFluxAutoConfiguration.java @@ -0,0 +1,47 @@ +/* + * Copyright 2013-2015 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.instrument.web; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication; +import org.springframework.cloud.sleuth.Tracer; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration + * Auto-configuration} enables tracing to HTTP requests with Spring WebFlux. + * + * @author Marcin Grzejszczak + * @since 2.0.0 + */ +@Configuration +@ConditionalOnProperty(value = "spring.sleuth.web.enabled", matchIfMissing = true) +@ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.REACTIVE) +@ConditionalOnBean(Tracer.class) +@AutoConfigureAfter(TraceWebAutoConfiguration.class) +public class TraceWebFluxAutoConfiguration { + + @Bean + public TraceWebFilter traceFilter(BeanFactory beanFactory, + SkipPatternProvider skipPatternProvider) { + return new TraceWebFilter(beanFactory, skipPatternProvider.skipPattern()); + } + +} diff --git a/spring-cloud-sleuth-core/src/main/resources/META-INF/spring.factories b/spring-cloud-sleuth-core/src/main/resources/META-INF/spring.factories index 7cca648e5..16dd38a41 100644 --- a/spring-cloud-sleuth-core/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-sleuth-core/src/main/resources/META-INF/spring.factories @@ -9,10 +9,12 @@ org.springframework.cloud.sleuth.instrument.messaging.websocket.TraceWebSocketAu org.springframework.cloud.sleuth.instrument.async.AsyncCustomAutoConfiguration,\ org.springframework.cloud.sleuth.instrument.async.AsyncDefaultAutoConfiguration,\ org.springframework.cloud.sleuth.instrument.hystrix.SleuthHystrixAutoConfiguration,\ +org.springframework.cloud.sleuth.instrument.reactor.TraceReactorAutoConfiguration,\ org.springframework.cloud.sleuth.instrument.scheduling.TraceSchedulingAutoConfiguration,\ org.springframework.cloud.sleuth.instrument.web.TraceHttpAutoConfiguration,\ org.springframework.cloud.sleuth.instrument.web.TraceWebAutoConfiguration,\ org.springframework.cloud.sleuth.instrument.web.TraceWebServletAutoConfiguration,\ +org.springframework.cloud.sleuth.instrument.web.TraceWebFluxAutoConfiguration,\ org.springframework.cloud.sleuth.instrument.web.client.TraceWebClientAutoConfiguration,\ org.springframework.cloud.sleuth.instrument.web.client.TraceWebAsyncClientAutoConfiguration,\ org.springframework.cloud.sleuth.instrument.web.client.feign.TraceFeignClientAutoConfiguration,\ 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 new file mode 100644 index 000000000..7e39f6413 --- /dev/null +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java @@ -0,0 +1,199 @@ +package org.springframework.cloud.sleuth.instrument.reactor; + +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.awaitility.Awaitility; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Ignore; +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; +import org.springframework.cloud.sleuth.Sampler; +import org.springframework.cloud.sleuth.Span; +import org.springframework.cloud.sleuth.Tracer; +import org.springframework.cloud.sleuth.sampler.AlwaysSampler; +import org.springframework.cloud.sleuth.trace.TestSpanContextHolder; +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; + +@RunWith(SpringRunner.class) +@SpringBootTest(classes = SpanSubscriberTests.Config.class, + webEnvironment = SpringBootTest.WebEnvironment.NONE) +public class SpanSubscriberTests { + + private static final Log log = LogFactory.getLog(SpanSubscriberTests.class); + + @Autowired Tracer tracer; + + @Before + public void setup() { + ExceptionUtils.setFail(true); + } + + @Test public void should_pass_tracing_info_when_using_reactor() { + Span span = this.tracer.createSpan("foo"); + final AtomicReference spanInOperation = new AtomicReference<>(); + Publisher traced = Flux.just(1, 2, 3); + log.info("Hello"); + + Flux.from(traced) + .map( d -> d + 1) + .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(); + } + + @Ignore("Ignored until fixed in Reactor") + @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() { + 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() { + + Span span = this.tracer.createSpan("foo"); + final AtomicReference spanInOperation = new AtomicReference<>(); + log.info("Hello"); + + 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.getCurrentSpan()); + return d + 1; + }) + .map( d -> d + 1) + .blockLast(); + + Awaitility.await().untilAsserted(() -> { + then(spanInOperation.get().getTraceId()).isEqualTo(span.getTraceId()); + then(ExceptionUtils.getLastException()).isNull(); + }); + then(this.tracer.getCurrentSpan()).isEqualTo(span); + this.tracer.close(span); + + Span foo2 = this.tracer.createSpan("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.getCurrentSpan()); + return d + 1; + }) + .map( d -> d + 1) + .blockLast(); + + then(this.tracer.getCurrentSpan()).isEqualTo(foo2); + then(ExceptionUtils.getLastException()).isNull(); + // parent cause there's an async span in the meantime + then(spanInOperation.get().getTraceId()).isEqualTo(foo2.getTraceId()); + tracer.close(foo2); + } + + @AfterClass + public static void cleanup() { + Hooks.resetOnLastOperator(); + Schedulers.resetFactory(); + } + + @EnableAutoConfiguration + @Configuration + static class Config { + @Bean Sampler sampler() { + return new AlwaysSampler(); + } + } +} \ No newline at end of file diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFluxTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFluxTests.java new file mode 100644 index 000000000..50b560c17 --- /dev/null +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFluxTests.java @@ -0,0 +1,105 @@ +package org.springframework.cloud.sleuth.instrument.web; + +import org.awaitility.Awaitility; +import org.junit.BeforeClass; +import org.junit.Ignore; +import org.junit.Test; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.sleuth.Sampler; +import org.springframework.cloud.sleuth.Span; +import org.springframework.cloud.sleuth.SpanReporter; +import org.springframework.cloud.sleuth.Tracer; +import org.springframework.cloud.sleuth.assertions.ListOfSpans; +import org.springframework.cloud.sleuth.assertions.SleuthAssertions; +import org.springframework.cloud.sleuth.sampler.AlwaysSampler; +import org.springframework.cloud.sleuth.util.ArrayListSpanAccumulator; +import org.springframework.cloud.sleuth.util.ExceptionUtils; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.env.Environment; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.reactive.function.client.ClientResponse; +import org.springframework.web.reactive.function.client.WebClient; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Hooks; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +public class TraceWebFluxTests { + + @BeforeClass + public static void setup() { + Hooks.resetOnLastOperator(); + Schedulers.resetFactory(); + } + + @Ignore("Ignored until fixed in Reactor") + @Test public void should_instrument_web_filter() throws Exception { + ConfigurableApplicationContext context = new SpringApplicationBuilder(TraceWebFluxTests.Config.class) + .web(WebApplicationType.REACTIVE).properties("server.port=0", "spring.jmx.enabled=false", + "spring.application.name=TraceWebFluxTests").run(); + ExceptionUtils.setFail(true); + Span span = null; + try { + span = context.getBean(Tracer.class).createSpan("foo"); + int port = context.getBean(Environment.class).getProperty("local.server.port", Integer.class); + ArrayListSpanAccumulator accumulator = context.getBean(ArrayListSpanAccumulator.class); + + Mono exchange = context.getBean(WebClient.class).get().uri("http://localhost:" + port + "/api/c2/10").exchange(); + + Awaitility.await().untilAsserted(() -> { + ClientResponse response = exchange.block(); + SleuthAssertions.then(response.statusCode().value()).isEqualTo(200); + SleuthAssertions.then(ExceptionUtils.getLastException()).isNull(); + SleuthAssertions.then(new ListOfSpans(accumulator.getSpans())) + .hasASpanWithLogEqualTo(Span.CLIENT_SEND) + .hasASpanWithLogEqualTo(Span.SERVER_RECV) + .hasASpanWithLogEqualTo(Span.SERVER_SEND) + .hasASpanWithLogEqualTo(Span.CLIENT_RECV) + .hasASpanWithTagEqualTo("mvc.controller.method", "successful") + .hasASpanWithTagEqualTo("mvc.controller.class", "Controller2"); + }); + } finally { + context.getBean(Tracer.class).close(span); + } + + } + + @Configuration + @EnableAutoConfiguration + static class Config { + + @Bean WebClient webClient() { + return WebClient.create(); + } + + @Bean Sampler sampler() { + return new AlwaysSampler(); + } + + @Bean SpanReporter spanReporter() { + return new ArrayListSpanAccumulator(); + } + + @Bean + Controller2 controller2() { + return new Controller2(); + } + } + + @RestController + @RequestMapping("/api/c2") + static class Controller2 { + @GetMapping("/{id}") + public Flux successful(@PathVariable Long id) { + return Flux.just(id.toString()); + } + } +} +