From 1c33332320e0516e9dcf4fa3b06f642bb24ac2cf Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Tue, 28 Jul 2020 18:02:16 +0200 Subject: [PATCH] Further WebFlux and Messaging Reactor Sleuth improvements (#1699) * Improving performance and adding docs --- benchmarks/pom.xml | 7 +- .../SleuthBenchmarkingStreamApplication.java | 85 ++++++------ .../SleuthBenchmarkingSpringWebFluxApp.java | 13 +- .../jmh/stream/MicroBenchmarkStreamTests.java | 33 +++-- .../jmh/webflux/MicroBenchmarkHttpTests.java | 18 +-- .../main/asciidoc/spring-cloud-sleuth.adoc | 33 ++++- .../async/AsyncAutoConfiguration.java | 4 + ...tor.java => MessagingSleuthOperators.java} | 12 +- .../TraceFunctionAutoConfiguration.java | 4 +- .../reactor/SleuthReactorProperties.java | 16 ++- .../sleuth/instrument/web/TraceWebFilter.java | 41 ++++-- .../web/WebFluxSleuthOperators.java | 106 ++++++++++++++- .../client/TraceRequestHttpHeadersFilter.java | 21 ++- .../TraceWebClientBeanPostProcessor.java | 40 +++--- .../StreamMessageOperatorsTests.java | 10 +- .../reactor/sample/FlatMapTests.java | 121 +++++++++++++++--- .../reactor/sample/ManualRequestSender.java | 62 +++++++++ .../reactor/sample/RequestSender.java | 4 +- 18 files changed, 479 insertions(+), 151 deletions(-) rename spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/{MessagingSleuthOperator.java => MessagingSleuthOperators.java} (97%) create mode 100644 tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/ManualRequestSender.java diff --git a/benchmarks/pom.xml b/benchmarks/pom.xml index 50f1e61a6..90d2b6157 100644 --- a/benchmarks/pom.xml +++ b/benchmarks/pom.xml @@ -28,7 +28,7 @@ org.springframework.boot spring-boot-starter-parent - 2.4.0-M1 + 2.4.0-SNAPSHOT @@ -147,6 +147,11 @@ org.apache.httpcomponents httpclient + + org.awaitility + awaitility + test + diff --git a/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java b/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java index a304afcbd..5a04f43cf 100644 --- a/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java +++ b/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java @@ -24,7 +24,6 @@ import java.util.concurrent.TimeUnit; import java.util.function.Function; import brave.Tracing; -import brave.propagation.TraceContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import reactor.core.publisher.Flux; @@ -35,8 +34,7 @@ import reactor.core.scheduler.Schedulers; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.cloud.sleuth.instrument.messaging.MessagingSleuthOperator; -import org.springframework.cloud.sleuth.instrument.web.WebFluxSleuthOperators; +import org.springframework.cloud.sleuth.instrument.messaging.MessagingSleuthOperators; import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; @@ -45,7 +43,6 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; -import org.springframework.util.Assert; import static org.assertj.core.api.Assertions.assertThat; @@ -88,35 +85,42 @@ public class SleuthBenchmarkingStreamApplication { @Bean(name = "myFlux") @ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "simple") - public Function nonReactiveSimpleSleuthFunction() { + public Function simple() { System.out.println("simple_function"); return new SimpleFunction(); } @Bean(name = "myFlux") @ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "reactive_simple") - public Function, Flux> reactiveSimpleSleuthFunction() { + public Function, Flux> reactiveSimple() { System.out.println("simple_reactive_function"); return new SimpleReactiveFunction(); } + @Bean(name = "myFlux") + @ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "simple_function_with_around") + public Function, Message> simpleFunctionWithAround() { + System.out.println("simple_function_with_around"); + return new SimpleMessageFunction(); + } + @Bean(name = "myFlux") @ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "simple_manual") - public Function, Message> nonReactiveSimpleManualSleuthFunction(Tracing tracing) { + public Function, Message> simpleManual(Tracing tracing) { System.out.println("simple_manual_function"); return new SimpleManualFunction(tracing); } @Bean(name = "myFlux") @ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "reactive_simple_manual") - public Function>, Flux>> reactiveSimpleManualSleuthFunction(Tracing tracing) { + public Function>, Flux>> reactiveSimpleManual(Tracing tracing) { System.out.println("simple_reactive_manual_function"); return new SimpleReactiveManualFunction(tracing); } @Bean(name = "myFlux") @ConditionalOnProperty(value = "spring.sleuth.nonreactive.function.enabled", havingValue = "true") - public Function nonReactiveSleuthFunction(ExecutorService executorService) { + public Function nonReactiveFunction(ExecutorService executorService) { System.out.println("no sleuth non reactive function"); return new SleuthNonReactiveFunction(executorService); } @@ -136,13 +140,6 @@ public class SleuthBenchmarkingStreamApplication { return new SleuthFunction(); } - @Bean(name = "myFlux") - @ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "MANUAL") - public Function, Flux> manualFunction() { - System.out.println("manual function"); - return new SleuthManualFunction(); - } - } class SimpleFunction implements Function { @@ -151,6 +148,7 @@ class SimpleFunction implements Function { @Override public String apply(String input) { + // tracing works cause headers from the input message get propagated to the output message log.info("Hello from simple [{}]", input); return input.toUpperCase(); } @@ -180,18 +178,31 @@ class SimpleManualFunction implements Function, Message> @Override public Message apply(Message input) { - return (MessagingSleuthOperator.asFunction(this.tracing, input) - .andThen(msg -> MessagingSleuthOperator.withSpanInScope(this.tracing, msg, stringMessage -> { + return (MessagingSleuthOperators.asFunction(this.tracing, input) + .andThen(msg -> MessagingSleuthOperators.withSpanInScope(this.tracing, msg, stringMessage -> { log.info("Hello from simple manual [{}]", stringMessage.getPayload()); return stringMessage; - })).andThen(msg -> MessagingSleuthOperator.afterMessageHandled(this.tracing, msg, null)) - .andThen(msg -> MessagingSleuthOperator.handleOutputMessage(this.tracing, msg)) + })).andThen(msg -> MessagingSleuthOperators.afterMessageHandled(this.tracing, msg, null)) + .andThen(msg -> MessagingSleuthOperators.handleOutputMessage(this.tracing, msg)) .andThen(msg -> MessageBuilder.createMessage(msg.getPayload().toUpperCase(), msg.getHeaders())) - .andThen(msg -> MessagingSleuthOperator.afterMessageHandled(this.tracing, msg, null)).apply(input)); + .andThen(msg -> MessagingSleuthOperators.afterMessageHandled(this.tracing, msg, null)).apply(input)); } } +class SimpleMessageFunction implements Function, Message> { + + private static final Logger log = LoggerFactory.getLogger(SimpleFunction.class); + + @Override + public Message apply(Message input) { + log.info("Hello from message simple [{}]", input.getPayload()); + return MessageBuilder.withPayload(input.getPayload().toUpperCase()).build(); + } + +} + +// tag::simple_reactive[] class SimpleReactiveManualFunction implements Function>, Flux>> { private static final Logger log = LoggerFactory.getLogger(SimpleReactiveFunction.class); @@ -204,16 +215,17 @@ class SimpleReactiveManualFunction implements Function>, Fl @Override public Flux> apply(Flux> input) { - return input.map(message -> (MessagingSleuthOperator.asFunction(this.tracing, message)) - .andThen(msg -> MessagingSleuthOperator.withSpanInScope(this.tracing, msg, stringMessage -> { + return input.map(message -> (MessagingSleuthOperators.asFunction(this.tracing, message)) + .andThen(msg -> MessagingSleuthOperators.withSpanInScope(this.tracing, msg, stringMessage -> { log.info("Hello from simple manual [{}]", stringMessage.getPayload()); return stringMessage; - })).andThen(msg -> MessagingSleuthOperator.afterMessageHandled(this.tracing, msg, null)) + })).andThen(msg -> MessagingSleuthOperators.afterMessageHandled(this.tracing, msg, null)) .andThen(msg -> MessageBuilder.createMessage(msg.getPayload().toUpperCase(), msg.getHeaders())) - .andThen(msg -> MessagingSleuthOperator.handleOutputMessage(this.tracing, msg)).apply(message)); + .andThen(msg -> MessagingSleuthOperators.handleOutputMessage(this.tracing, msg)).apply(message)); } } +// end::simple_reactive[] class SleuthNonReactiveFunction implements Function { @@ -257,26 +269,3 @@ class SleuthFunction implements Function, Flux> { } } - -class SleuthManualFunction implements Function, Flux> { - - private static final Logger log = LoggerFactory.getLogger(SleuthManualFunction.class); - - static final Scheduler SCHEDULER = Schedulers.newParallel("sleuthManualFunction"); - - @Override - public Flux apply(Flux input) { - return input.doOnEach(WebFluxSleuthOperators.withSpanInScope(() -> log.info("Got a message"))) - .flatMap(s -> Mono.subscriberContext().delayElement(Duration.ofMillis(1), SCHEDULER).map(ctx -> { - WebFluxSleuthOperators.withSpanInScope(ctx, () -> log.info("Logging [{}] from flat map", s)); - return s.toUpperCase(); - })).doOnEach(signal -> { - WebFluxSleuthOperators.withSpanInScope(signal.getContext(), () -> log.info("Doing assertions")); - TraceContext traceContext = signal.getContext().get(TraceContext.class); - Assert.notNull(traceContext, "Context must be set by Sleuth instrumentation"); - Assert.state(traceContext.traceIdString().equals("4883117762eb9420"), "TraceId must be propagated"); - log.info("Assertions passed"); - }); - } - -} diff --git a/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/webflux/SleuthBenchmarkingSpringWebFluxApp.java b/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/webflux/SleuthBenchmarkingSpringWebFluxApp.java index 0757f7ade..062f68403 100644 --- a/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/webflux/SleuthBenchmarkingSpringWebFluxApp.java +++ b/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/webflux/SleuthBenchmarkingSpringWebFluxApp.java @@ -27,6 +27,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.publisher.SignalType; import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; @@ -102,9 +103,17 @@ public class SleuthBenchmarkingSpringWebFluxApp implements ApplicationListener simple() { - return Mono.just("hello").map(String::toUpperCase); + return Mono.just("hello").map(String::toUpperCase).doOnNext(s -> log.info("Hello from simple [{}]", s)); } + // tag::simple_manual[] + @GetMapping("/simpleManual") + public Mono simpleManual() { + return Mono.just("hello").map(String::toUpperCase).doOnEach(WebFluxSleuthOperators + .withSpanInScope(SignalType.ON_NEXT, signal -> log.info("Hello from simple [{}]", signal.get()))); + } + // end::simple_manual[] + @GetMapping("/complexNoSleuth") public Mono complexNoSleuth() { return Flux.range(1, 10).map(String::valueOf).collect(Collectors.toList()) @@ -134,7 +143,7 @@ public class SleuthBenchmarkingSpringWebFluxApp implements ApplicationListener complexManual() { return Flux.range(1, 10).map(String::valueOf).collect(Collectors.toList()) - .doOnEach(WebFluxSleuthOperators.withSpanInScope(() -> log.info("Got a request"))) + .doOnEach(WebFluxSleuthOperators.withSpanInScope(SignalType.ON_NEXT, () -> log.info("Got a request"))) .flatMap(s -> Mono.subscriberContext().delayElement(Duration.ofMillis(1), FOO_SCHEDULER).map(ctx -> { WebFluxSleuthOperators.withSpanInScope(ctx, () -> log.info("Logging [{}] from flat map", s)); return ""; diff --git a/benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/stream/MicroBenchmarkStreamTests.java b/benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/stream/MicroBenchmarkStreamTests.java index 74abc61a8..1717e8a1f 100644 --- a/benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/stream/MicroBenchmarkStreamTests.java +++ b/benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/stream/MicroBenchmarkStreamTests.java @@ -89,9 +89,15 @@ public class MicroBenchmarkStreamTests { this.output = this.applicationContext.getBean(OutputDestination.class); } + private void sendInputMessage() { + // System.out.println("Sending the message to input"); + input.send(MessageBuilder.withPayload("hello".getBytes()) + .setHeader("b3", "4883117762eb9420-4883117762eb9420-1").build()); + } + protected ConfigurableApplicationContext initContext() { SpringApplication application = new SpringApplicationBuilder(SleuthBenchmarkingStreamApplication.class) - .web(WebApplicationType.REACTIVE).application(); + .web(WebApplicationType.NONE).application(); return application.run(runArgs()); } @@ -104,9 +110,11 @@ public class MicroBenchmarkStreamTests { } void run() { - // System.out.println("Sending the message to input"); - input.send(MessageBuilder.withPayload("hello".getBytes()) - .setHeader("b3", "4883117762eb9420-4883117762eb9420-1").build()); + sendInputMessage(); + assertThatOutputMessageGotReceived(); + } + + private void assertThatOutputMessageGotReceived() { // System.out.println("Retrieving the message for tests"); Message message = output.receive(200L); // System.out.println("Got the message from output"); @@ -138,19 +146,18 @@ public class MicroBenchmarkStreamTests { public enum Instrumentation { noSleuthSimple("spring.sleuth.enabled=false,spring.sleuth.function.type=simple"), sleuthSimple( - "spring.sleuth.reactor.instrumentation-type=MANUAL,spring.sleuth.function.type=simple"), noSleuthReactiveSimple( - "spring.sleuth.enabled=false,spring.sleuth.function.type=reactive_simple"), sleuthReactiveSimpleOnEach( - "spring.sleuth.reactor.instrumentation-type=DECORATE_ON_EACH,spring.sleuth.integration.enabled=true,spring.sleuth.function.type=DECORATE_ON_EACH"), + "spring.sleuth.function.type=simple"), sleuthSimpleWithAround( + "spring.sleuth.function.type=simple_function_with_around"), noSleuthReactiveSimple( + "spring.sleuth.enabled=false,spring.sleuth.function.type=reactive_simple"), sleuthReactiveSimpleManual( + "spring.sleuth.function.type=reactive_simple_manual"), sleuthReactiveSimpleOnEach( + "spring.sleuth.reactor.instrumentation-type=DECORATE_ON_EACH,spring.sleuth.integration.enabled=true,spring.sleuth.function.type=DECORATE_ON_EACH"), // This won't work with messaging // sleuthReactiveSimpleOnLast("spring.sleuth.reactor.instrumentation-type=DECORATE_ON_LAST,spring.sleuth.function.type=DECORATE_ON_LAST"), // NO FUNCTION, NO INTEGRATION, MANUAL OPERATORS sleuthSimpleManual( - "spring.sleuth.function.enabled=false,spring.sleuth.integration.enabled=false,spring.sleuth.function.type=simple_manual"), sleuthReactiveSimpleManual( - "spring.sleuth.function.enabled=false,spring.sleuth.integration.enabled=false,spring.sleuth.function.type=reactive_simple_manual"), - // NO FUNCTION - OLD INTEGRATION STYLE - sleuthSimpleNoFunctionInstrumentationManual( - "spring.sleuth.function.type=simple_manual,spring.sleuth.function.enabled=false,spring.sleuth.integration.enabled=true,spring.sleuth.reactor.instrumentation-type=MANUAL"), sleuthReactiveSimpleNoFunctionInstrumentationManual( - "spring.sleuth.function.type=reactive_simple_manual,spring.sleuth.function.enabled=false,spring.sleuth.integration.enabled=true,spring.sleuth.reactor.instrumentation-type=MANUAL"); + "spring.sleuth.function.enabled=false,spring.sleuth.integration.enabled=false,spring.sleuth.function.type=simple_manual"), sleuthSimpleNoFunctionInstrumentationManual( + "spring.sleuth.function.type=simple_manual,spring.sleuth.function.enabled=false,spring.sleuth.integration.enabled=true,spring.sleuth.reactor.instrumentation-type=MANUAL"), sleuthReactiveSimpleNoFunctionInstrumentationManual( + "spring.sleuth.function.type=reactive_simple_manual,spring.sleuth.function.enabled=false,spring.sleuth.integration.enabled=true,spring.sleuth.reactor.instrumentation-type=MANUAL"); private Set entires = new HashSet<>(); diff --git a/benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/webflux/MicroBenchmarkHttpTests.java b/benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/webflux/MicroBenchmarkHttpTests.java index b972b56df..360b8baf7 100644 --- a/benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/webflux/MicroBenchmarkHttpTests.java +++ b/benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/webflux/MicroBenchmarkHttpTests.java @@ -106,16 +106,18 @@ public class MicroBenchmarkHttpTests { noSleuthSimple("spring.sleuth.enabled", "false", "/simple"), sleuthSimpleManual( "spring.sleuth.reactor.instrumentation-type", "MANUAL", - "/simple"), sleuthSimpleOnEach("spring.sleuth.reactor.instrumentation-type", "DECORATE_ON_EACH", - "/simple"), sleuthSimpleOnLast("spring.sleuth.reactor.instrumentation-type", - "DECORATE_ON_LAST", "/simple"), noSleuthComplex("spring.sleuth.enabled", "false", - "/complexNoSleuth"), onEachComplex( - "spring.sleuth.reactor.instrumentation-type", "DECORATE_ON_EACH", - "/complex"), onLastComplex( + "/simple"), sleuthManual("spring.sleuth.reactor.instrumentation-type", "MANUAL", + "/simpleManual"), sleuthSimpleOnEach("spring.sleuth.reactor.instrumentation-type", + "DECORATE_ON_EACH", + "/simple"), sleuthSimpleOnLast("spring.sleuth.reactor.instrumentation-type", + "DECORATE_ON_LAST", "/simple"), noSleuthComplex("spring.sleuth.enabled", + "false", "/complexNoSleuth"), onEachComplex( "spring.sleuth.reactor.instrumentation-type", - "DECORATE_ON_LAST", "/complex"), onManualComplex( + "DECORATE_ON_EACH", "/complex"), onLastComplex( "spring.sleuth.reactor.instrumentation-type", - "MANUAL", "/complexManual"); + "DECORATE_ON_LAST", "/complex"), onManualComplex( + "spring.sleuth.reactor.instrumentation-type", + "MANUAL", "/complexManual"); private String key; diff --git a/docs/src/main/asciidoc/spring-cloud-sleuth.adoc b/docs/src/main/asciidoc/spring-cloud-sleuth.adoc index 2f816ffbd..64aa235bf 100644 --- a/docs/src/main/asciidoc/spring-cloud-sleuth.adoc +++ b/docs/src/main/asciidoc/spring-cloud-sleuth.adoc @@ -708,6 +708,8 @@ You can configure which URIs you would like to skip by using the `spring.sleuth. If you have `ManagementServerProperties` on the classpath, its value of `contextPath` gets appended to the provided skip pattern. If you want to reuse Sleuth's default skip patterns and append your own, pass those patterns by using the `spring.sleuth.web.additionalSkipPattern`. +In order to achieve best results in terms of performance and context propagation we suggest that you switch the `spring.sleuth.reactor.instrumentation-type` to `MANUAL`. In order to execute code with the span in scope you can call `WebFluxSleuthOperators.withSpanInScope`. + To change the order of tracing filter registration, please set the `spring.sleuth.web.filter-order` property. @@ -929,7 +931,7 @@ to add the `@Role(BeanDefinition.ROLE_INFRASTRUCTURE)` on your Features from this section can be disabled by setting the `spring.sleuth.messaging.enabled` property with value equal to `false`. -==== Spring Integration and Spring Cloud Stream +==== Spring Integration Spring Cloud Sleuth integrates with https://projects.spring.io/spring-integration/[Spring Integration]. It creates spans for publish and subscribe events. @@ -947,6 +949,21 @@ it's enough for you to register beans of types: * `Propagation.Setter` - for writing headers to the message * `Propagation.Getter` - for reading headers from the message +==== Spring Cloud Function and Spring Cloud Stream + +Spring Cloud Sleuth can instrument Spring Cloud Function. The way to achieve it is to provide a `Function` or `Consumer` or `Supplier` that takes in a `Message` as a parameter e.g. `Function, Message>`. If the type is not `Message` then instrumentation will not take place. Out of the box instrumentation will not take place when dealing with Reactor based streams - e.g. `Function>, Flux>>`. + +Since Spring Cloud Stream reuses Spring Cloud Function, you'll get the instrumentation out of the box. + +You can disable this behavior by setting the value of `spring.sleuth.function.enabled` to `false`. + +In order to work with reactive Stream functions you can leverage the `MessagingSleuthOperators` utility class that allows you to manipulate the input and output messages in order to continue the tracing context and to execute custom code within the tracing context. + +[source,java] +----- +include::{project-root}/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java[tags=simple_reactive,indent=0] +----- + ==== Spring RabbitMq We instrument the `RabbitTemplate` so that tracing headers get injected @@ -1000,7 +1017,19 @@ To turn off this feature, set the `spring.sleuth.quartz.enabled` property to `fa === Project Reactor -For projects depending on Project Reactor such as Spring Cloud Gateway, we suggest turning the `spring.sleuth.reactor.decorate-on-each` option to `false`. That way an increased performance gain should be observed in comparison to the standard instrumentation mechanism. What this option does is it will wrap decorate `onLast` operator instead of `onEach` which will result in creation of far fewer objects. The downside of this is that when Project Reactor will change threads, the trace propagation will continue without issues, however anything relying on the `ThreadLocal` such as e.g. MDC entries can be buggy. +We have three modes of instrumenting reactor based applications that can +be set via `spring.sleuth.reactor.instrumentation-type` property: + +* `ON_EACH` - wraps every Reactor operator in a trace representation. Passes the tracing context in most cases. This mode might lead to drastic performance degradation. +* `ON_LAST` - wraps last Reactor operator in a trace representation. Passes the tracing context in some cases thus accessing MDC context might not work. This mode might lead to medium performance degradation. +* `MANUAL` - wraps every Reactor in the least invasive way without passing of tracing context. It's up to the user to do it. + +Current default is `ON_EACH` for backward compatibility reasons, however we encourage the users to migrate to the `MANUAL` instrumentation and profit from `WebFluxSleuthOperators` and `MessagingSleuthOperators`. The performance improvement can be substantial. Example: + +[source,java] +----- +include::{project-root}/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/webflux/SleuthBenchmarkingSpringWebFluxApp.java[tags=simple_manual,indent=0] +----- == Log integration Sleuth configures the logging context with variables including the service name diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/async/AsyncAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/async/AsyncAutoConfiguration.java index fa6f419e8..8cf6dd3f9 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/async/AsyncAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/async/AsyncAutoConfiguration.java @@ -16,6 +16,9 @@ package org.springframework.cloud.sleuth.instrument.async; +import brave.Tracing; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; @@ -29,6 +32,7 @@ import org.springframework.context.annotation.Configuration; * @since 2.1.0 */ @Configuration(proxyBeanMethods = false) +@ConditionalOnBean(Tracing.class) @ConditionalOnProperty(value = "spring.sleuth.scheduled.enabled", matchIfMissing = true) @EnableConfigurationProperties(SleuthAsyncProperties.class) class AsyncAutoConfiguration { diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/MessagingSleuthOperator.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/MessagingSleuthOperators.java similarity index 97% rename from spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/MessagingSleuthOperator.java rename to spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/MessagingSleuthOperators.java index 0f4aad931..dccb9eb1a 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/MessagingSleuthOperator.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/MessagingSleuthOperators.java @@ -32,6 +32,8 @@ import org.springframework.messaging.Message; * Messaging helpers to manually parse and inject spans. We're treating message headers as * a context that gets passed through. * + * IMPORTANT: This API is experimental and might change in the future. + * * The {@code forInputMessage} factory methods will retrieve the tracing context from the * message headers and set up a a child span in the header under key * {@link Span#getClass()} name. If you need to continue it or tag it, it's enough to @@ -40,16 +42,14 @@ import org.springframework.messaging.Message; * The first messaging span (the one that was first found in the input message) is present * under the {@code traceHandlerParentSpan} header key. * - * When calling the {@code toOutputMessage} factory method, we will - * * @author Marcin Grzejszczak * @since 3.0.0 */ -public final class MessagingSleuthOperator { +public final class MessagingSleuthOperators { - private static final Log log = LogFactory.getLog(MessagingSleuthOperator.class); + private static final Log log = LogFactory.getLog(MessagingSleuthOperators.class); - private MessagingSleuthOperator() { + private MessagingSleuthOperators() { throw new IllegalStateException("You can't instantiate a utility class"); } @@ -114,7 +114,7 @@ public final class MessagingSleuthOperator { */ public static Function, Message> asFunction(Tracing tracing, Message inputMessage) { - return stringMessage -> MessagingSleuthOperator.forInputMessage(tracing, + return stringMessage -> MessagingSleuthOperators.forInputMessage(tracing, inputMessage); } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceFunctionAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceFunctionAutoConfiguration.java index 73b687865..a136088b1 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceFunctionAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceFunctionAutoConfiguration.java @@ -130,7 +130,7 @@ class TraceFunctionAroundWrapper extends FunctionAroundWrapper private String inputDestination( SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction) { String functionDefinition = targetFunction.getFunctionDefinition(); - return functionToDestinationCache + return this.functionToDestinationCache .computeIfAbsent(functionDefinition, s -> this.environment.getProperty( "spring.cloud.stream.bindings." + s + "-in-0.destination", @@ -150,7 +150,7 @@ class TraceFunctionAroundWrapper extends FunctionAroundWrapper if (log.isDebugEnabled()) { log.debug("Context refreshed, will reset the cache"); } - functionToDestinationCache.clear(); + this.functionToDestinationCache.clear(); } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SleuthReactorProperties.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SleuthReactorProperties.java index b9212065b..0f3887310 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SleuthReactorProperties.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SleuthReactorProperties.java @@ -16,6 +16,9 @@ package org.springframework.cloud.sleuth.instrument.reactor; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.boot.context.properties.ConfigurationProperties; /** @@ -25,7 +28,9 @@ import org.springframework.boot.context.properties.ConfigurationProperties; * @since 2.0.2 */ @ConfigurationProperties("spring.sleuth.reactor") -class SleuthReactorProperties { +public class SleuthReactorProperties { + + private static final Log log = LogFactory.getLog(SleuthReactorProperties.class); /** * When true enables instrumentation for reactor. @@ -55,15 +60,22 @@ class SleuthReactorProperties { @Deprecated public boolean isDecorateOnEach() { - return this.decorateOnEach; + warn(); + return this.instrumentationType == InstrumentationType.DECORATE_ON_EACH; } @Deprecated public void setDecorateOnEach(boolean decorateOnEach) { + warn(); this.instrumentationType = decorateOnEach ? InstrumentationType.DECORATE_ON_EACH : InstrumentationType.DECORATE_ON_LAST; } + private void warn() { + log.warn( + "You're using the deprecated [spring.sleuth.reactor.decorate-on-each] property. Please use the [spring.sleuth.reactor.instrumentation-type]"); + } + public InstrumentationType getInstrumentationType() { return this.instrumentationType; } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFilter.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFilter.java index cbe11de6f..635546dce 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFilter.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFilter.java @@ -26,6 +26,7 @@ import brave.http.HttpServerRequest; import brave.http.HttpServerResponse; import brave.http.HttpTracing; import brave.propagation.TraceContext; +import brave.propagation.TraceContextOrSamplingFlags; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.reactivestreams.Subscription; @@ -36,6 +37,7 @@ import reactor.util.annotation.Nullable; import reactor.util.context.Context; import org.springframework.beans.factory.BeanFactory; +import org.springframework.cloud.sleuth.instrument.reactor.SleuthReactorProperties; import org.springframework.core.Ordered; import org.springframework.http.HttpStatus; import org.springframework.http.server.reactive.ServerHttpRequest; @@ -62,8 +64,9 @@ final class TraceWebFilter implements WebFilter, Ordered { */ public static final int ORDER = TraceHttpAutoConfiguration.TRACING_FILTER_ORDER; - protected static final String TRACE_REQUEST_ATTR = TraceWebFilter.class.getName() - + ".TRACE"; + // Remember that this can be used in other packages + protected static final String TRACE_REQUEST_ATTR = TraceContext.class.getName(); + static final String MVC_CONTROLLER_CLASS_KEY = "mvc.controller.class"; static final String MVC_CONTROLLER_METHOD_KEY = "mvc.controller.method"; @@ -82,6 +85,8 @@ final class TraceWebFilter implements WebFilter, Ordered { SleuthWebProperties webProperties; + SleuthReactorProperties sleuthReactorProperties; + TraceWebFilter(BeanFactory beanFactory) { this.beanFactory = beanFactory; } @@ -113,6 +118,14 @@ final class TraceWebFilter implements WebFilter, Ordered { return this.webProperties; } + SleuthReactorProperties sleuthReactorProperties() { + if (this.sleuthReactorProperties == null) { + this.sleuthReactorProperties = this.beanFactory + .getBean(SleuthReactorProperties.class); + } + return this.sleuthReactorProperties; + } + @Override public Mono filter(ServerWebExchange exchange, WebFilterChain chain) { String uri = exchange.getRequest().getPath().pathWithinApplication().value(); @@ -120,14 +133,21 @@ final class TraceWebFilter implements WebFilter, Ordered { log.debug("Received a request to uri [" + uri + "]"); } Mono source = chain.filter(exchange); + boolean tracePresent = isTracePresent(); + return new MonoWebFilterTrace(source, exchange, tracePresent, this); + } + + private boolean isTracePresent() { + if (sleuthReactorProperties() + .getInstrumentationType() == SleuthReactorProperties.InstrumentationType.MANUAL) { + return false; + } boolean tracePresent = tracer().currentSpan() != null; - // if we're in manual instrumentation type mode then we control how threads are - // set if (tracePresent) { // clear any previous trace tracer().withSpanInScope(null); // TODO: dangerous and also allocates stuff } - return new MonoWebFilterTrace(source, exchange, tracePresent, this); + return tracePresent; } @Override @@ -141,7 +161,7 @@ final class TraceWebFilter implements WebFilter, Ordered { final Tracer tracer; - final Span attrSpan; + final TraceContext traceContext; final HttpServerHandler handler; @@ -155,7 +175,7 @@ final class TraceWebFilter implements WebFilter, Ordered { this.tracer = parent.tracer(); this.handler = parent.handler(); this.exchange = exchange; - this.attrSpan = exchange.getAttribute(TRACE_REQUEST_ATTR); + this.traceContext = exchange.getAttribute(TRACE_REQUEST_ATTR); this.initialTracePresent = initialTracePresent; } @@ -184,8 +204,9 @@ final class TraceWebFilter implements WebFilter, Ordered { } } else { - if (this.attrSpan != null) { - span = this.attrSpan; + if (this.traceContext != null) { + span = this.tracer.nextSpan( + TraceContextOrSamplingFlags.create(this.traceContext)); if (log.isDebugEnabled()) { log.debug("Found span in attribute " + span); } @@ -197,7 +218,7 @@ final class TraceWebFilter implements WebFilter, Ordered { log.debug("Handled receive of span " + span); } } - this.exchange.getAttributes().put(TRACE_REQUEST_ATTR, span); + this.exchange.getAttributes().put(TRACE_REQUEST_ATTR, span.context()); } return span; } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/WebFluxSleuthOperators.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/WebFluxSleuthOperators.java index 7da181f55..136296799 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/WebFluxSleuthOperators.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/WebFluxSleuthOperators.java @@ -28,8 +28,11 @@ import reactor.core.publisher.Signal; import reactor.core.publisher.SignalType; import reactor.util.context.Context; +import org.springframework.web.server.ServerWebExchange; + /** * WebFlux operators that are capable to reuse tracing context from Reactor's Context. + * IMPORTANT: This API is experimental and might change in the future. * * @author Marcin Grzejszczak * @since 3.0.0 @@ -58,6 +61,22 @@ public final class WebFluxSleuthOperators { }; } + /** + * Wraps a runnable with a span. + * @param signalType - Reactor's signal type + * @param consumer - lambda to execute within the tracing context + * @return consumer of a signal + */ + public static Consumer withSpanInScope(SignalType signalType, + Consumer consumer) { + return signal -> { + if (signalType != signal.getType()) { + return; + } + withSpanInScope(signal.getContext(), () -> consumer.accept(signal)); + }; + } + /** * Wraps a runnable with a span. * @param runnable - lambda to execute within the tracing context @@ -84,6 +103,19 @@ public final class WebFluxSleuthOperators { } } + /** + * Wraps a callable with a span. + * @param context - Reactor context that contains the {@link TraceContext} + * @param callable - lambda to execute within the tracing context + * @param callable's return type + * @return value from the callable + */ + public static T withSpanInScope(Context context, Callable callable) { + CurrentTraceContext currentTraceContext = context.get(CurrentTraceContext.class); + TraceContext traceContext = traceContextOrNew(context); + return withContext(callable, currentTraceContext, traceContext); + } + private static TraceContext traceContextOrNew(Context context) { Tracing tracing = context.get(Tracing.class); if (!context.hasKey(TraceContext.class)) { @@ -97,14 +129,68 @@ public final class WebFluxSleuthOperators { /** * Wraps a runnable with a span. - * @param context - Reactor context that contains the {@link TraceContext} + * @param tracing - tracing bean + * @param exchange - server web exchange that can contain the {@link TraceContext} in + * its attribute + * @param runnable - lambda to execute within the tracing context + */ + public static void withSpanInScope(Tracing tracing, ServerWebExchange exchange, + Runnable runnable) { + CurrentTraceContext currentTraceContext = tracing.currentTraceContext(); + TraceContext traceContext = traceContextFromExchangeOrNew(tracing, exchange); + try (CurrentTraceContext.Scope scope = currentTraceContext + .maybeScope(traceContext)) { + runnable.run(); + } + } + + /** + * Wraps a callable with a span. + * @param tracing - tracing bean + * @param exchange - server web exchange that can contain the {@link TraceContext} in + * its attribute * @param callable - lambda to execute within the tracing context * @param callable's return type * @return value from the callable */ - public static T withSpanInScope(Context context, Callable callable) { - CurrentTraceContext currentTraceContext = context.get(CurrentTraceContext.class); - TraceContext traceContext = traceContextOrNew(context); + public static T withSpanInScope(Tracing tracing, ServerWebExchange exchange, + Callable callable) { + CurrentTraceContext currentTraceContext = tracing.currentTraceContext(); + TraceContext traceContext = traceContextFromExchangeOrNew(tracing, exchange); + return withContext(callable, currentTraceContext, traceContext); + } + + /** + * Returns the current trace context. + * @param exchange - server web exchange that can contain the {@link TraceContext} in + * its attribute + * @return current trace context or {@code null} if it's not present + */ + public static TraceContext currentTraceContext(ServerWebExchange exchange) { + return exchange.getAttribute(TraceContext.class.getName()); + } + + /** + * Returns the current trace context. + * @param context - Reactor context that can contain the {@link TraceContext} + * @return current trace context or {@code null} if it's not present + */ + public static TraceContext currentTraceContext(Context context) { + return context.getOrDefault(TraceContext.class, null); + } + + /** + * Returns the current trace context. + * @param signal - Reactor signal that can contain the {@link TraceContext} in its + * context + * @return current trace context or {@code null} if it's not present + */ + public static TraceContext currentTraceContext(Signal signal) { + return currentTraceContext(signal.getContext()); + } + + private static T withContext(Callable callable, + CurrentTraceContext currentTraceContext, TraceContext traceContext) { try (CurrentTraceContext.Scope scope = currentTraceContext .maybeScope(traceContext)) { try { @@ -116,4 +202,16 @@ public final class WebFluxSleuthOperators { } } + private static TraceContext traceContextFromExchangeOrNew(Tracing tracing, + ServerWebExchange exchange) { + TraceContext traceContext = exchange.getAttribute(TraceContext.class.getName()); + if (traceContext == null) { + if (log.isDebugEnabled()) { + log.debug("No trace context found, will create a new span"); + } + traceContext = tracing.tracer().nextSpan().context(); + } + return traceContext; + } + } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceRequestHttpHeadersFilter.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceRequestHttpHeadersFilter.java index 11c33b8e7..83abb19c4 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceRequestHttpHeadersFilter.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceRequestHttpHeadersFilter.java @@ -23,6 +23,7 @@ import brave.Span; import brave.Tracer; import brave.http.HttpClientHandler; import brave.http.HttpTracing; +import brave.propagation.TraceContext; import brave.propagation.TraceContext.Extractor; import brave.propagation.TraceContextOrSamplingFlags; import org.apache.commons.logging.Log; @@ -38,6 +39,8 @@ final class TraceRequestHttpHeadersFilter extends AbstractHttpHeadersFilter { private static final Log log = LogFactory.getLog(TraceRequestHttpHeadersFilter.class); + static final String TRACE_REQUEST_ATTR = TraceContext.class.getName(); + private TraceRequestHttpHeadersFilter(HttpTracing httpTracing) { super(httpTracing); } @@ -53,7 +56,7 @@ final class TraceRequestHttpHeadersFilter extends AbstractHttpHeadersFilter { + exchange.getRequest().getHeaders() + "]"); } HttpClientRequest request = new HttpClientRequest(exchange.getRequest(), input); - Span currentSpan = currentSpan(request); + Span currentSpan = currentSpan(exchange, request); Span span = injectedSpan(request, currentSpan); if (log.isDebugEnabled()) { log.debug( @@ -71,8 +74,8 @@ final class TraceRequestHttpHeadersFilter extends AbstractHttpHeadersFilter { return headersWithInput; } - private Span currentSpan(HttpClientRequest request) { - Span currentSpan = this.tracer.currentSpan(); + private Span currentSpan(ServerWebExchange exchange, HttpClientRequest request) { + Span currentSpan = currentSpan(exchange); if (currentSpan != null) { return currentSpan; } @@ -83,6 +86,18 @@ final class TraceRequestHttpHeadersFilter extends AbstractHttpHeadersFilter { return this.tracer.nextSpan(contextOrFlags); } + private Span currentSpan(ServerWebExchange exchange) { + Object attribute = exchange.getAttribute(TRACE_REQUEST_ATTR); + if (attribute instanceof Span) { + if (log.isDebugEnabled()) { + log.debug("Found trace request attribute in the server web exchange [" + + attribute + "]"); + } + return (Span) attribute; + } + return this.tracer.currentSpan(); + } + private Span injectedSpan(HttpClientRequest request, Span currentSpan) { if (currentSpan == null) { return this.handler.handleSend(request); diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java index 477f51d37..c537ba7be 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java @@ -161,37 +161,33 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { final CurrentTraceContext currentTraceContext; - final Function, ? extends Publisher> scopePassingTransformer; - - @Nullable - final TraceContext parent; - MonoWebClientTrace(ExchangeFunction next, ClientRequest request, TraceExchangeFilterFunction filterFunction) { this.next = next; this.request = request; this.handler = filterFunction.handler(); this.currentTraceContext = filterFunction.currentTraceContext(); - this.scopePassingTransformer = filterFunction.scopePassingTransformer; - this.parent = currentTraceContext.get(); } @Override public void subscribe(CoreSubscriber subscriber) { - Context context = subscriber.currentContext(); - + if (log.isTraceEnabled()) { + log.trace("Got the following context [" + context + "]"); + } ClientRequestWrapper wrapper = new ClientRequestWrapper(request); + TraceContext parent = context.hasKey(TraceContext.class) + ? context.get(TraceContext.class) : null; Span span = handler.handleSendWithParent(wrapper, parent); if (log.isDebugEnabled()) { log.debug("HttpClientHandler::handleSend: " + span); } - // NOTE: We are starting the client span for the request here, but it could be // canceled prior to actually being invoked. TraceWebClientSubscription will // abandon this span, if cancel() happens before request(). - this.next.exchange(wrapper.buildRequest()).subscribe( - new TraceWebClientSubscriber(subscriber, context, span, this)); + this.next.exchange(wrapper.buildRequest()) + .subscribe(new TraceWebClientSubscriber(subscriber, context, span, + parent, this)); } } @@ -208,20 +204,18 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { final HttpClientHandler handler; - final Function, ? extends Publisher> scopePassingTransformer; - final CurrentTraceContext currentTraceContext; TraceWebClientSubscriber(CoreSubscriber actual, - Context ctx, Span clientSpan, MonoWebClientTrace mono) { + Context ctx, Span clientSpan, TraceContext parent, + MonoWebClientTrace mono) { this.actual = actual; - this.parent = mono.parent; + this.parent = parent; this.handler = mono.handler; this.currentTraceContext = mono.currentTraceContext; - this.scopePassingTransformer = mono.scopePassingTransformer; - this.context = parent != null - && !parent.equals(ctx.getOrDefault(TraceContext.class, null)) - ? ctx.put(TraceContext.class, parent) : ctx; + this.context = this.parent != null + && !this.parent.equals(ctx.getOrDefault(TraceContext.class, null)) + ? ctx.put(TraceContext.class, this.parent) : ctx; set(clientSpan); } @@ -234,11 +228,7 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { public void onNext(ClientResponse response) { try (Scope scope = currentTraceContext.maybeScope(parent)) { // decorate response body - this.actual.onNext(ClientResponse.from(response) - // TODO: Why are we using scope passing transformer - .body(response.bodyToFlux(DataBuffer.class) - .transform(this.scopePassingTransformer)) - .build()); + this.actual.onNext(response); } finally { Span span = getAndSet(null); diff --git a/tests/spring-cloud-sleuth-instrumentation-messaging-tests/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/StreamMessageOperatorsTests.java b/tests/spring-cloud-sleuth-instrumentation-messaging-tests/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/StreamMessageOperatorsTests.java index f7ab2f87d..30024d2e2 100644 --- a/tests/spring-cloud-sleuth-instrumentation-messaging-tests/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/StreamMessageOperatorsTests.java +++ b/tests/spring-cloud-sleuth-instrumentation-messaging-tests/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/StreamMessageOperatorsTests.java @@ -112,17 +112,17 @@ class SimpleReactiveManualFunction @Override public Flux> apply(Flux> input) { return input.map( - message -> (MessagingSleuthOperator.asFunction(this.tracing, message)) - .andThen(msg -> MessagingSleuthOperator + message -> (MessagingSleuthOperators.asFunction(this.tracing, message)) + .andThen(msg -> MessagingSleuthOperators .withSpanInScope(this.tracing, msg, stringMessage -> { log.info("Hello from simple manual [{}]", stringMessage.getPayload()); return stringMessage; })) - .andThen(msg -> MessagingSleuthOperator + .andThen(msg -> MessagingSleuthOperators .afterMessageHandled(this.tracing, msg, null)) .andThen(msg -> { - MessagingSleuthOperator.withSpanInScope(this.tracing, msg, + MessagingSleuthOperators.withSpanInScope(this.tracing, msg, stringMessage -> { log.info("Here we may do some processing"); }); @@ -131,7 +131,7 @@ class SimpleReactiveManualFunction return MessageBuilder.createMessage( msg.getPayload().toUpperCase(), new MessageHeaders(headers)); - }).andThen(msg -> MessagingSleuthOperator + }).andThen(msg -> MessagingSleuthOperators .handleOutputMessage(this.tracing, msg)) .apply(message)); } diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/FlatMapTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/FlatMapTests.java index 324f09078..6bd84cb6a 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/FlatMapTests.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/FlatMapTests.java @@ -18,9 +18,12 @@ package org.springframework.cloud.sleuth.instrument.reactor.sample; import java.util.Arrays; import java.util.List; +import java.util.function.Supplier; import java.util.stream.Collectors; +import brave.Span; import brave.Tracer; +import brave.Tracing; import brave.handler.MutableSpan; import brave.handler.SpanHandler; import brave.sampler.Sampler; @@ -40,9 +43,9 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.boot.test.system.CapturedOutput; import org.springframework.boot.test.system.OutputCaptureExtension; -import org.springframework.cloud.context.refresh.ContextRefresher; import org.springframework.cloud.sleuth.instrument.reactor.Issue866Configuration; import org.springframework.cloud.sleuth.instrument.reactor.TraceReactorAutoConfigurationAccessorConfiguration; +import org.springframework.cloud.sleuth.instrument.web.WebFluxSleuthOperators; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -51,6 +54,7 @@ import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.reactive.function.server.RouterFunction; import org.springframework.web.reactive.function.server.ServerResponse; +import org.springframework.web.server.ServerWebExchange; import static org.assertj.core.api.BDDAssertions.then; import static org.springframework.web.reactive.function.server.RequestPredicates.GET; @@ -84,7 +88,8 @@ public class FlatMapTests { "security.basic.enabled=false", "management.security.enabled=false") .run(); - assertReactorTracing(context, capture); + assertReactorTracing(context, capture, + () -> context.getBean(TestConfiguration.class).spanInFoo); } @Test @@ -100,26 +105,33 @@ public class FlatMapTests { "security.basic.enabled=false", "management.security.enabled=false") .run(); - assertReactorTracing(context, capture); + assertReactorTracing(context, capture, + () -> context.getBean(TestConfiguration.class).spanInFoo); + } - try { - System.setProperty("spring.sleuth.reactor.decorate-on-each", "true"); - // trigger context refreshed - context.getBean(ContextRefresher.class).refresh(); - assertReactorTracing(context, capture); - } - finally { - System.clearProperty("spring.sleuth.reactor.decorate-on-each"); - } + @Test + public void should_work_with_flat_maps_with_on_manual_operator_instrumentation( + CapturedOutput capture) { + // given + ConfigurableApplicationContext context = new SpringApplicationBuilder( + FlatMapTests.TestManualConfiguration.class, Issue866Configuration.class) + .web(WebApplicationType.REACTIVE) + .properties("server.port=0", "spring.jmx.enabled=false", + "spring.sleuth.reactor.instrumentation-type=MANUAL", + "spring.application.name=TraceWebFlux3Tests", + "security.basic.enabled=false", + "management.security.enabled=false") + .run(); + assertReactorTracing(context, capture, + () -> context.getBean(TestManualConfiguration.class).spanInFoo); } private void assertReactorTracing(ConfigurableApplicationContext context, - CapturedOutput capture) { + CapturedOutput capture, SpanProvider spanProvider) { TestSpanHandler spans = context.getBean(TestSpanHandler.class); int port = context.getBean(Environment.class).getProperty("local.server.port", Integer.class); RequestSender sender = context.getBean(RequestSender.class); - TestConfiguration config = context.getBean(TestConfiguration.class); FactoryUser factoryUser = context.getBean(FactoryUser.class); sender.port = port; spans.clear(); @@ -132,7 +144,7 @@ public class FlatMapTests { // then LOGGER.info("Checking first trace id"); thenAllWebClientCallsHaveSameTraceId(firstTraceId, sender); - thenSpanInFooHasSameTraceId(firstTraceId, config); + thenSpanInFooHasSameTraceId(firstTraceId, spanProvider); spans.clear(); LOGGER.info("All web client calls have same trace id"); @@ -143,7 +155,7 @@ public class FlatMapTests { then(firstTraceId).as("Id will not be reused between calls") .isNotEqualTo(secondTraceId); LOGGER.info("Id was not reused between calls"); - thenSpanInFooHasSameTraceId(secondTraceId, config); + thenSpanInFooHasSameTraceId(secondTraceId, spanProvider); LOGGER.info("Span in Foo has same trace id"); // and List requestUri = Arrays.stream(capture.toString().split("\n")) @@ -166,8 +178,8 @@ public class FlatMapTests { then(sender.span.context().traceIdString()).isEqualTo(traceId); } - private void thenSpanInFooHasSameTraceId(String traceId, TestConfiguration config) { - then(config.spanInFoo.context().traceIdString()).isEqualTo(traceId); + private void thenSpanInFooHasSameTraceId(String traceId, SpanProvider spanProvider) { + then(spanProvider.get().context().traceIdString()).isEqualTo(traceId); } private Mono callFlatMap(int port) { @@ -246,6 +258,75 @@ public class FlatMapTests { } + @Configuration + @EnableAutoConfiguration + static class TestManualConfiguration { + + brave.Span spanInFoo; + + @Bean + RouterFunction handlers(Tracing tracing, + ManualRequestSender requestSender) { + return route(GET("/noFlatMap"), request -> { + ServerWebExchange exchange = request.exchange(); + WebFluxSleuthOperators.withSpanInScope(tracing, exchange, + () -> LOGGER.info("noFlatMap")); + Flux one = requestSender.getAll().map(String::length); + return ServerResponse.ok().body(one, Integer.class); + }).andRoute(GET("/withFlatMap"), request -> { + ServerWebExchange exchange = request.exchange(); + WebFluxSleuthOperators.withSpanInScope(tracing, exchange, + () -> LOGGER.info("withFlatMap")); + Flux one = requestSender.getAll().map(String::length); + Flux response = one + .flatMap(size -> requestSender.getAll() + .doOnEach(sig -> WebFluxSleuthOperators.withSpanInScope( + sig.getContext(), + () -> LOGGER.info(sig.getContext().toString())))) + .map(string -> { + WebFluxSleuthOperators.withSpanInScope(tracing, exchange, + () -> LOGGER.info("WHATEVER YEAH")); + return string.length(); + }); + return ServerResponse.ok().body(response, Integer.class); + }).andRoute(GET("/foo"), request -> { + ServerWebExchange exchange = request.exchange(); + WebFluxSleuthOperators.withSpanInScope(tracing, exchange, () -> { + LOGGER.info("foo"); + this.spanInFoo = tracing.tracer().currentSpan(); + }); + return ServerResponse.ok().body(Flux.just(1), Integer.class); + }); + } + + @Bean + WebClient webClient() { + return WebClient.create(); + } + + @Bean + SpanHandler testSpanHandler() { + return new TestSpanHandler(); + } + + @Bean + Sampler sampler() { + return Sampler.ALWAYS_SAMPLE; + } + + @Bean + ManualRequestSender sender(WebClient client, Tracer tracer) { + return new ManualRequestSender(client, tracer); + } + + // https://github.com/spring-cloud/spring-cloud-sleuth/issues/866 + @Bean + FactoryUser factoryUser() { + return new FactoryUser(); + } + + } + } class FactoryUser { @@ -258,3 +339,7 @@ class FactoryUser { } } + +interface SpanProvider extends Supplier { + +} diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/ManualRequestSender.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/ManualRequestSender.java new file mode 100644 index 000000000..9500eab7b --- /dev/null +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/ManualRequestSender.java @@ -0,0 +1,62 @@ +/* + * Copyright 2013-2020 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 + * + * https://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.reactor.sample; + +import brave.Tracer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.SignalType; + +import org.springframework.cloud.sleuth.instrument.web.WebFluxSleuthOperators; +import org.springframework.http.HttpMethod; +import org.springframework.web.reactive.function.client.WebClient; + +class ManualRequestSender extends RequestSender { + + private static final Logger LOGGER = LoggerFactory + .getLogger(ManualRequestSender.class); + + ManualRequestSender(WebClient webClient, Tracer tracer) { + super(webClient, tracer); + } + + @Override + public Mono get(Integer someParameterNotUsedNow) { + return Mono.just(this.webClient).doOnEach( + WebFluxSleuthOperators.withSpanInScope(SignalType.ON_NEXT, () -> { + this.span = this.tracer.currentSpan(); + LOGGER.info("getting for parameter {}", someParameterNotUsedNow); + })) + .flatMap(webClient -> Mono.subscriberContext() + .flatMap(ctx -> WebFluxSleuthOperators.withSpanInScope(ctx, + () -> webClient.method(HttpMethod.GET) + .uri("http://localhost:" + port + "/foo") + .retrieve().bodyToMono(String.class)))); + } + + @Override + public Flux getAll() { + return Flux.just("").flatMap(s -> Flux.deferWithContext(ctx -> Flux.just("") + .doOnNext(t -> WebFluxSleuthOperators.withSpanInScope(ctx, + () -> LOGGER.info("before merge"))) + .mergeWith(get(2)).mergeWith(get(3)).doOnNext(t -> WebFluxSleuthOperators + .withSpanInScope(ctx, () -> LOGGER.info("after merge"))))); + } + +} diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/RequestSender.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/RequestSender.java index 10481cb15..582299db0 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/RequestSender.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/RequestSender.java @@ -30,9 +30,9 @@ class RequestSender { private static final Logger LOGGER = LoggerFactory.getLogger(RequestSender.class); - private final WebClient webClient; + final WebClient webClient; - private final Tracer tracer; + final Tracer tracer; int port;