Improvement with Reactor instrumentation

- adds Spring Cloud Function instrumentation
- adds Operators to manually provide instrumentation for Fluxes
- introduces Manual instrumentation mode for Reactor

TODO: Documentation (will add it soon)

related gh-1684

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/mvc/SleuthBenchmarkingSpringApp.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * 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.
@@ -56,13 +56,15 @@ import org.springframework.web.bind.annotation.RestController;
 @SpringBootApplication
 @RestController
 @EnableAsync
-public class SleuthBenchmarkingSpringApp
-		implements ApplicationListener<ServletWebServerInitializedEvent> {
+public class SleuthBenchmarkingSpringApp implements ApplicationListener<ServletWebServerInitializedEvent> {
 
 	private static final Log log = LogFactory.getLog(SleuthBenchmarkingSpringApp.class);
 
 	public final ExecutorService pool = Executors.newWorkStealingPool();
 
+	/**
+	 * Port of the app.
+	 */
 	public int port;
 
 	@Autowired(required = false)
@@ -109,11 +111,9 @@ public class SleuthBenchmarkingSpringApp
 	}
 
 	@Bean
-	public ServletWebServerFactory servletContainer(
-			@Value("${server.port:0}") int serverPort) {
+	public ServletWebServerFactory servletContainer(@Value("${server.port:0}") int serverPort) {
 		log.info("Starting container at port [" + serverPort + "]");
-		return new TomcatServletWebServerFactory(
-				serverPort == 0 ? SocketUtils.findAvailableTcpPort() : serverPort);
+		return new TomcatServletWebServerFactory(serverPort == 0 ? SocketUtils.findAvailableTcpPort() : serverPort);
 	}
 
 	@PreDestroy

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java

@@ -0,0 +1,281 @@
+/*
+ * 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.benchmarks.app.stream;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+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;
+import reactor.core.publisher.Mono;
+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.stream.binder.test.InputDestination;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.context.ConfigurableApplicationContext;
+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;
+
+@SpringBootApplication
+@Import(TestChannelBinderConfiguration.class)
+public class SleuthBenchmarkingStreamApplication {
+
+	public static void main(String[] args) throws InterruptedException, IOException {
+		// System.setProperty("spring.sleuth.enabled", "false");
+		// System.setProperty("spring.sleuth.reactor.instrumentation-type",
+		// "DECORATE_ON_EACH");
+		// System.setProperty("spring.sleuth.reactor.instrumentation-type",
+		// "DECORATE_ON_LAST");
+		// System.setProperty("spring.sleuth.reactor.instrumentation-type", "MANUAL");
+		System.setProperty("spring.sleuth.reactor.instrumentation-type", "MANUAL");
+		System.setProperty("spring.sleuth.function.type", "simple");
+		ConfigurableApplicationContext context = SpringApplication.run(SleuthBenchmarkingStreamApplication.class, args);
+		System.out.println("PRess any key to continue");
+		System.in.read();
+		for (int i = 0; i < 1; i++) {
+			InputDestination input = context.getBean(InputDestination.class);
+			input.send(MessageBuilder.withPayload("hello".getBytes())
+					.setHeader("b3", "4883117762eb9420-4883117762eb9420-1").build());
+			System.out.println("Retrieving the message for tests");
+			OutputDestination output = context.getBean(OutputDestination.class);
+			Message<byte[]> message = output.receive(200L);
+			System.out.println("Got the message from output");
+			assertThat(message).isNotNull();
+			System.out.println("Message is not null");
+			assertThat(message.getPayload()).isEqualTo("HELLO".getBytes());
+			System.out.println("Payload is HELLO");
+			String b3 = message.getHeaders().get("b3", String.class);
+			System.out.println("Checking the b3 header [" + b3 + "]");
+			assertThat(b3).startsWith("4883117762eb9420");
+		}
+	}
+
+	@Bean
+	ExecutorService sleuthExecutorService() {
+		return Executors.newCachedThreadPool();
+	}
+
+	@Bean(name = "myFlux")
+	@ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "simple")
+	public Function<String, String> nonReactiveSimpleSleuthFunction() {
+		System.out.println("simple_function");
+		return new SimpleFunction();
+	}
+
+	@Bean(name = "myFlux")
+	@ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "reactive_simple")
+	public Function<Flux<String>, Flux<String>> reactiveSimpleSleuthFunction() {
+		System.out.println("simple_reactive_function");
+		return new SimpleReactiveFunction();
+	}
+
+	@Bean(name = "myFlux")
+	@ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "simple_manual")
+	public Function<Message<String>, Message<String>> nonReactiveSimpleManualSleuthFunction(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<Message<String>>, Flux<Message<String>>> reactiveSimpleManualSleuthFunction(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<String, String> nonReactiveSleuthFunction(ExecutorService executorService) {
+		System.out.println("no sleuth non reactive function");
+		return new SleuthNonReactiveFunction(executorService);
+	}
+
+	@Bean(name = "myFlux")
+	@ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "DECORATE_ON_EACH",
+			matchIfMissing = true)
+	public Function<Flux<String>, Flux<String>> onEachFunction() {
+		System.out.println("on each function");
+		return new SleuthFunction();
+	}
+
+	@Bean(name = "myFlux")
+	@ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "DECORATE_ON_LAST")
+	public Function<Flux<String>, Flux<String>> onLastFunction() {
+		System.out.println("on last function");
+		return new SleuthFunction();
+	}
+
+	@Bean(name = "myFlux")
+	@ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "MANUAL")
+	public Function<Flux<String>, Flux<String>> manualFunction() {
+		System.out.println("manual function");
+		return new SleuthManualFunction();
+	}
+
+}
+
+class SimpleFunction implements Function<String, String> {
+
+	private static final Logger log = LoggerFactory.getLogger(SimpleFunction.class);
+
+	@Override
+	public String apply(String input) {
+		log.info("Hello from simple [{}]", input);
+		return input.toUpperCase();
+	}
+
+}
+
+class SimpleReactiveFunction implements Function<Flux<String>, Flux<String>> {
+
+	private static final Logger log = LoggerFactory.getLogger(SimpleReactiveFunction.class);
+
+	@Override
+	public Flux<String> apply(Flux<String> input) {
+		return input.doOnNext(s -> log.info("Hello from simple [{}]", s)).map(String::toUpperCase);
+	}
+
+}
+
+class SimpleManualFunction implements Function<Message<String>, Message<String>> {
+
+	private static final Logger log = LoggerFactory.getLogger(SimpleFunction.class);
+
+	private final Tracing tracing;
+
+	SimpleManualFunction(Tracing tracing) {
+		this.tracing = tracing;
+	}
+
+	@Override
+	public Message<String> apply(Message<String> input) {
+		return (MessagingSleuthOperator.asFunction(this.tracing, input)
+				.andThen(msg -> MessagingSleuthOperator.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 -> MessageBuilder.createMessage(msg.getPayload().toUpperCase(), msg.getHeaders()))
+				.andThen(msg -> MessagingSleuthOperator.afterMessageHandled(this.tracing, msg, null)).apply(input));
+	}
+
+}
+
+class SimpleReactiveManualFunction implements Function<Flux<Message<String>>, Flux<Message<String>>> {
+
+	private static final Logger log = LoggerFactory.getLogger(SimpleReactiveFunction.class);
+
+	private final Tracing tracing;
+
+	SimpleReactiveManualFunction(Tracing tracing) {
+		this.tracing = tracing;
+	}
+
+	@Override
+	public Flux<Message<String>> apply(Flux<Message<String>> input) {
+		return input
+				.map(message -> (MessagingSleuthOperator.asFunction(this.tracing, message))
+						.andThen(msg -> MessagingSleuthOperator.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 -> MessageBuilder.createMessage(msg.getPayload().toUpperCase(), msg.getHeaders()))
+						.andThen(msg -> MessagingSleuthOperator.handleOutputMessage(this.tracing, msg))
+						.apply(message));
+	}
+
+}
+
+class SleuthNonReactiveFunction implements Function<String, String> {
+
+	private static final Logger log = LoggerFactory.getLogger(SleuthNonReactiveFunction.class);
+
+	private final ExecutorService executorService;
+
+	SleuthNonReactiveFunction(ExecutorService executorService) {
+		this.executorService = executorService;
+	}
+
+	@Override
+	public String apply(String input) {
+		log.info("Got a message");
+		try {
+			return this.executorService.submit(() -> {
+				log.info("Logging [{}] from a new thread", input);
+				return input.toUpperCase();
+			}).get(20, TimeUnit.MILLISECONDS);
+		}
+		catch (Exception e) {
+			throw new IllegalStateException(e);
+		}
+	}
+
+}
+
+class SleuthFunction implements Function<Flux<String>, Flux<String>> {
+
+	private static final Logger log = LoggerFactory.getLogger(SleuthFunction.class);
+
+	@Override
+	public Flux<String> apply(Flux<String> input) {
+		return input.doOnEach(signal -> log.info("Got a message"))
+				.flatMap(s -> Mono.delay(Duration.ofMillis(1), Schedulers.newParallel("foo")).map(aLong -> {
+					log.info("Logging [{}] from flat map", s);
+					return s.toUpperCase();
+				}));
+	}
+
+}
+
+class SleuthManualFunction implements Function<Flux<String>, Flux<String>> {
+
+	private static final Logger log = LoggerFactory.getLogger(SleuthManualFunction.class);
+
+	@Override
+	public Flux<String> apply(Flux<String> input) {
+		return input.doOnEach(WebFluxSleuthOperators.withSpanInScope(() -> log.info("Got a message"))).flatMap(s -> Mono
+				.subscriberContext().delayElement(Duration.ofMillis(1), Schedulers.newParallel("foo")).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");
+				});
+	}
+
+}

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/webflux/SleuthBenchmarkingSpringWebFluxApp.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * 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.
@@ -16,13 +16,18 @@
 
 package org.springframework.cloud.sleuth.benchmarks.app.webflux;
 
+import java.time.Duration;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
-import brave.sampler.Sampler;
 import brave.handler.SpanHandler;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+import brave.propagation.TraceContext;
+import brave.sampler.Sampler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.WebApplicationType;
@@ -31,9 +36,12 @@ import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.boot.web.embedded.netty.NettyReactiveWebServerFactory;
 import org.springframework.boot.web.reactive.context.ReactiveWebServerInitializedEvent;
 import org.springframework.cloud.sleuth.instrument.web.SkipPatternProvider;
+import org.springframework.cloud.sleuth.instrument.web.WebFluxSleuthOperators;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.annotation.Bean;
+import org.springframework.util.Assert;
 import org.springframework.util.SocketUtils;
+import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
@@ -42,17 +50,18 @@ import org.springframework.web.bind.annotation.RestController;
  */
 @SpringBootApplication
 @RestController
-public class SleuthBenchmarkingSpringWebFluxApp
-		implements ApplicationListener<ReactiveWebServerInitializedEvent> {
+public class SleuthBenchmarkingSpringWebFluxApp implements ApplicationListener<ReactiveWebServerInitializedEvent> {
 
-	private static final Log log = LogFactory
-			.getLog(SleuthBenchmarkingSpringWebFluxApp.class);
+	private static final Logger log = LoggerFactory.getLogger(SleuthBenchmarkingSpringWebFluxApp.class);
 
+	/**
+	 * Port to set.
+	 */
 	public int port;
 
 	public static void main(String... args) {
-		new SpringApplicationBuilder(SleuthBenchmarkingSpringWebFluxApp.class)
-				.web(WebApplicationType.REACTIVE).application().run(args);
+		new SpringApplicationBuilder(SleuthBenchmarkingSpringWebFluxApp.class).web(WebApplicationType.REACTIVE)
+				.application().run(args);
 	}
 
 	@RequestMapping("/foo")
@@ -71,11 +80,9 @@ public class SleuthBenchmarkingSpringWebFluxApp
 	}
 
 	@Bean
-	NettyReactiveWebServerFactory nettyReactiveWebServerFactory(
-			@Value("${server.port:0}") int serverPort) {
+	NettyReactiveWebServerFactory nettyReactiveWebServerFactory(@Value("${server.port:0}") int serverPort) {
 		log.info("Starting container at port [" + serverPort + "]");
-		return new NettyReactiveWebServerFactory(
-				serverPort == 0 ? SocketUtils.findAvailableTcpPort() : serverPort);
+		return new NettyReactiveWebServerFactory(serverPort == 0 ? SocketUtils.findAvailableTcpPort() : serverPort);
 	}
 
 	@Bean
@@ -90,4 +97,54 @@ public class SleuthBenchmarkingSpringWebFluxApp
 		this.port = event.getWebServer().getPort();
 	}
 
+	@GetMapping("/simple")
+	public Mono<String> simple() {
+		return Mono.just("hello").map(String::toUpperCase);
+	}
+
+	@GetMapping("/complexNoSleuth")
+	public Mono<String> complexNoSleuth() {
+		return Flux.range(1, 10).map(String::valueOf).collect(Collectors.toList())
+				.doOnEach(signal -> log.info("Got a request"))
+				.flatMap(s -> Mono.delay(Duration.ofMillis(1), Schedulers.newParallel("foo")).map(aLong -> {
+					log.info("Logging [{}] from flat map", s);
+					return "";
+				}));
+	}
+
+	@GetMapping("/complex")
+	public Mono<String> complex() {
+		return Flux.range(1, 10).map(String::valueOf).collect(Collectors.toList())
+				.doOnEach(signal -> log.info("Got a request"))
+				.flatMap(s -> Mono.delay(Duration.ofMillis(1), Schedulers.newParallel("foo")).map(aLong -> {
+					log.info("Logging [{}] from flat map", s);
+					return "";
+				})).doOnEach(signal -> {
+					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");
+				});
+	}
+
+	@GetMapping("/complexManual")
+	public Mono<String> complexManual() {
+		return Flux.range(1, 10).map(String::valueOf).collect(Collectors.toList())
+				.doOnEach(WebFluxSleuthOperators.withSpanInScope(() -> log.info("Got a request")))
+				.flatMap(s -> Mono.subscriberContext().delayElement(Duration.ofMillis(1), Schedulers.newParallel("foo"))
+						.map(ctx -> {
+							WebFluxSleuthOperators.withSpanInScope(ctx,
+									() -> log.info("Logging [{}] from flat map", s));
+							return "";
+						}))
+				.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");
+				});
+	}
+
 }

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/SpringWebFluxOnLastBenchmark.java

@@ -1,29 +0,0 @@
-/*
- * Copyright 2013-2019 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.benchmarks.jmh.benchmarks;
-
-public class SpringWebFluxOnLastBenchmark extends SpringWebFluxBenchmarks {
-
-	@Override
-	protected String[] runArgs() {
-		return new String[] { "--spring.jmx.enabled=false",
-				"--spring.application.name=defaultTraceContextWithOnLastOperator",
-				"--spring.sleuth.enabled=true",
-				"--spring.sleuth.reactor.on-each-operator=false" };
-	}
-
-}

benchmarks/src/main/resources/application.yml

@@ -1,3 +1,4 @@
 logging.level:
   org.springframework: ERROR
-  org.springframework.cloud.sleuth.benchmarks: INFO
+  org.springframework.sleuth: ERROR
+  brave: ERROR

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/ProcessLauncherState.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.sleuth.benchmarks.jmh.benchmarks;
+package org.springframework.cloud.sleuth.benchmarks.jmh;
 
 import java.io.BufferedReader;
 import java.io.File;
@@ -57,15 +57,13 @@ public class ProcessLauncherState {
 		this.args.add(count++, "-Djava.security.egd=file:/dev/./urandom");
 		this.args.add(count++, "-XX:TieredStopAtLevel=1"); // zoom
 		if (System.getProperty("bench.args") != null) {
-			this.args.addAll(count++,
-					Arrays.asList(System.getProperty("bench.args").split(" ")));
+			this.args.addAll(count++, Arrays.asList(System.getProperty("bench.args").split(" ")));
 		}
 		this.length = args.length;
 		this.home = new File(dir);
 	}
 
-	protected static String output(InputStream inputStream, String marker)
-			throws IOException {
+	protected static String output(InputStream inputStream, String marker) throws IOException {
 		StringBuilder sb = new StringBuilder();
 		BufferedReader br = null;
 		br = new BufferedReader(new InputStreamReader(inputStream));
@@ -100,8 +98,7 @@ public class ProcessLauncherState {
 
 	public void after() throws Exception {
 		if (started != null && started.isAlive()) {
-			System.err.println(
-					"Stopped " + mainClass + ": " + started.destroyForcibly().waitFor());
+			System.err.println("Stopped " + mainClass + ": " + started.destroyForcibly().waitFor());
 		}
 	}
 

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/RunSleuthJmhBenchmarksFromIde.java

@@ -1,36 +0,0 @@
-/*
- * Copyright 2013-2019 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.benchmarks.jmh;
-
-import org.openjdk.jmh.runner.Runner;
-import org.openjdk.jmh.runner.RunnerException;
-import org.openjdk.jmh.runner.options.Options;
-import org.openjdk.jmh.runner.options.OptionsBuilder;
-
-public class RunSleuthJmhBenchmarksFromIde {
-
-	// Convenience main entry-point for testing from IDE
-	public static void main(String[] args) throws RunnerException {
-		Options opt = new OptionsBuilder()
-				.include(RunSleuthJmhBenchmarksFromIde.class.getPackage().getName()
-						+ ".benchmarks.*")
-				.build();
-
-		new Runner(opt).run();
-	}
-
-}

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/SampleTests.java

@@ -0,0 +1,160 @@
+/*
+ * 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.benchmarks.jmh;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import brave.Tracing;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.sleuth.benchmarks.app.stream.SleuthBenchmarkingStreamApplication;
+import org.springframework.cloud.stream.binder.test.InputDestination;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.StringUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@Disabled
+public class SampleTests {
+
+	@Test
+	public void testStream() throws Exception {
+		for (BenchmarkContext.Instrumentation value : BenchmarkContext.Instrumentation.values()) {
+			run(value);
+		}
+		// run(BenchmarkContext.Instrumentation.sleuthReactiveSimpleManual);
+	}
+
+	private void run(BenchmarkContext.Instrumentation value) throws Exception {
+		BenchmarkContext context = new BenchmarkContext();
+		System.out.println("\n\n\n\n WILL WORK WITH [" + value + "]\n\n\n\n");
+		context.instrumentation = value;
+		context.setup();
+
+		try {
+			context.run(value);
+		}
+		finally {
+			context.clean();
+		}
+		System.out.println("\n\n FINISHED WITH [" + value + "]\n\n\n\n");
+	}
+
+	public static class BenchmarkContext {
+
+		volatile ConfigurableApplicationContext applicationContext;
+		volatile InputDestination input;
+		volatile OutputDestination output;
+		@Param
+		private Instrumentation instrumentation;
+
+		@Setup
+		public void setup() {
+			this.applicationContext = initContext();
+			this.input = this.applicationContext.getBean(InputDestination.class);
+			this.output = this.applicationContext.getBean(OutputDestination.class);
+		}
+
+		protected ConfigurableApplicationContext initContext() {
+			SpringApplication application = new SpringApplicationBuilder(SleuthBenchmarkingStreamApplication.class)
+					.web(WebApplicationType.REACTIVE).application();
+			return application.run(runArgs());
+		}
+
+		protected String[] runArgs() {
+			List<String> strings = new ArrayList<>();
+			strings.addAll(Arrays.asList("--spring.jmx.enabled=false",
+					"--spring.application.name=defaultTraceContextForStream" + instrumentation.name()));
+			strings.addAll(instrumentation.entires.stream().map(s -> "--" + s).collect(Collectors.toList()));
+			return strings.toArray(new String[0]);
+		}
+
+		void run(Instrumentation value) {
+			System.out.println("Sending the message to input");
+			input.send(MessageBuilder.withPayload("hello".getBytes())
+					.setHeader("b3", "4883117762eb9420-4883117762eb9420-1").build());
+			System.out.println("Retrieving the message for tests");
+			Message<byte[]> message = output.receive(200L);
+			System.out.println("Got the message from output");
+			assertThat(message).isNotNull();
+			System.out.println("Message is not null");
+			assertThat(message.getPayload()).isEqualTo("HELLO".getBytes());
+			System.out.println("Payload is HELLO");
+			if (!value.toString().toLowerCase().contains("nosleuth")) {
+				String b3 = message.getHeaders().get("b3", String.class);
+				System.out.println("Checking the b3 header [" + b3 + "]");
+				assertThat(b3).startsWith("4883117762eb9420");
+			}
+		}
+
+		@TearDown
+		public void clean() throws Exception {
+			Tracing current = Tracing.current();
+			if (current != null) {
+				current.close();
+			}
+			try {
+				this.applicationContext.close();
+			}
+			catch (Exception ig) {
+
+			}
+		}
+
+		public enum Instrumentation {
+
+			noSleuthSimple("spring.sleuth.enabled=false,spring.sleuth.function.type=simple");
+
+			private Set<String> entires = new HashSet<>();
+
+			Instrumentation(String key, String value) {
+				this.entires.add(key + "=" + value);
+			}
+
+			Instrumentation(String commaSeparated) {
+				this.entires.addAll(StringUtils.commaDelimitedListToSet(commaSeparated));
+			}
+
+		}
+
+	}
+
+	@Configuration
+	@Import(TestChannelBinderConfiguration.class)
+	static class TestConfiguration {
+
+	}
+
+}

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/mvc/AnnotationBenchmarksTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * 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.
@@ -14,10 +14,11 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.sleuth.benchmarks.jmh.benchmarks;
+package org.springframework.cloud.sleuth.benchmarks.jmh.mvc;
 
 import java.util.concurrent.TimeUnit;
 
+import jmh.mbr.junit5.Microbenchmark;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
 import org.openjdk.jmh.annotations.Fork;
@@ -37,13 +38,18 @@ import org.springframework.context.ConfigurableApplicationContext;
 
 import static org.assertj.core.api.BDDAssertions.then;
 
-@Measurement(iterations = 5)
-@Warmup(iterations = 10)
-@Fork(3)
+@Measurement(iterations = 5, time = 1)
+@Warmup(iterations = 10, time = 1)
+@Fork(2)
 @BenchmarkMode(Mode.SampleTime)
 @OutputTimeUnit(TimeUnit.MICROSECONDS)
 @Threads(Threads.MAX)
-public class AnnotationBenchmarks {
+@Microbenchmark
+public class AnnotationBenchmarksTests {
+
+	static {
+		System.setProperty("jmh.mbr.report.publishTo", "csv:annotation.csv");
+	}
 
 	@Benchmark
 	public void manuallyCreatedSpans(BenchmarkContext context) throws Exception {
@@ -64,9 +70,8 @@ public class AnnotationBenchmarks {
 
 		@Setup
 		public void setup() {
-			this.withSleuth = new SpringApplication(SleuthBenchmarkingSpringApp.class)
-					.run("--spring.jmx.enabled=false",
-							"--spring.application.name=withSleuth");
+			this.withSleuth = new SpringApplication(SleuthBenchmarkingSpringApp.class).run("--spring.jmx.enabled=false",
+					"--spring.application.name=withSleuth");
 			this.sleuth = this.withSleuth.getBean(SleuthBenchmarkingSpringApp.class);
 		}
 

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/mvc/AsyncBenchmarksTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * 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.
@@ -14,10 +14,11 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.sleuth.benchmarks.jmh.benchmarks;
+package org.springframework.cloud.sleuth.benchmarks.jmh.mvc;
 
 import java.util.concurrent.TimeUnit;
 
+import jmh.mbr.junit5.Microbenchmark;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
 import org.openjdk.jmh.annotations.Fork;
@@ -37,13 +38,18 @@ import org.springframework.context.ConfigurableApplicationContext;
 
 import static org.assertj.core.api.BDDAssertions.then;
 
-@Measurement(iterations = 5)
-@Warmup(iterations = 10)
-@Fork(3)
+@Measurement(iterations = 5, time = 1)
+@Warmup(iterations = 10, time = 1)
+@Fork(2)
 @BenchmarkMode(Mode.SampleTime)
 @OutputTimeUnit(TimeUnit.MICROSECONDS)
 @Threads(Threads.MAX)
-public class AsyncBenchmarks {
+@Microbenchmark
+public class AsyncBenchmarksTests {
+
+	static {
+		System.setProperty("jmh.mbr.report.publishTo", "csv:async.csv");
+	}
 
 	@Benchmark
 	public void asyncMethodWithoutSleuth(BenchmarkContext context) throws Exception {
@@ -68,18 +74,13 @@ public class AsyncBenchmarks {
 
 		@Setup
 		public void setup() {
-			this.withSleuth = new SpringApplication(SleuthBenchmarkingSpringApp.class)
-					.run("--spring.jmx.enabled=false",
-							"--spring.application.name=withSleuth");
-			this.withoutSleuth = new SpringApplication(SleuthBenchmarkingSpringApp.class)
-					.run("--spring.jmx.enabled=false",
-							"--spring.application.name=withoutSleuth",
-							"--spring.sleuth.enabled=false",
-							"--spring.sleuth.async.enabled=false");
-			this.tracedAsyncMethodHavingBean = this.withSleuth
-					.getBean(SleuthBenchmarkingSpringApp.class);
-			this.untracedAsyncMethodHavingBean = this.withoutSleuth
-					.getBean(SleuthBenchmarkingSpringApp.class);
+			this.withSleuth = new SpringApplication(SleuthBenchmarkingSpringApp.class).run("--spring.jmx.enabled=false",
+					"--spring.application.name=withSleuth");
+			this.withoutSleuth = new SpringApplication(SleuthBenchmarkingSpringApp.class).run(
+					"--spring.jmx.enabled=false", "--spring.application.name=withoutSleuth",
+					"--spring.sleuth.enabled=false", "--spring.sleuth.async.enabled=false");
+			this.tracedAsyncMethodHavingBean = this.withSleuth.getBean(SleuthBenchmarkingSpringApp.class);
+			this.untracedAsyncMethodHavingBean = this.withoutSleuth.getBean(SleuthBenchmarkingSpringApp.class);
 		}
 
 		@TearDown

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/mvc/HttpFilterBenchmarksTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * 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.
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.sleuth.benchmarks.jmh.benchmarks;
+package org.springframework.cloud.sleuth.benchmarks.jmh.mvc;
 
 import java.io.IOException;
 import java.util.concurrent.Callable;
@@ -28,6 +28,7 @@ import javax.servlet.ServletRequest;
 import javax.servlet.ServletResponse;
 
 import brave.servlet.TracingFilter;
+import jmh.mbr.junit5.Microbenchmark;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
 import org.openjdk.jmh.annotations.Fork;
@@ -62,17 +63,21 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
-@Warmup(iterations = 10)
+@Warmup(iterations = 5)
 @BenchmarkMode(Mode.SampleTime)
 @OutputTimeUnit(TimeUnit.MICROSECONDS)
 @Threads(Threads.MAX)
-public class HttpFilterBenchmarks {
+@Microbenchmark
+public class HttpFilterBenchmarksTests {
+
+	static {
+		System.setProperty("jmh.mbr.report.publishTo", "csv:http_filter.csv");
+	}
 
 	@Benchmark
 	@Measurement(iterations = 5, time = 1)
-	@Fork(3)
-	public void filterWithoutSleuth(BenchmarkContext context)
-			throws IOException, ServletException {
+	@Fork(2)
+	public void filterWithoutSleuth(BenchmarkContext context) throws IOException, ServletException {
 		MockHttpServletRequest request = builder().buildRequest(new MockServletContext());
 		MockHttpServletResponse response = new MockHttpServletResponse();
 		response.setContentType(MediaType.APPLICATION_JSON_VALUE);
@@ -82,9 +87,8 @@ public class HttpFilterBenchmarks {
 
 	@Benchmark
 	@Measurement(iterations = 5, time = 1)
-	@Fork(3)
-	public void filterWithSleuth(BenchmarkContext context)
-			throws ServletException, IOException {
+	@Fork(2)
+	public void filterWithSleuth(BenchmarkContext context) throws ServletException, IOException {
 		MockHttpServletRequest request = builder().buildRequest(new MockServletContext());
 		MockHttpServletResponse response = new MockHttpServletResponse();
 		response.setContentType(MediaType.APPLICATION_JSON_VALUE);
@@ -107,12 +111,10 @@ public class HttpFilterBenchmarks {
 	}
 
 	private MockHttpServletRequestBuilder builder() {
-		return get("/").accept(MediaType.APPLICATION_JSON).header("User-Agent",
-				"MockMvc");
+		return get("/").accept(MediaType.APPLICATION_JSON).header("User-Agent", "MockMvc");
 	}
 
-	private void performRequest(MockMvc mockMvc, String url, String expectedResult)
-			throws Exception {
+	private void performRequest(MockMvc mockMvc, String url, String expectedResult) throws Exception {
 		MvcResult mvcResult = mockMvc.perform(get("/" + url)).andExpect(status().isOk())
 				.andExpect(request().asyncStarted()).andReturn();
 
@@ -135,16 +137,12 @@ public class HttpFilterBenchmarks {
 
 		@Setup
 		public void setup() {
-			this.withSleuth = new SpringApplication(SleuthBenchmarkingSpringApp.class)
-					.run("--spring.jmx.enabled=false",
-							"--spring.application.name=withSleuth");
+			this.withSleuth = new SpringApplication(SleuthBenchmarkingSpringApp.class).run("--spring.jmx.enabled=false",
+					"--spring.application.name=withSleuth");
 			this.tracingFilter = this.withSleuth.getBean(TracingFilter.class);
 			this.mockMvcForTracedController = MockMvcBuilders
-					.standaloneSetup(
-							this.withSleuth.getBean(SleuthBenchmarkingSpringApp.class))
-					.build();
-			this.mockMvcForUntracedController = MockMvcBuilders
-					.standaloneSetup(new VanillaController()).build();
+					.standaloneSetup(this.withSleuth.getBean(SleuthBenchmarkingSpringApp.class)).build();
+			this.mockMvcForUntracedController = MockMvcBuilders.standaloneSetup(new VanillaController()).build();
 		}
 
 		@TearDown
@@ -162,8 +160,8 @@ public class HttpFilterBenchmarks {
 		}
 
 		@Override
-		public void doFilter(ServletRequest request, ServletResponse response,
-				FilterChain chain) throws IOException, ServletException {
+		public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
+				throws IOException, ServletException {
 			chain.doFilter(request, response);
 		}
 

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/mvc/RestTemplateBenchmarkTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * 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.
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.sleuth.benchmarks.jmh.benchmarks;
+package org.springframework.cloud.sleuth.benchmarks.jmh.mvc;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit;
 import javax.servlet.ServletException;
 
 import brave.spring.web.TracingClientHttpRequestInterceptor;
+import jmh.mbr.junit5.Microbenchmark;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
 import org.openjdk.jmh.annotations.Fork;
@@ -49,24 +50,26 @@ import static org.assertj.core.api.BDDAssertions.then;
 /**
  * We're checking how much overhead does the instrumentation of the RestTemplate take
  */
-@Measurement(iterations = 5)
-@Warmup(iterations = 10)
-@Fork(3)
+@Measurement(iterations = 5, time = 1)
+@Warmup(iterations = 10, time = 1)
+@Fork(2)
 @BenchmarkMode(Mode.SampleTime)
 @OutputTimeUnit(TimeUnit.MICROSECONDS)
 @Threads(Threads.MAX)
-public class RestTemplateBenchmark {
+@Microbenchmark
+public class RestTemplateBenchmarkTests {
 
-	@Benchmark
-	public void syncEndpointWithoutSleuth(BenchmarkContext context)
-			throws IOException, ServletException {
-		then(context.untracedTemplate.getForObject("/foo", String.class))
-				.isEqualTo("foo");
+	static {
+		System.setProperty("jmh.mbr.report.publishTo", "csv:rest_template.csv");
 	}
 
 	@Benchmark
-	public void syncEndpointWithSleuth(BenchmarkContext context)
-			throws ServletException, IOException {
+	public void syncEndpointWithoutSleuth(BenchmarkContext context) throws IOException, ServletException {
+		then(context.untracedTemplate.getForObject("/foo", String.class)).isEqualTo("foo");
+	}
+
+	@Benchmark
+	public void syncEndpointWithSleuth(BenchmarkContext context) throws ServletException, IOException {
 		then(context.tracedTemplate.getForObject("/foo", String.class)).isEqualTo("foo");
 	}
 
@@ -83,18 +86,14 @@ public class RestTemplateBenchmark {
 
 		@Setup
 		public void setup() {
-			new SpringApplication(SleuthBenchmarkingSpringApp.class).run(
-					"--spring.jmx.enabled=false", "--spring.application.name=withSleuth");
-			this.mockMvc = MockMvcBuilders
-					.standaloneSetup(
-							this.withSleuth.getBean(SleuthBenchmarkingSpringApp.class))
+			new SpringApplication(SleuthBenchmarkingSpringApp.class).run("--spring.jmx.enabled=false",
+					"--spring.application.name=withSleuth");
+			this.mockMvc = MockMvcBuilders.standaloneSetup(this.withSleuth.getBean(SleuthBenchmarkingSpringApp.class))
 					.build();
-			this.tracedTemplate = new RestTemplate(
-					new MockMvcClientHttpRequestFactory(this.mockMvc));
-			this.tracedTemplate.setInterceptors(Collections.singletonList(
-					this.withSleuth.getBean(TracingClientHttpRequestInterceptor.class)));
-			this.untracedTemplate = new RestTemplate(
-					new MockMvcClientHttpRequestFactory(this.mockMvc));
+			this.tracedTemplate = new RestTemplate(new MockMvcClientHttpRequestFactory(this.mockMvc));
+			this.tracedTemplate.setInterceptors(
+					Collections.singletonList(this.withSleuth.getBean(TracingClientHttpRequestInterceptor.class)));
+			this.untracedTemplate = new RestTemplate(new MockMvcClientHttpRequestFactory(this.mockMvc));
 		}
 
 		@TearDown

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/mvc/StartupBenchmarkTests.java

@@ -14,8 +14,9 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.sleuth.benchmarks.jmh.benchmarks;
+package org.springframework.cloud.sleuth.benchmarks.jmh.mvc;
 
+import jmh.mbr.junit5.Microbenchmark;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
 import org.openjdk.jmh.annotations.Fork;
@@ -27,11 +28,18 @@ import org.openjdk.jmh.annotations.State;
 import org.openjdk.jmh.annotations.TearDown;
 import org.openjdk.jmh.annotations.Warmup;
 
+import org.springframework.cloud.sleuth.benchmarks.jmh.ProcessLauncherState;
+
 @Measurement(iterations = 5)
 @Warmup(iterations = 1)
 @Fork(value = 2, warmups = 0)
 @BenchmarkMode(Mode.AverageTime)
-public class StartupBenchmark {
+@Microbenchmark
+public class StartupBenchmarkTests {
+
+	static {
+		System.setProperty("jmh.mbr.report.publishTo", "csv:startup.csv");
+	}
 
 	@Benchmark
 	public void withAnnotations(ApplicationState state) throws Exception {
@@ -46,25 +54,21 @@ public class StartupBenchmark {
 
 	@Benchmark
 	public void withoutAsync(ApplicationState state) throws Exception {
-		state.setExtraArgs("--spring.sleuth.async.enabled=false",
-				"--spring.sleuth.annotation.enabled=false");
+		state.setExtraArgs("--spring.sleuth.async.enabled=false", "--spring.sleuth.annotation.enabled=false");
 		state.run();
 	}
 
 	@Benchmark
 	public void withoutScheduled(ApplicationState state) throws Exception {
-		state.setExtraArgs("--spring.sleuth.scheduled.enabled=false",
-				"--spring.sleuth.async.enabled=false",
+		state.setExtraArgs("--spring.sleuth.scheduled.enabled=false", "--spring.sleuth.async.enabled=false",
 				"--spring.sleuth.annotation.enabled=false");
 		state.run();
 	}
 
 	@Benchmark
 	public void withoutWeb(ApplicationState state) throws Exception {
-		state.setExtraArgs("--spring.sleuth.web.enabled=false",
-				"--spring.sleuth.scheduled.enabled=false",
-				"--spring.sleuth.async.enabled=false",
-				"--spring.sleuth.annotation.enabled=false");
+		state.setExtraArgs("--spring.sleuth.web.enabled=false", "--spring.sleuth.scheduled.enabled=false",
+				"--spring.sleuth.async.enabled=false", "--spring.sleuth.annotation.enabled=false");
 		state.run();
 	}
 

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/stream/MicroBenchmarkStreamTests.java

@@ -0,0 +1,179 @@
+/*
+ * 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.benchmarks.jmh.stream;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import brave.Tracing;
+import jmh.mbr.junit5.Microbenchmark;
+import org.junit.platform.commons.annotation.Testable;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.sleuth.benchmarks.app.stream.SleuthBenchmarkingStreamApplication;
+import org.springframework.cloud.stream.binder.test.InputDestination;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.StringUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@Measurement(iterations = 5, time = 1)
+@Warmup(iterations = 10, time = 1)
+@Fork(2)
+@BenchmarkMode(Mode.SampleTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Microbenchmark
+public class MicroBenchmarkStreamTests {
+
+	static {
+		System.setProperty("jmh.mbr.report.publishTo", "csv:stream.csv");
+	}
+
+	@Benchmark
+	@Testable
+	public void testStream(BenchmarkContext context) throws Exception {
+		context.run();
+	}
+
+	@State(Scope.Benchmark)
+	public static class BenchmarkContext {
+
+		volatile ConfigurableApplicationContext applicationContext;
+
+		volatile InputDestination input;
+
+		volatile OutputDestination output;
+
+		@Param
+		private Instrumentation instrumentation;
+
+		@Setup
+		public void setup() {
+			this.applicationContext = initContext();
+			this.input = this.applicationContext.getBean(InputDestination.class);
+			this.output = this.applicationContext.getBean(OutputDestination.class);
+		}
+
+		protected ConfigurableApplicationContext initContext() {
+			SpringApplication application = new SpringApplicationBuilder(SleuthBenchmarkingStreamApplication.class)
+					.web(WebApplicationType.REACTIVE).application();
+			return application.run(runArgs());
+		}
+
+		protected String[] runArgs() {
+			List<String> strings = new ArrayList<>();
+			strings.addAll(Arrays.asList("--spring.jmx.enabled=false",
+					"--spring.application.name=defaultTraceContextForStream" + instrumentation.name()));
+			strings.addAll(instrumentation.entires.stream().map(s -> "--" + s).collect(Collectors.toList()));
+			return strings.toArray(new String[0]);
+		}
+
+		void run() {
+			// System.out.println("Sending the message to input");
+			input.send(MessageBuilder.withPayload("hello".getBytes())
+					.setHeader("b3", "4883117762eb9420-4883117762eb9420-1").build());
+			// System.out.println("Retrieving the message for tests");
+			Message<byte[]> message = output.receive(200L);
+			// System.out.println("Got the message from output");
+			assertThat(message).isNotNull();
+			// System.out.println("Message is not null");
+			assertThat(message.getPayload()).isEqualTo("HELLO".getBytes());
+			// System.out.println("Payload is HELLO");
+			if (!instrumentation.toString().toLowerCase().contains("nosleuth")) {
+				String b3 = message.getHeaders().get("b3", String.class);
+				// System.out.println("Checking the b3 header [" + b3 + "]");
+				assertThat(b3).startsWith("4883117762eb9420");
+			}
+		}
+
+		@TearDown
+		public void clean() throws Exception {
+			Tracing current = Tracing.current();
+			if (current != null) {
+				current.close();
+			}
+			try {
+				this.applicationContext.close();
+			}
+			catch (Exception ig) {
+
+			}
+		}
+
+		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"),
+			// 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");
+
+			private Set<String> entires = new HashSet<>();
+
+			Instrumentation(String key, String value) {
+				this.entires.add(key + "=" + value);
+			}
+
+			Instrumentation(String commaSeparated) {
+				this.entires.addAll(StringUtils.commaDelimitedListToSet(commaSeparated));
+			}
+
+		}
+
+	}
+
+	@Configuration
+	@Import(TestChannelBinderConfiguration.class)
+	static class TestConfiguration {
+
+	}
+
+}

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/webflux/MicroBenchmarkHttpTests.java

@@ -0,0 +1,135 @@
+/*
+ * 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.benchmarks.jmh.webflux;
+
+import java.util.concurrent.TimeUnit;
+
+import brave.Tracing;
+import jmh.mbr.junit5.Microbenchmark;
+import org.junit.platform.commons.annotation.Testable;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.sleuth.benchmarks.app.webflux.SleuthBenchmarkingSpringWebFluxApp;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.test.web.reactive.server.WebTestClient;
+
+@Measurement(iterations = 5, time = 1)
+@Warmup(iterations = 10, time = 1)
+@Fork(2)
+@BenchmarkMode(Mode.SampleTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Microbenchmark
+public class MicroBenchmarkHttpTests {
+
+	static {
+		System.setProperty("jmh.mbr.report.publishTo", "csv:http.csv");
+	}
+
+	@Benchmark
+	@Testable
+	public void test(BenchmarkContext context) throws Exception {
+		context.run();
+	}
+
+	@State(Scope.Benchmark)
+	public static class BenchmarkContext {
+
+		volatile ConfigurableApplicationContext applicationContext;
+
+		volatile WebTestClient webTestClient;
+
+		@Param
+		private Instrumentation instrumentation;
+
+		@Setup
+		public void setup() {
+			this.applicationContext = initContext();
+			this.webTestClient = WebTestClient.bindToApplicationContext(applicationContext).build();
+		}
+
+		protected ConfigurableApplicationContext initContext() {
+			SpringApplication application = new SpringApplicationBuilder(SleuthBenchmarkingSpringWebFluxApp.class)
+					.web(WebApplicationType.REACTIVE).application();
+			return application.run(runArgs());
+		}
+
+		protected String[] runArgs() {
+			return new String[] { "--spring.jmx.enabled=false",
+					"--spring.application.name=defaultTraceContext" + instrumentation.name(),
+					"--" + instrumentation.key + "=" + instrumentation.value };
+		}
+
+		void run() {
+			this.webTestClient.get().uri(this.instrumentation.url).header("X-B3-TraceId", "4883117762eb9420")
+					.header("X-B3-SpanId", "4883117762eb9420").exchange().expectStatus().isOk();
+		}
+
+		@TearDown
+		public void clean() throws Exception {
+			Tracing current = Tracing.current();
+			if (current != null) {
+				current.close();
+			}
+			try {
+				this.applicationContext.close();
+			}
+			catch (Exception ig) {
+
+			}
+		}
+
+		public enum Instrumentation {
+
+			noSleuthSimple("spring.sleuth.enabled", "false", "/simple"), sleuthSimpleManual(
+					"spring.sleuth.reactor.instrumentation-type", "MANUAL",
+					"/simple"), noSleuthComplex("spring.sleuth.enabled", "false", "/complexNoSleuth"), onEachComplex(
+							"spring.sleuth.reactor.instrumentation-type", "DECORATE_ON_EACH",
+							"/complex"), onLastComplex("spring.sleuth.reactor.instrumentation-type", "DECORATE_ON_LAST",
+									"/complex"), onManualComplex("spring.sleuth.reactor.instrumentation-type", "MANUAL",
+											"/complexManual");
+
+			private String key;
+
+			private String value;
+
+			private String url;
+
+			Instrumentation(String key, String value, String url) {
+				this.key = key;
+				this.value = value;
+				this.url = url;
+			}
+
+		}
+
+	}
+
+}

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/webflux/SpringWebFluxBenchmarksTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * 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.
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.sleuth.benchmarks.jmh.benchmarks;
+package org.springframework.cloud.sleuth.benchmarks.jmh.webflux;
 
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
@@ -26,6 +26,7 @@ import brave.httpclient.TracingHttpClientBuilder;
 import brave.propagation.CurrentTraceContext;
 import brave.propagation.TraceContext;
 import brave.sampler.Sampler;
+import jmh.mbr.junit5.Microbenchmark;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClients;
@@ -55,28 +56,38 @@ import org.springframework.context.ConfigurableApplicationContext;
 
 @Measurement(iterations = 5, time = 1)
 @Warmup(iterations = 10, time = 1)
-@Fork(3)
+@Fork(2)
 @BenchmarkMode(Mode.SampleTime)
 @OutputTimeUnit(TimeUnit.MICROSECONDS)
 @Threads(2)
 @State(Scope.Benchmark)
-public class SpringWebFluxBenchmarks {
+@Microbenchmark
+public class SpringWebFluxBenchmarksTests {
+
 	static final SpanHandler FAKE_SPAN_HANDLER = new SpanHandler() {
 		// intentionally anonymous to prevent logging fallback on NOOP
 	};
+	protected static TraceContext defaultTraceContext = TraceContext.newBuilder().traceIdHigh(333L).traceId(444L)
+			.spanId(3).sampled(true).build();
+
+	static {
+		System.setProperty("jmh.mbr.report.publishTo", "csv:webflux.csv");
+	}
 
-	protected static TraceContext defaultTraceContext = TraceContext.newBuilder()
-			.traceIdHigh(333L).traceId(444L).spanId(3).sampled(true).build();
 	protected ConfigurableApplicationContext applicationContext;
+
 	protected SleuthBenchmarkingSpringWebFluxApp springWebFluxApp;
+
 	CloseableHttpClient client;
+
 	CloseableHttpClient tracedClient;
+
 	CloseableHttpClient unsampledClient;
+
 	private String baseUrl;
 
 	public static void main(String[] args) throws RunnerException {
-		Options opt = new OptionsBuilder()
-				.include(".*" + SpringWebFluxBenchmarks.class.getSimpleName() + ".*")
+		Options opt = new OptionsBuilder().include(".*" + SpringWebFluxBenchmarksTests.class.getSimpleName() + ".*")
 				.build();
 
 		new Runner(opt).run();
@@ -87,8 +98,7 @@ public class SpringWebFluxBenchmarks {
 	}
 
 	protected CloseableHttpClient newClient(HttpTracing httpTracing) {
-		return TracingHttpClientBuilder.create(httpTracing).disableAutomaticRetries()
-				.build();
+		return TracingHttpClientBuilder.create(httpTracing).disableAutomaticRetries().build();
 	}
 
 	protected CloseableHttpClient newClient() {
@@ -107,21 +117,18 @@ public class SpringWebFluxBenchmarks {
 	public void setup() {
 		ConfigurableApplicationContext context = initContext();
 		this.applicationContext = context;
-		this.springWebFluxApp = this.applicationContext
-				.getBean(SleuthBenchmarkingSpringWebFluxApp.class);
+		this.springWebFluxApp = this.applicationContext.getBean(SleuthBenchmarkingSpringWebFluxApp.class);
 		baseUrl = "http://127.0.0.1:" + springWebFluxApp.port + "/foo";
 		client = newClient();
-		tracedClient = newClient(HttpTracing
-				.create(Tracing.newBuilder().addSpanHandler(FAKE_SPAN_HANDLER).build()));
-		unsampledClient = newClient(HttpTracing.create(Tracing.newBuilder()
-				.sampler(Sampler.NEVER_SAMPLE).addSpanHandler(FAKE_SPAN_HANDLER).build()));
+		tracedClient = newClient(HttpTracing.create(Tracing.newBuilder().addSpanHandler(FAKE_SPAN_HANDLER).build()));
+		unsampledClient = newClient(HttpTracing
+				.create(Tracing.newBuilder().sampler(Sampler.NEVER_SAMPLE).addSpanHandler(FAKE_SPAN_HANDLER).build()));
 		postSetUp();
 	}
 
 	protected ConfigurableApplicationContext initContext() {
-		SpringApplication application = new SpringApplicationBuilder(
-				SleuthBenchmarkingSpringWebFluxApp.class).web(WebApplicationType.REACTIVE)
-						.application();
+		SpringApplication application = new SpringApplicationBuilder(SleuthBenchmarkingSpringWebFluxApp.class)
+				.web(WebApplicationType.REACTIVE).application();
 		customSpringApplication(application);
 		return application.run(runArgs());
 	}
@@ -134,8 +141,7 @@ public class SpringWebFluxBenchmarks {
 	}
 
 	protected String[] runArgs() {
-		return new String[] { "--spring.jmx.enabled=false",
-				"--spring.application.name=defaultTraceContext",
+		return new String[] { "--spring.jmx.enabled=false", "--spring.application.name=defaultTraceContext",
 				"--spring.sleuth.enabled=true" };
 	}
 
@@ -171,8 +177,7 @@ public class SpringWebFluxBenchmarks {
 
 	@Benchmark
 	public void tracedClient_get_resumeTrace() throws Exception {
-		try (CurrentTraceContext.Scope scope = Tracing.current().currentTraceContext()
-				.newScope(defaultTraceContext)) {
+		try (CurrentTraceContext.Scope scope = Tracing.current().currentTraceContext().newScope(defaultTraceContext)) {
 			get(tracedClient);
 		}
 	}

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/webflux/WithOutReactorSleuthSpringWebFluxBenchmarksTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * 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.
@@ -14,8 +14,9 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.sleuth.benchmarks.jmh.benchmarks;
+package org.springframework.cloud.sleuth.benchmarks.jmh.webflux;
 
+import jmh.mbr.junit5.Microbenchmark;
 import org.openjdk.jmh.runner.Runner;
 import org.openjdk.jmh.runner.RunnerException;
 import org.openjdk.jmh.runner.options.Options;
@@ -24,21 +25,23 @@ import org.openjdk.jmh.runner.options.OptionsBuilder;
 /**
  * @author alvin
  */
-public class WithOutReactorSleuthSpringWebFluxBenchmarks extends SpringWebFluxBenchmarks {
+@Microbenchmark
+public class WithOutReactorSleuthSpringWebFluxBenchmarksTests extends SpringWebFluxBenchmarksTests {
+
+	static {
+		System.setProperty("jmh.mbr.report.publishTo", "csv:webflux_no_reactor_instrumentation.csv");
+	}
 
 	public static void main(String[] args) throws RunnerException {
-		Options opt = new OptionsBuilder().include(
-				".*" + WithOutReactorSleuthSpringWebFluxBenchmarks.class.getSimpleName()
-						+ ".*")
-				.build();
+		Options opt = new OptionsBuilder()
+				.include(".*" + WithOutReactorSleuthSpringWebFluxBenchmarksTests.class.getSimpleName() + ".*").build();
 
 		new Runner(opt).run();
 	}
 
 	@Override
 	protected String[] runArgs() {
-		return new String[] { "--spring.jmx.enabled=false",
-				"--spring.application.name=defaultTraceContext",
+		return new String[] { "--spring.jmx.enabled=false", "--spring.application.name=defaultTraceContext",
 				"--spring.sleuth.enabled=true", "--spring.sleuth.reactor.enabled=false"
 
 		};

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/webflux/WithOutSleuthSpringWebFluxBenchmarksTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * 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.
@@ -14,8 +14,9 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.sleuth.benchmarks.jmh.benchmarks;
+package org.springframework.cloud.sleuth.benchmarks.jmh.webflux;
 
+import jmh.mbr.junit5.Microbenchmark;
 import org.openjdk.jmh.runner.Runner;
 import org.openjdk.jmh.runner.RunnerException;
 import org.openjdk.jmh.runner.options.Options;
@@ -24,20 +25,23 @@ import org.openjdk.jmh.runner.options.OptionsBuilder;
 /**
  * @author alvin
  */
-public class WithOutSleuthSpringWebFluxBenchmarks extends SpringWebFluxBenchmarks {
+@Microbenchmark
+public class WithOutSleuthSpringWebFluxBenchmarksTests extends SpringWebFluxBenchmarksTests {
+
+	static {
+		System.setProperty("jmh.mbr.report.publishTo", "csv:webflux_no_sleuth_instrumentation.csv");
+	}
 
 	public static void main(String[] args) throws RunnerException {
-		Options opt = new OptionsBuilder().include(
-				".*" + WithOutSleuthSpringWebFluxBenchmarks.class.getSimpleName() + ".*")
-				.build();
+		Options opt = new OptionsBuilder()
+				.include(".*" + WithOutSleuthSpringWebFluxBenchmarksTests.class.getSimpleName() + ".*").build();
 
 		new Runner(opt).run();
 	}
 
 	@Override
 	protected String[] runArgs() {
-		return new String[] { "--spring.jmx.enabled=false",
-				"--spring.application.name=defaultTraceContext",
+		return new String[] { "--spring.jmx.enabled=false", "--spring.application.name=defaultTraceContext",
 				"--spring.sleuth.enabled=false" };
 	}