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

New reactor queue wrapping (#1858)

Browse Source 
fixes gh-1843
This commit is contained in:
Marcin Grzejszczak
2021-02-25 12:43:00 +01:00
committed by GitHub
parent b9419438eb
commit a937765c9f
40 changed files with 1670 additions and 464 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
benchmarks/pom.xml
View File

@@ -25,16 +25,23 @@
<version>2.2.8.BUILD-SNAPSHOT</version>
<artifactId>benchmarks</artifactId>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.9.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<main.basedir>${project.basedir}/..</main.basedir>
<jmh.version>1.22</jmh.version>
<maven-shade-plugin.version>3.2.1</maven-shade-plugin.version>
<sonar.skip>true</sonar.skip>
<maven.compiler.target>1.8</maven.compiler.target>
<maven.compiler.source>1.8</maven.compiler.source>
<spring-boot.version>2.3.8.RELEASE</spring-boot.version>
<brave.version>5.12.7</brave.version>
<okhttp.version>3.14.6</okhttp.version>
<okhttp.version>4.9.0</okhttp.version>
<microbenchmark-runner.version>0.2.0.RELEASE</microbenchmark-runner.version>
<jmh.version>1.21</jmh.version>
<spring-cloud-stream.version>Horsham.SR11</spring-cloud-stream.version>
</properties>
<dependencyManagement>
@@ -47,10 +54,9 @@
<scope>import</scope>
</dependency>
<dependency>
<!-- Import dependency management from Spring Boot -->
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-dependencies</artifactId>
<version>${spring-boot.version}</version>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-dependencies</artifactId>
<version>${spring-cloud-stream.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
@@ -91,20 +97,35 @@
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<version>3.14.0</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.hamcrest</groupId>
<artifactId>hamcrest-core</artifactId>
<version>1.3</version>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
<type>test-jar</type>
<scope>compile</scope>
<classifier>test-binder</classifier>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.openjdk.jmh</groupId>
<artifactId>jmh-core</artifactId>
<version>${jmh.version}</version>
<groupId>com.github.mp911de.microbenchmark-runner</groupId>
<artifactId>microbenchmark-runner-junit5</artifactId>
<version>${microbenchmark-runner.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.github.mp911de.microbenchmark-runner</groupId>
<artifactId>microbenchmark-runner-extras</artifactId>
<version>${microbenchmark-runner.version}</version>
<scope>test</scope>
</dependency>
<dependency>
@@ -122,12 +143,16 @@
<dependency>
<groupId>io.zipkin.brave</groupId>
<artifactId>brave-instrumentation-httpclient</artifactId>
<version>${brave.version}</version>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
@@ -140,28 +165,19 @@
<target>${maven.compiler.target}</target>
</configuration>
</plugin>
<plugin>
<!--skip deploy (this is just a test module) -->
<artifactId>maven-deploy-plugin</artifactId>
<configuration>
<skip>true</skip>
</configuration>
</plugin>
<plugin>
<artifactId>maven-install-plugin</artifactId>
<configuration>
<skip>true</skip>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<id>jitpack.io</id>
<url>https://jitpack.io</url>
</repository>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
@@ -184,7 +200,7 @@
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
@@ -202,7 +218,7 @@
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
@@ -213,7 +229,7 @@
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
@@ -221,7 +237,7 @@
<pluginRepository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/libs-release-local</url>
<url>https://repo.spring.io/release</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
@@ -229,75 +245,6 @@
</pluginRepositories>
<profiles>
<profile>
<id>jmh</id>
<activation>
<activeByDefault>false</activeByDefault>
</activation>
<build>
<plugins>
<plugin>
<artifactId>maven-shade-plugin</artifactId>
<version>${maven-shade-plugin.version}</version>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>${spring-boot.version}</version>
</dependency>
</dependencies>
<configuration>
<keepDependenciesWithProvidedScope>true
</keepDependenciesWithProvidedScope>
<createDependencyReducedPom>true</createDependencyReducedPom>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
</configuration>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<finalName>benchmarks</finalName>
<transformers>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/spring.handlers</resource>
</transformer>
<transformer
implementation="org.springframework.boot.maven.PropertiesMergingResourceTransformer">
<resource>META-INF/spring.factories</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/spring.schemas</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>org.openjdk.jmh.Main</mainClass>
</transformer>
</transformers>
<createDependencyReducedPom>false
</createDependencyReducedPom>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</profile>
<profile>
<id>jmeter</id>
<activation>
@@ -380,7 +327,7 @@
<plugin>
<groupId>com.lazerycode.jmeter</groupId>
<artifactId>jmeter-maven-plugin</artifactId>
<version>1.10.1</version>
<version>3.1.1</version>
<configuration>
<testResultsTimestamp>false</testResultsTimestamp>
<suppressJMeterOutput>false</suppressJMeterOutput>

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

@@ -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,10 +16,6 @@
package org.springframework.cloud.sleuth.benchmarks.app.mvc;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.regex.Pattern;
@@ -27,7 +23,6 @@ import javax.annotation.PreDestroy;
import brave.Span;
import brave.Tracer;
import brave.sampler.Sampler;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -41,28 +36,26 @@ import org.springframework.boot.web.servlet.server.ServletWebServerFactory;
import org.springframework.cloud.sleuth.annotation.ContinueSpan;
import org.springframework.cloud.sleuth.annotation.NewSpan;
import org.springframework.cloud.sleuth.annotation.SpanTag;
import org.springframework.cloud.sleuth.benchmarks.app.mvc.controller.AsyncSimulationController;
import org.springframework.cloud.sleuth.instrument.web.SkipPatternProvider;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.Async;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.util.SocketUtils;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* @author Marcin Grzejszczak
*/
@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)
@@ -71,28 +64,16 @@ public class SleuthBenchmarkingSpringApp
@Autowired
AClass aClass;
@Autowired
AsyncSimulationController controller;
public static void main(String... args) {
SpringApplication.run(SleuthBenchmarkingSpringApp.class, args);
}
@RequestMapping("/foo")
public String foo() {
return "foo";
}
@RequestMapping("/bar")
public Callable<String> bar() {
return () -> "bar";
}
@RequestMapping("/async")
public String asyncHttp() throws ExecutionException, InterruptedException {
return this.async().get();
}
@Async
public Future<String> async() {
return this.pool.submit(() -> "async");
@PreDestroy
public void clean() {
this.controller.clean();
}
public String manualSpan() {
@@ -108,48 +89,43 @@ public class SleuthBenchmarkingSpringApp
this.port = event.getSource().getPort();
}
@Bean
public ServletWebServerFactory servletContainer(
@Value("${server.port:0}") int serverPort) {
log.info("Starting container at port [" + serverPort + "]");
return new TomcatServletWebServerFactory(
serverPort == 0 ? SocketUtils.findAvailableTcpPort() : serverPort);
public Future<String> async() {
return this.controller.async();
}
@PreDestroy
public void clean() {
this.pool.shutdownNow();
}
@Configuration
static class Config {
@Autowired(required = false)
Tracer tracer;
@Bean
Sampler alwaysSampler() {
return Sampler.ALWAYS_SAMPLE;
}
@Bean
AnotherClass anotherClass() {
return new AnotherClass(this.tracer);
}
@Bean
AnotherClass anotherClass() {
return new AnotherClass(this.tracer);
}
@Bean
AClass aClass() {
return new AClass(this.tracer, anotherClass());
}
@Bean
AClass aClass() {
return new AClass(this.tracer, anotherClass());
}
@Bean
SkipPatternProvider patternProvider() {
return new SkipPatternProvider() {
@Override
public Pattern skipPattern() {
return Pattern.compile("");
}
};
}
@Bean
SkipPatternProvider patternProvider() {
return new SkipPatternProvider() {
@Override
public Pattern skipPattern() {
return Pattern.compile("");
}
};
}
public ExecutorService getPool() {
return this.pool;
}
@Bean
public ServletWebServerFactory servletContainer(@Value("${server.port:0}") int serverPort) {
log.info("Starting container at port [" + serverPort + "]");
return new TomcatServletWebServerFactory(serverPort == 0 ? SocketUtils.findAvailableTcpPort() : serverPort);
}
}
}
class AClass {

50
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/mvc/controller/AsyncSimulationController.java Normal file
View File

@@ -0,0 +1,50 @@
package org.springframework.cloud.sleuth.benchmarks.app.mvc.controller;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import javax.annotation.PreDestroy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* @author Marcin Grzejszczak
*/
@RestController
public class AsyncSimulationController {
private final ExecutorService pool = Executors.newWorkStealingPool();
@RequestMapping("/foo")
public String foo() {
return "foo";
}
@RequestMapping("/bar")
public Callable<String> bar() {
return () -> "bar";
}
@RequestMapping("/async")
public String asyncHttp() throws ExecutionException, InterruptedException {
return this.async().get();
}
@Async
public Future<String> async() {
return this.pool.submit(() -> "async");
}
@PreDestroy
public void clean() {
this.pool.shutdownNow();
}
public ExecutorService getPool() {
return this.pool;
}
}

209
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java Normal file
View File

@@ -0,0 +1,209 @@
/*
* 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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
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.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 static org.assertj.core.api.Assertions.assertThat;
@SpringBootApplication
@Import(TestChannelBinderConfiguration.class)
public class SleuthBenchmarkingStreamApplication {
private static final Logger log = LoggerFactory.getLogger(SleuthBenchmarkingStreamApplication.class);
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);
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());
log.info("Retrieving the message for tests");
OutputDestination output = context.getBean(OutputDestination.class);
Message<byte[]> message = output.receive(200L);
log.info("Got the message from output");
assertThat(message).isNotNull();
log.info("Message is not null");
assertThat(message.getPayload()).isEqualTo("HELLO".getBytes());
log.info("Payload is HELLO");
String b3 = message.getHeaders().get("b3", String.class);
log.info("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> simple() {
log.info("simple_function");
return new SimpleFunction();
}
@Bean(name = "myFlux")
@ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "reactive_simple")
public Function<Flux<String>, Flux<String>> reactiveSimple() {
log.info("simple_reactive_function");
return new SimpleReactiveFunction();
}
@Bean(name = "myFlux")
@ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "simple_function_with_around")
public Function<Message<String>, Message<String>> simpleFunctionWithAround() {
log.info("simple_function_with_around");
return new SimpleMessageFunction();
}
@Bean(name = "myFlux")
@ConditionalOnProperty(value = "spring.sleuth.nonreactive.function.enabled", havingValue = "true")
public Function<String, String> nonReactiveFunction(ExecutorService executorService) {
log.info("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() {
log.info("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() {
log.info("on last function");
return new SleuthFunction();
}
}
class SimpleFunction implements Function<String, String> {
private static final Logger log = LoggerFactory.getLogger(SimpleFunction.class);
@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();
}
}
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 SimpleMessageFunction implements Function<Message<String>, Message<String>> {
private static final Logger log = LoggerFactory.getLogger(SimpleFunction.class);
@Override
public Message<String> apply(Message<String> input) {
log.info("Hello from message simple [{}]", input.getPayload());
return MessageBuilder.withPayload(input.getPayload().toUpperCase()).build();
}
}
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);
static final Scheduler SCHEDULER = Schedulers.newParallel("sleuthFunction");
@Override
public Flux<String> apply(Flux<String> input) {
return input.doOnEach(signal -> log.info("Got a message"))
.flatMap(s -> Mono.delay(Duration.ofMillis(1), SCHEDULER).map(aLong -> {
log.info("Logging [{}] from flat map", s);
return s.toUpperCase();
}));
}
}

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

@@ -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,17 @@
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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.WebApplicationType;
@@ -33,7 +37,9 @@ import org.springframework.boot.web.reactive.context.ReactiveWebServerInitialize
import org.springframework.cloud.sleuth.instrument.web.SkipPatternProvider;
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 +48,20 @@ 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);
static final Scheduler FOO_SCHEDULER = Schedulers.newParallel("foo");
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")
@@ -60,29 +69,15 @@ public class SleuthBenchmarkingSpringWebFluxApp
return Mono.just("foo");
}
@Bean
Sampler alwaysSampler() {
return Sampler.ALWAYS_SAMPLE;
}
@Bean
SkipPatternProvider patternProvider() {
return () -> Pattern.compile("");
}
@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);
}
@Bean
public SpanHandler spanHandler() {
return new SpanHandler() {
// intentionally anonymous to prevent logging fallback on NOOP
};
return new NettyReactiveWebServerFactory(serverPort == 0 ? SocketUtils.findAvailableTcpPort() : serverPort);
}
@Override
@@ -90,4 +85,40 @@ public class SleuthBenchmarkingSpringWebFluxApp
this.port = event.getWebServer().getPort();
}
@GetMapping("/simple")
public Mono<String> simple() {
return Mono.just("hello").map(String::toUpperCase).doOnNext(s -> log.info("Hello from simple [{}]", s));
}
@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), FOO_SCHEDULER).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), FOO_SCHEDULER).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");
if (traceContext.traceIdString().startsWith("0000000000000000")) {
Assert.state(traceContext.traceIdString().equals("00000000000000004883117762eb9420"), "TraceId must be propagated");
} else {
Assert.state(traceContext.traceIdString().equals("4883117762eb9420"), "TraceId must be propagated");
}
log.info("Assertions passed");
});
}
}

4
benchmarks/src/main/resources/application.yml
View File

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

51
benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/Pair.java Normal file
View File

@@ -0,0 +1,51 @@
/*
* Copyright 2016-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;
public class Pair {
final String key;
final String value;
public Pair(String key, String value) {
this.key = key;
this.value = value;
}
public String asProp() {
return this.key + "=" + this.value;
}
public static Pair of(String key, String value) {
return new Pair(key, value);
}
public static Pair noHook() {
return new Pair("spring.sleuth.reactor.decorate-hooks", "false");
}
public static Pair noSleuth() {
return new Pair("spring.sleuth.enabled", "false");
}
public static Pair onEach() {
return new Pair("spring.sleuth.reactor.decorate-on-each", "true");
}
public static Pair onLast() {
return new Pair("spring.sleuth.reactor.decorate-on-each", "false");
}
}

12
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/ProcessLauncherState.java → benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/ProcessLauncherState.java
View File

@@ -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());
}
}
@@ -120,6 +117,7 @@ public class ProcessLauncherState {
}
public void run() throws Exception {
System.out.println("Running process");
List<String> args = new ArrayList<>(this.args);
args.add(args.size() - this.length, this.mainClass);
if (extraArgs != null) {

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

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

163
benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/SampleTests.java Normal file
View File

@@ -0,0 +1,163 @@
/*
* 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(proxyBeanMethods = false)
@Import(TestChannelBinderConfiguration.class)
static class TestConfiguration {
}
}

17
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/SpringWebFluxOnLastBenchmark.java → benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/TracerImplementation.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2019 the original author or authors.
* Copyright 2016-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.
@@ -14,16 +14,15 @@
* limitations under the License.
*/
package org.springframework.cloud.sleuth.benchmarks.jmh.benchmarks;
package org.springframework.cloud.sleuth.benchmarks.jmh;
public class SpringWebFluxOnLastBenchmark extends SpringWebFluxBenchmarks {
public enum TracerImplementation {
brave;
@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" };
public String toString() {
return this.name();
}
}
}

25
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/AnnotationBenchmarks.java → benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/mvc/AnnotationBenchmarksTests.java
View File

@@ -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,16 +14,18 @@
* 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;
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;
@@ -33,17 +35,19 @@ import org.openjdk.jmh.annotations.Warmup;
import org.springframework.boot.SpringApplication;
import org.springframework.cloud.sleuth.benchmarks.app.mvc.SleuthBenchmarkingSpringApp;
import org.springframework.cloud.sleuth.benchmarks.jmh.TracerImplementation;
import org.springframework.context.ConfigurableApplicationContext;
import static org.assertj.core.api.BDDAssertions.then;
@Measurement(iterations = 5)
@Warmup(iterations = 10)
@Fork(3)
@Measurement(iterations = 10, time = 1)
@Warmup(iterations = 10, time = 1)
@Fork(4)
@BenchmarkMode(Mode.SampleTime)
@OutputTimeUnit(TimeUnit.MICROSECONDS)
@Threads(Threads.MAX)
public class AnnotationBenchmarks {
@Microbenchmark
public class AnnotationBenchmarksTests {
@Benchmark
public void manuallyCreatedSpans(BenchmarkContext context) throws Exception {
@@ -62,11 +66,14 @@ public class AnnotationBenchmarks {
volatile SleuthBenchmarkingSpringApp sleuth;
@Param
private TracerImplementation tracerImplementation;
@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.tracerImplementation.name());
this.sleuth = this.withSleuth.getBean(SleuthBenchmarkingSpringApp.class);
}

81
benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/mvc/AsyncWithSleuthBenchmarksTests.java Normal file
View File

@@ -0,0 +1,81 @@
/*
* 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.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;
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.Threads;
import org.openjdk.jmh.annotations.Warmup;
import org.springframework.boot.SpringApplication;
import org.springframework.cloud.sleuth.benchmarks.app.mvc.SleuthBenchmarkingSpringApp;
import org.springframework.cloud.sleuth.benchmarks.jmh.TracerImplementation;
import org.springframework.context.ConfigurableApplicationContext;
import static org.assertj.core.api.BDDAssertions.then;
@Measurement(iterations = 10, time = 1)
@Warmup(iterations = 10, time = 1)
@Fork(4)
@BenchmarkMode(Mode.SampleTime)
@OutputTimeUnit(TimeUnit.MICROSECONDS)
@Threads(Threads.MAX)
@Microbenchmark
public class AsyncWithSleuthBenchmarksTests {
@Benchmark
public void asyncMethodWithSleuth(BenchmarkContext context) throws Exception {
then(context.app.async().get()).isEqualTo("async");
}
@State(Scope.Benchmark)
public static class BenchmarkContext {
volatile ConfigurableApplicationContext context;
volatile SleuthBenchmarkingSpringApp app;
@Param
private TracerImplementation tracerImplementation;
@Setup
public void setup() {
this.context = new SpringApplication(SleuthBenchmarkingSpringApp.class).run(
"--spring.jmx.enabled=false",
"--spring.application.name=withSleuth_" + this.tracerImplementation.name()
);
this.app = this.context.getBean(SleuthBenchmarkingSpringApp.class);
}
@TearDown
public void clean() {
this.app.clean();
this.context.close();
}
}
}

57
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/AsyncBenchmarks.java → benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/mvc/AsyncWithoutSleuthBenchmarksTests.java
View File

@@ -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,57 +38,39 @@ import org.springframework.context.ConfigurableApplicationContext;
import static org.assertj.core.api.BDDAssertions.then;
@Measurement(iterations = 5)
@Warmup(iterations = 10)
@Fork(3)
@Measurement(iterations = 10, time = 1)
@Warmup(iterations = 10, time = 1)
@Fork(4)
@BenchmarkMode(Mode.SampleTime)
@OutputTimeUnit(TimeUnit.MICROSECONDS)
@Threads(Threads.MAX)
public class AsyncBenchmarks {
@Microbenchmark
public class AsyncWithoutSleuthBenchmarksTests {
@Benchmark
public void asyncMethodWithoutSleuth(BenchmarkContext context) throws Exception {
then(context.untracedAsyncMethodHavingBean.async().get()).isEqualTo("async");
}
@Benchmark
public void asyncMethodWithSleuth(BenchmarkContext context) throws Exception {
then(context.tracedAsyncMethodHavingBean.async().get()).isEqualTo("async");
then(context.app.async().get()).isEqualTo("async");
}
@State(Scope.Benchmark)
public static class BenchmarkContext {
volatile ConfigurableApplicationContext withSleuth;
volatile ConfigurableApplicationContext withoutSleuth;
volatile SleuthBenchmarkingSpringApp tracedAsyncMethodHavingBean;
volatile SleuthBenchmarkingSpringApp untracedAsyncMethodHavingBean;
volatile ConfigurableApplicationContext context;
volatile SleuthBenchmarkingSpringApp app;
@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.context = new SpringApplication(SleuthBenchmarkingSpringApp.class).run(
"--spring.jmx.enabled=false",
"--spring.application.name=withoutSleuth",
"--spring.sleuth.enabled=false",
"--spring.sleuth.async.enabled=false"
);
this.app = this.context.getBean(SleuthBenchmarkingSpringApp.class);
}
@TearDown
public void clean() {
this.tracedAsyncMethodHavingBean.clean();
this.untracedAsyncMethodHavingBean.clean();
this.withSleuth.close();
this.withoutSleuth.close();
this.app.clean();
this.context.close();
}
}

49
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/HttpFilterBenchmarks.java → benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/mvc/HttpFilterBenchmarksTests.java
View File

@@ -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,12 +28,14 @@ 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;
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;
@@ -43,6 +45,8 @@ import org.openjdk.jmh.annotations.Warmup;
import org.springframework.boot.SpringApplication;
import org.springframework.cloud.sleuth.benchmarks.app.mvc.SleuthBenchmarkingSpringApp;
import org.springframework.cloud.sleuth.benchmarks.app.mvc.controller.AsyncSimulationController;
import org.springframework.cloud.sleuth.benchmarks.jmh.TracerImplementation;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.http.MediaType;
import org.springframework.mock.web.MockFilterChain;
@@ -62,17 +66,17 @@ 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 {
@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 +86,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 +110,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();
@@ -133,18 +134,18 @@ public class HttpFilterBenchmarks {
volatile MockMvc mockMvcForUntracedController;
@Param
private TracerImplementation tracerImplementation;
@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.tracerImplementation.name());
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(AsyncSimulationController.class)).build();
this.mockMvcForUntracedController = MockMvcBuilders.standaloneSetup(new VanillaController()).build();
}
@TearDown
@@ -162,8 +163,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);
}

49
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/RestTemplateBenchmark.java → benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/mvc/RestTemplateBenchmarkTests.java
View File

@@ -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,12 +23,14 @@ 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;
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;
@@ -38,6 +40,8 @@ import org.openjdk.jmh.annotations.Warmup;
import org.springframework.boot.SpringApplication;
import org.springframework.cloud.sleuth.benchmarks.app.mvc.SleuthBenchmarkingSpringApp;
import org.springframework.cloud.sleuth.benchmarks.app.mvc.controller.AsyncSimulationController;
import org.springframework.cloud.sleuth.benchmarks.jmh.TracerImplementation;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.test.web.client.MockMvcClientHttpRequestFactory;
import org.springframework.test.web.servlet.MockMvc;
@@ -49,24 +53,22 @@ 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 = 10, time = 1)
@Warmup(iterations = 10, time = 1)
@Fork(4)
@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");
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 {
public void syncEndpointWithSleuth(BenchmarkContext context) throws ServletException, IOException {
then(context.tracedTemplate.getForObject("/foo", String.class)).isEqualTo("foo");
}
@@ -81,20 +83,21 @@ public class RestTemplateBenchmark {
volatile RestTemplate untracedTemplate;
@Param
private TracerImplementation tracerImplementation;
@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))
this.withSleuth = new SpringApplication(SleuthBenchmarkingSpringApp.class).run(
"--spring.jmx.enabled=false",
"--spring.application.name=withSleuth_" + this.tracerImplementation.name()
);
this.mockMvc = MockMvcBuilders.standaloneSetup(this.withSleuth.getBean(AsyncSimulationController.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

27
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/StartupBenchmark.java → benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/mvc/StartupBenchmarkTests.java
View File

@@ -14,24 +14,32 @@
* 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.junit.jupiter.api.Disabled;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
import org.openjdk.jmh.annotations.Level;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Scope;
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;
import org.springframework.cloud.sleuth.benchmarks.jmh.TracerImplementation;
@Measurement(iterations = 5)
@Warmup(iterations = 1)
@Fork(value = 2, warmups = 0)
@BenchmarkMode(Mode.AverageTime)
public class StartupBenchmark {
@Microbenchmark
@Disabled("Process doesn't stop")
public class StartupBenchmarkTests {
@Benchmark
public void withAnnotations(ApplicationState state) throws Exception {
@@ -46,31 +54,30 @@ 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();
}
@State(Scope.Benchmark)
public static class ApplicationState extends ProcessLauncherState {
@Param
private TracerImplementation tracerImplementation;
public ApplicationState() {
super("target", "--server.port=0");
}

196
benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/stream/MicroBenchmarkStreamTests.java Normal file
View File

@@ -0,0 +1,196 @@
/*
* 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.List;
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.sleuth.benchmarks.jmh.Pair;
import org.springframework.cloud.sleuth.benchmarks.jmh.TracerImplementation;
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 static org.assertj.core.api.Assertions.assertThat;
@Measurement(iterations = 10, time = 1)
@Warmup(iterations = 10, time = 1)
@Fork(4)
@BenchmarkMode(Mode.SampleTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Microbenchmark
public class MicroBenchmarkStreamTests {
@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;
@Param
private TracerImplementation tracerImplementation;
@Setup
public void setup() {
this.applicationContext = initContext();
this.input = this.applicationContext.getBean(InputDestination.class);
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.NONE).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() + "_"
+ tracerImplementation.name()));
strings.addAll(Arrays.asList(instrumentation.asParams()));
return strings.toArray(new String[0]);
}
void run() {
sendInputMessage();
assertThatOutputMessageGotReceived();
}
private void assertThatOutputMessageGotReceived() {
// 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).isNotEmpty();
if (b3.startsWith("0000000000000000")) {
assertThat(b3).startsWith("00000000000000004883117762eb9420");
} else {
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 {
// @formatter:off
noSleuthSimple(Pair.noSleuth(), function("simple")),
sleuthSimpleOnHooks(function("simple")),
sleuthSimpleOnEach(function("simple"), Pair.noHook(), Pair.onEach()),
sleuthSimpleOnLast(function("simple"), Pair.noHook(), Pair.onLast()),
sleuthSimpleWithAroundOnHooks(function("simple_function_with_around")),
sleuthSimpleWithAroundOnEach(function("simple_function_with_around"), Pair.noHook(), Pair.onEach()),
sleuthSimpleWithAroundOnLast(function("simple_function_with_around"), Pair.noHook(), Pair.onLast()),
noSleuthReactiveSimple(function("reactive_simple"), Pair.noSleuth()),
sleuthReactiveSimpleOnHooks(function("DECORATE_ON_EACH")),
sleuthReactiveSimpleOnEach(function("DECORATE_ON_EACH"), Pair.noHook(), Pair.onEach(), integrationEnabled());
// @formatter:on
private List<Pair> pairs;
Instrumentation(Pair... pairs) {
this.pairs = Arrays.asList(pairs);
}
String[] asParams() {
return this.pairs.stream().map(p -> "--" + p.asProp()).collect(Collectors.toList()).toArray(new String[0]);
}
static Pair function(String type) {
return Pair.of("spring.sleuth.function.type", type);
}
static Pair integrationEnabled() {
return Pair.of("spring.sleuth.integration.enabled", "true");
}
}
}
@Configuration(proxyBeanMethods = false)
@Import(TestChannelBinderConfiguration.class)
static class TestConfiguration {
}
}

146
benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/webflux/MicroBenchmarkHttpTests.java Normal file
View File

@@ -0,0 +1,146 @@
/*
* 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.ArrayList;
import java.util.Arrays;
import java.util.List;
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.webflux.SleuthBenchmarkingSpringWebFluxApp;
import org.springframework.cloud.sleuth.benchmarks.jmh.Pair;
import org.springframework.cloud.sleuth.benchmarks.jmh.TracerImplementation;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.test.web.reactive.server.WebTestClient;
@Measurement(iterations = 10, time = 1)
@Warmup(iterations = 10, time = 1)
@Fork(4)
@BenchmarkMode(Mode.SampleTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Microbenchmark
public class MicroBenchmarkHttpTests {
@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;
@Param
private TracerImplementation tracerImplementation;
@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() {
String[] defaultArgs = new String[] { "--spring.jmx.enabled=false",
"--spring.application.name=defaultTraceContext" + instrumentation.name() + "_"
+ tracerImplementation.name() };
List<String> list = new ArrayList<>(Arrays.asList(defaultArgs));
list.addAll(Arrays.asList(instrumentation.asParams()));
return list.toArray(new String[0]);
}
void run() {
this.webTestClient.get().uri(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 {
// @formatter:off
noSleuthSimple("/simple", Pair.noSleuth()),
sleuthSimpleOnHooks("/simple"),
sleuthSimpleOnEach("/simple", Pair.noHook(), Pair.onEach()),
sleuthSimpleOnLast("/simple", Pair.noHook(), Pair.onLast()),
noSleuthComplex("/complexNoSleuth", Pair.noSleuth()),
onHooksComplex("/complex"),
onEachComplex("/complex", Pair.noHook(), Pair.onEach()),
onLastComplex("/complex", Pair.noHook(), Pair.onLast());
// @formatter:on
private String url;
private List<Pair> pairs;
Instrumentation(String url, Pair... pairs) {
this.url = url;
this.pairs = Arrays.asList(pairs);
}
String[] asParams() {
return this.pairs.stream().map(p -> "--" + p.asProp()).collect(Collectors.toList()).toArray(new String[0]);
}
}
}
}

53
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/SpringWebFluxBenchmarks.java → benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/webflux/SpringWebFluxBenchmarksTests.java
View File

@@ -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;
@@ -51,32 +52,40 @@ 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.cloud.sleuth.benchmarks.jmh.TracerImplementation;
import org.springframework.context.ConfigurableApplicationContext;
@Measurement(iterations = 5, time = 1)
@Measurement(iterations = 10, time = 1)
@Warmup(iterations = 10, time = 1)
@Fork(3)
@Fork(4)
@BenchmarkMode(Mode.SampleTime)
@OutputTimeUnit(TimeUnit.MICROSECONDS)
@Threads(2)
@State(Scope.Benchmark)
public class SpringWebFluxBenchmarks {
@Microbenchmark
public abstract 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();
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 +96,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 +115,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,9 +139,8 @@ public class SpringWebFluxBenchmarks {
}
protected String[] runArgs() {
return new String[] { "--spring.jmx.enabled=false",
"--spring.application.name=defaultTraceContext",
"--spring.sleuth.enabled=true" };
return new String[] { "--spring.jmx.enabled=false", "--spring.application.name=defaultTraceContext",
TracerImplementation.brave.toString(), "--spring.sleuth.enabled=true" };
}
@TearDown
@@ -171,8 +175,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);
}
}

23
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/WithOutSleuthSpringWebFluxBenchmarks.java → benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/webflux/WithOutReactorSleuthSpringWebFluxBenchmarksTests.java
View File

@@ -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,31 +14,36 @@
* 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;
import org.openjdk.jmh.runner.options.OptionsBuilder;
import org.springframework.cloud.sleuth.benchmarks.jmh.TracerImplementation;
/**
* @author alvin
*/
public class WithOutSleuthSpringWebFluxBenchmarks extends SpringWebFluxBenchmarks {
@Microbenchmark
public class WithOutReactorSleuthSpringWebFluxBenchmarksTests extends SpringWebFluxBenchmarksTests {
public static void main(String[] args) throws RunnerException {
Options opt = new OptionsBuilder().include(
".*" + WithOutSleuthSpringWebFluxBenchmarks.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",
"--spring.sleuth.enabled=false" };
return new String[] { "--spring.jmx.enabled=false", "--spring.application.name=defaultTraceContext",
TracerImplementation.brave.toString(), "--spring.sleuth.enabled=true",
"--spring.sleuth.reactor.enabled=false"
};
}
@Override

23
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/WithOutReactorSleuthSpringWebFluxBenchmarks.java → benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/webflux/WithOutSleuthSpringWebFluxBenchmarksTests.java
View File

@@ -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,34 +14,33 @@
* 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;
import org.openjdk.jmh.runner.options.OptionsBuilder;
import org.springframework.cloud.sleuth.benchmarks.jmh.TracerImplementation;
/**
* @author alvin
*/
public class WithOutReactorSleuthSpringWebFluxBenchmarks extends SpringWebFluxBenchmarks {
@Microbenchmark
public class WithOutSleuthSpringWebFluxBenchmarksTests extends SpringWebFluxBenchmarksTests {
public static void main(String[] args) throws RunnerException {
Options opt = new OptionsBuilder().include(
".*" + WithOutReactorSleuthSpringWebFluxBenchmarks.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",
"--spring.sleuth.enabled=true", "--spring.sleuth.reactor.enabled=false"
};
return new String[] { "--spring.jmx.enabled=false", "--spring.application.name=defaultTraceContext",
TracerImplementation.brave.toString(), "--spring.sleuth.enabled=false" };
}
@Override

17
docs/src/main/asciidoc/spring-cloud-sleuth.adoc
View File

@@ -1526,19 +1526,14 @@ To turn off this feature, set the `spring.sleuth.quartz.enabled` property to `fa
=== Project Reactor
==== From Spring Cloud Sleuth 2.2.8 (inclusive)
With the new Reactor https://github.com/reactor/reactor-core/pull/2566[queue wrapping mechanism] (Reactor 3.3.14) we're instrumenting the way threads are switched by Reactor. You should observe significant improvement in performance. In order to disable this feature you have to set the `spring.sleuth.reactor.decorate-hooks` option to `false`. You'll fall back to the previous instrumentation mode mechanism.
==== To Spring Cloud Sleuth 2.2.8 (exclusive)
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.
== Configuration properties
To see the list of all Sleuth related configuration properties please check link:appendix.html[the Appendix page].
== Running examples
You can see the running examples deployed in the https://run.pivotal.io/[Pivotal Web Services].
Check them out at the following links:
* https://docssleuth-zipkin-server.cfapps.io/[Zipkin for apps presented in the samples to the top]. First make
a request to https://docssleuth-service1.cfapps.io/start[Service 1] and then check out the trace in Zipkin.
* https://docsbrewing-zipkin-server.cfapps.io/[Zipkin for Brewery on PWS], its https://github.com/spring-cloud-samples/brewery[Github Code].
Ensure that you've picked the lookback period of 7 days. If there are no traces, go to https://docsbrewing-presenting.cfapps.io/[Presenting application]
and order some beers. Then check Zipkin for traces.

2
pom.xml
View File

@@ -29,7 +29,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-build</artifactId>
<version>2.3.2.RELEASE</version>
<version>2.3.3.BUILD-SNAPSHOT</version>
<relativePath/>
<!-- lookup parent from repository -->
</parent>

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

@@ -129,6 +129,29 @@ public abstract class ReactorSleuth {
});
}
public static Function<Runnable, Runnable> scopePassingOnScheduleHook(
ConfigurableApplicationContext springContext) {
LazyBean<CurrentTraceContext> lazyCurrentTraceContext = LazyBean
.create(springContext, CurrentTraceContext.class);
return delegate -> {
if (springContext.isActive()) {
final CurrentTraceContext currentTraceContext = lazyCurrentTraceContext
.get();
if (currentTraceContext == null) {
return delegate;
}
final TraceContext traceContext = currentTraceContext.get();
return () -> {
try (CurrentTraceContext.Scope scope = currentTraceContext
.maybeScope(traceContext)) {
delegate.run();
}
};
}
return delegate;
};
}
private static <T> Context context(CoreSubscriber<? super T> sub) {
try {
return sub.currentContext();

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

@@ -33,11 +33,21 @@ public class SleuthReactorProperties {
*/
private boolean enabled = true;
/**
* When true uses the new decorate hooks feature from Project Reactor. Should allow
* the feature set of {@link SleuthReactorProperties#decorateOnEach} with the least
* impact on the performance.
*/
private boolean decorateHooks = true;
/**
* When true decorates on each operator, will be less performing, but logging will
* always contain the tracing entries in each operator. When false decorates on last
* operator, will be more performing, but logging might not always contain the tracing
* entries.
*
* If {@link SleuthReactorProperties#decorateHooks} is used, this decoration mode will
* NOT be used.
*/
private boolean decorateOnEach = true;
@@ -49,6 +59,14 @@ public class SleuthReactorProperties {
this.enabled = enabled;
}
public boolean isDecorateHooks() {
return this.decorateHooks;
}
public void setDecorateHooks(boolean decorateHooks) {
this.decorateHooks = decorateHooks;
}
public boolean isDecorateOnEach() {
return this.decorateOnEach;
}

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

@@ -16,9 +16,18 @@
package org.springframework.cloud.sleuth.instrument.reactor;
import java.io.Closeable;
import java.io.IOException;
import java.util.AbstractQueue;
import java.util.Iterator;
import java.util.Queue;
import java.util.function.Function;
import javax.annotation.PreDestroy;
import brave.Tracing;
import brave.propagation.CurrentTraceContext;
import brave.propagation.TraceContext;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.Hooks;
@@ -37,15 +46,16 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.context.scope.refresh.RefreshScope;
import org.springframework.cloud.context.scope.refresh.RefreshScopeRefreshedEvent;
import org.springframework.cloud.sleuth.instrument.async.TraceableScheduledExecutorService;
import org.springframework.cloud.sleuth.instrument.web.TraceWebFluxAutoConfiguration;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.util.ReflectionUtils;
import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.scopePassingSpanOperator;
import static org.springframework.cloud.sleuth.instrument.reactor.TraceReactorAutoConfiguration.SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY;
import static org.springframework.cloud.sleuth.instrument.reactor.TraceReactorAutoConfiguration.TraceReactorConfiguration.SLEUTH_TRACE_REACTOR_KEY;
/**
@@ -77,6 +87,8 @@ public class TraceReactorAutoConfiguration {
private static final Log log = LogFactory.getLog(TraceReactorConfiguration.class);
static final boolean IS_QUEUE_WRAPPER_ON_THE_CLASSPATH = isQueueWrapperOnTheClasspath();
@Autowired
ConfigurableApplicationContext springContext;
@@ -87,6 +99,13 @@ public class TraceReactorAutoConfiguration {
}
SleuthReactorProperties reactorProperties = this.springContext
.getBean(SleuthReactorProperties.class);
if (TraceReactorAutoConfiguration.TraceReactorConfiguration.IS_QUEUE_WRAPPER_ON_THE_CLASSPATH
&& reactorProperties.isDecorateHooks()) {
if (log.isTraceEnabled()) {
log.trace("Resetting queue wrapper instrumentation");
}
Hooks.removeQueueWrapper(SLEUTH_TRACE_REACTOR_KEY);
}
if (reactorProperties.isDecorateOnEach()) {
if (log.isTraceEnabled()) {
log.trace("Resetting onEach operator instrumentation");
@@ -99,8 +118,11 @@ public class TraceReactorAutoConfiguration {
}
Hooks.resetOnLastOperator(SLEUTH_TRACE_REACTOR_KEY);
}
Schedulers
.removeExecutorServiceDecorator(SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY);
}
private static boolean isQueueWrapperOnTheClasspath() {
return ReflectionUtils.findMethod(Hooks.class, "addQueueWrapper",
String.class, Function.class) != null;
}
@Bean
@@ -152,12 +174,23 @@ class HooksRefresher implements ApplicationListener<RefreshScopeRefreshedEvent>
}
Hooks.resetOnEachOperator(SLEUTH_TRACE_REACTOR_KEY);
Hooks.resetOnLastOperator(SLEUTH_TRACE_REACTOR_KEY);
if (this.reactorProperties.isDecorateOnEach()) {
Hooks.removeQueueWrapper(SLEUTH_TRACE_REACTOR_KEY);
if (this.reactorProperties.isDecorateHooks()
&& TraceReactorAutoConfiguration.TraceReactorConfiguration.IS_QUEUE_WRAPPER_ON_THE_CLASSPATH) {
if (log.isTraceEnabled()) {
log.trace("Adding queue wrapper instrumentation");
}
HookRegisteringBeanDefinitionRegistryPostProcessor.addQueueWrapper(context);
}
else if (this.reactorProperties.isDecorateOnEach()) {
if (log.isTraceEnabled()) {
log.trace("Decorating onEach operator instrumentation");
}
Hooks.onEachOperator(SLEUTH_TRACE_REACTOR_KEY,
scopePassingSpanOperator(this.context));
Schedulers.onScheduleHook(
TraceReactorAutoConfiguration.SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY,
ReactorSleuth.scopePassingOnScheduleHook(this.context));
}
else {
if (log.isTraceEnabled()) {
@@ -171,7 +204,7 @@ class HooksRefresher implements ApplicationListener<RefreshScopeRefreshedEvent>
}
class HookRegisteringBeanDefinitionRegistryPostProcessor
implements BeanDefinitionRegistryPostProcessor {
implements BeanDefinitionRegistryPostProcessor, Closeable {
private static final Log log = LogFactory
.getLog(HookRegisteringBeanDefinitionRegistryPostProcessor.class);
@@ -194,27 +227,165 @@ class HookRegisteringBeanDefinitionRegistryPostProcessor
static void setupHooks(ConfigurableApplicationContext springContext) {
ConfigurableEnvironment environment = springContext.getEnvironment();
boolean decorateOnEach = environment.getProperty(
"spring.sleuth.reactor.decorate-on-each", Boolean.class, true);
if (decorateOnEach) {
if (log.isTraceEnabled()) {
log.trace("Decorating onEach operator instrumentation");
}
Hooks.onEachOperator(SLEUTH_TRACE_REACTOR_KEY,
scopePassingSpanOperator(springContext));
Boolean decorateHooks = environment
.getProperty("spring.sleuth.reactor.decorate-hooks", Boolean.class);
if (wrapperNotOnClasspathButPropertyHasValue(decorateHooks)) {
log.warn(
"You have explicitly set the decorate hooks option but you're using an old version of Reactor. Please upgrade to the latest Boot version (at least 2.3.9.RELEASE). Will fall back to the previous reactor instrumentation mode");
}
else {
if (log.isTraceEnabled()) {
log.trace("Decorating onLast operator instrumentation");
}
Hooks.onLastOperator(SLEUTH_TRACE_REACTOR_KEY,
scopePassingSpanOperator(springContext));
decorateHooks = decorateHooks != null ? decorateHooks : Boolean.TRUE;
}
Schedulers.setExecutorServiceDecorator(
if (wrapperOnClasspathHooksPropertyTurnedOn(decorateHooks)) {
if (log.isTraceEnabled()) {
log.trace("Adding queue wrapper instrumentation");
}
addQueueWrapper(springContext);
}
else {
boolean decorateOnEach = environment.getProperty(
"spring.sleuth.reactor.decorate-on-each", Boolean.class, true);
if (decorateOnEach) {
if (log.isTraceEnabled()) {
log.trace("Decorating onEach operator instrumentation");
}
Hooks.onEachOperator(SLEUTH_TRACE_REACTOR_KEY,
scopePassingSpanOperator(springContext));
}
else {
if (log.isTraceEnabled()) {
log.trace("Decorating onLast operator instrumentation");
}
Hooks.onLastOperator(SLEUTH_TRACE_REACTOR_KEY,
scopePassingSpanOperator(springContext));
}
}
decorateScheduler(springContext);
}
private static boolean wrapperOnClasspathHooksPropertyTurnedOn(Boolean decorateHooks) {
return Boolean.TRUE.equals(decorateHooks)
&& TraceReactorAutoConfiguration.TraceReactorConfiguration.IS_QUEUE_WRAPPER_ON_THE_CLASSPATH;
}
private static boolean wrapperNotOnClasspathButPropertyHasValue(Boolean decorateHooks) {
return !TraceReactorAutoConfiguration.TraceReactorConfiguration.IS_QUEUE_WRAPPER_ON_THE_CLASSPATH
&& decorateHooks != null;
}
static void addQueueWrapper(ConfigurableApplicationContext springContext) {
Hooks.addQueueWrapper(SLEUTH_TRACE_REACTOR_KEY,
queue -> traceQueue(springContext, queue));
}
@Override
public void close() throws IOException {
if (log.isTraceEnabled()) {
log.trace("Cleaning up hooks");
}
Hooks.resetOnEachOperator(SLEUTH_TRACE_REACTOR_KEY);
Hooks.resetOnLastOperator(SLEUTH_TRACE_REACTOR_KEY);
Hooks.removeQueueWrapper(SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY);
Schedulers.resetOnScheduleHook(SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY);
Schedulers.resetOnScheduleHook(
TraceReactorAutoConfiguration.SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY);
}
private static void decorateScheduler(ConfigurableApplicationContext springContext) {
Schedulers.onScheduleHook(
TraceReactorAutoConfiguration.SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY,
(scheduler,
scheduledExecutorService) -> new TraceableScheduledExecutorService(
springContext, scheduledExecutorService));
ReactorSleuth.scopePassingOnScheduleHook(springContext));
}
private static Queue<?> traceQueue(ConfigurableApplicationContext springContext,
Queue<?> queue) {
if (!springContext.isActive()) {
return queue;
}
CurrentTraceContext currentTraceContext = springContext
.getBean(CurrentTraceContext.class);
@SuppressWarnings("unchecked")
Queue envelopeQueue = queue;
return new AbstractQueue<Object>() {
@Override
public int size() {
return envelopeQueue.size();
}
@Override
public boolean offer(Object o) {
TraceContext traceContext = currentTraceContext.get();
return envelopeQueue.offer(new Envelope(o, traceContext));
}
@Override
public Object poll() {
Object object = envelopeQueue.poll();
if (object == null) {
return null;
}
else if (object instanceof Envelope) {
Envelope envelope = (Envelope) object;
restoreTheContext(envelope);
return envelope.body;
}
return object;
}
private void restoreTheContext(Envelope envelope) {
if (envelope.traceContext != null) {
currentTraceContext.maybeScope(envelope.traceContext);
}
}
@Override
public Object peek() {
Object peek = queue.peek();
if (peek instanceof Envelope) {
Envelope envelope = (Envelope) peek;
restoreTheContext(envelope);
return (envelope).body;
}
return peek;
}
@Override
@SuppressWarnings("unchecked")
public Iterator<Object> iterator() {
Iterator<?> iterator = queue.iterator();
return new Iterator<Object>() {
@Override
public boolean hasNext() {
return iterator.hasNext();
}
@Override
public Object next() {
Object next = iterator.next();
if (next instanceof Envelope) {
Envelope envelope = (Envelope) next;
restoreTheContext(envelope);
return (envelope).body;
}
return next;
}
};
}
};
}
static class Envelope {
final Object body;
final TraceContext traceContext;
Envelope(Object body, TraceContext traceContext) {
this.body = body;
this.traceContext = traceContext;
}
}
}

118
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceHandlerFunctionAdapterBeanPostProcessor.java Normal file
View File

@@ -0,0 +1,118 @@
/*
* 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.instrument.web;
import java.util.concurrent.atomic.AtomicReference;
import brave.Span;
import brave.propagation.CurrentTraceContext;
import reactor.core.publisher.Mono;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.web.reactive.HandlerAdapter;
import org.springframework.web.reactive.HandlerResult;
import org.springframework.web.reactive.function.server.HandlerFunction;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.support.HandlerFunctionAdapter;
import org.springframework.web.server.ServerWebExchange;
class TraceHandlerFunctionAdapterBeanPostProcessor implements BeanPostProcessor {
private final BeanFactory beanFactory;
TraceHandlerFunctionAdapterBeanPostProcessor(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName)
throws BeansException {
if (bean instanceof HandlerFunctionAdapter) {
return new TraceHandlerAdapter((HandlerAdapter) bean, this.beanFactory);
}
return bean;
}
private static final class TraceHandlerAdapter implements HandlerAdapter {
private final BeanFactory beanFactory;
private final HandlerAdapter delegate;
private TraceHandlerAdapter(HandlerAdapter delegate, BeanFactory beanFactory) {
this.delegate = delegate;
this.beanFactory = beanFactory;
}
@Override
public boolean supports(Object handler) {
return this.delegate.supports(handler);
}
@Override
public Mono<HandlerResult> handle(ServerWebExchange exchange, Object handler) {
HandlerFunction<?> handlerFunction = (HandlerFunction<?>) handler;
TraceHandlerFunction traceHandlerFunction = new TraceHandlerFunction(
handlerFunction, this.beanFactory);
return this.delegate.handle(exchange, traceHandlerFunction);
}
}
private static final class TraceHandlerFunction implements HandlerFunction {
private final HandlerFunction<?> delegate;
private final BeanFactory beanFactory;
private CurrentTraceContext currentTraceContext;
private TraceHandlerFunction(HandlerFunction<?> delegate,
BeanFactory beanFactory) {
this.delegate = delegate;
this.beanFactory = beanFactory;
}
@Override
public Mono<?> handle(ServerRequest serverRequest) {
AtomicReference<CurrentTraceContext.Scope> scope = new AtomicReference<>();
return Mono.just(scope).doFirst(() -> serverRequest
.attribute(TraceWebFilter.TRACE_REQUEST_ATTR)
.ifPresent(span -> scope.set(
currentTraceContext().maybeScope(((Span) span).context()))))
.flatMap(r -> this.delegate.handle(serverRequest))
.doFinally(signalType -> {
CurrentTraceContext.Scope spanInScope = scope.get();
if (spanInScope != null) {
spanInScope.close();
}
});
}
private CurrentTraceContext currentTraceContext() {
if (this.currentTraceContext == null) {
this.currentTraceContext = this.beanFactory
.getBean(CurrentTraceContext.class);
}
return this.currentTraceContext;
}
}
}

40
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFilter.java
View File

@@ -25,6 +25,7 @@ import brave.http.HttpServerHandler;
import brave.http.HttpServerRequest;
import brave.http.HttpServerResponse;
import brave.http.HttpTracing;
import brave.propagation.CurrentTraceContext;
import brave.propagation.TraceContext;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -79,11 +80,13 @@ public final class TraceWebFilter implements WebFilter, Ordered {
private final BeanFactory beanFactory;
Tracer tracer;
private Tracer tracer;
HttpServerHandler<HttpServerRequest, HttpServerResponse> handler;
private HttpServerHandler<HttpServerRequest, HttpServerResponse> handler;
SleuthWebProperties webProperties;
private SleuthWebProperties webProperties;
private CurrentTraceContext currentTraceContext;
TraceWebFilter(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
@@ -94,7 +97,7 @@ public final class TraceWebFilter implements WebFilter, Ordered {
}
@SuppressWarnings("unchecked")
HttpServerHandler<HttpServerRequest, HttpServerResponse> handler() {
private HttpServerHandler<HttpServerRequest, HttpServerResponse> handler() {
if (this.handler == null) {
this.handler = HttpServerHandler
.create(this.beanFactory.getBean(HttpTracing.class));
@@ -102,32 +105,40 @@ public final class TraceWebFilter implements WebFilter, Ordered {
return this.handler;
}
Tracer tracer() {
private Tracer tracer() {
if (this.tracer == null) {
this.tracer = this.beanFactory.getBean(HttpTracing.class).tracing().tracer();
}
return this.tracer;
}
SleuthWebProperties sleuthWebProperties() {
private SleuthWebProperties sleuthWebProperties() {
if (this.webProperties == null) {
this.webProperties = this.beanFactory.getBean(SleuthWebProperties.class);
}
return this.webProperties;
}
private CurrentTraceContext currentTraceContext() {
if (this.currentTraceContext == null) {
this.currentTraceContext = this.beanFactory
.getBean(CurrentTraceContext.class);
}
return this.currentTraceContext;
}
@Override
public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
String uri = exchange.getRequest().getPath().pathWithinApplication().value();
if (log.isDebugEnabled()) {
log.debug("Received a request to uri [" + uri + "]");
}
Mono<Void> source = chain.filter(exchange);
boolean tracePresent = tracer().currentSpan() != null;
if (tracePresent) {
// clear any previous trace
tracer().withSpanInScope(null); // TODO: dangerous and also allocates stuff
}
if (log.isDebugEnabled()) {
log.debug("Received a request to uri [" + uri + "]");
}
return new MonoWebFilterTrace(source, exchange, tracePresent, this);
}
@@ -150,11 +161,14 @@ public final class TraceWebFilter implements WebFilter, Ordered {
final boolean initialTracePresent;
final CurrentTraceContext currentTraceContext;
MonoWebFilterTrace(Mono<? extends Void> source, ServerWebExchange exchange,
boolean initialTracePresent, TraceWebFilter parent) {
super(source);
this.tracer = parent.tracer();
this.handler = parent.handler();
this.currentTraceContext = parent.currentTraceContext();
this.exchange = exchange;
this.attrSpan = exchange.getAttribute(TRACE_REQUEST_ATTR);
this.initialTracePresent = initialTracePresent;
@@ -163,8 +177,12 @@ public final class TraceWebFilter implements WebFilter, Ordered {
@Override
public void subscribe(CoreSubscriber<? super Void> subscriber) {
Context context = contextWithoutInitialSpan(subscriber.currentContext());
this.source.subscribe(new WebFilterTraceSubscriber(subscriber, context,
findOrCreateSpan(context), this));
Span span = findOrCreateSpan(context);
try (CurrentTraceContext.Scope scope = this.currentTraceContext
.maybeScope(span.context())) {
this.source.subscribe(
new WebFilterTraceSubscriber(subscriber, context, span, this));
}
}
private Context contextWithoutInitialSpan(Context context) {

6
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/TraceWebFluxAutoConfiguration.java
View File

@@ -48,4 +48,10 @@ public class TraceWebFluxAutoConfiguration {
return new TraceWebFilter(beanFactory);
}
@Bean
public TraceHandlerFunctionAdapterBeanPostProcessor traceHandlerFunctionAdapterBeanPostProcessor(
BeanFactory beanFactory) {
return new TraceHandlerFunctionAdapterBeanPostProcessor(beanFactory);
}
}

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

@@ -92,6 +92,7 @@ public class SleuthSpanCreatorAspectFluxTests {
public void setup() {
this.spans.clear();
this.testBean.reset();
tracer.withSpanInScope(null);
}
@Test

1
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectTests.java
View File

@@ -56,6 +56,7 @@ public class SleuthSpanCreatorAspectTests {
@Before
public void setup() {
this.spans.clear();
tracer.withSpanInScope(null);
}
@Test

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

@@ -44,7 +44,8 @@ public final class TraceReactorAutoConfigurationAccessorConfiguration {
}
Hooks.resetOnEachOperator(SLEUTH_TRACE_REACTOR_KEY);
Hooks.resetOnLastOperator(SLEUTH_TRACE_REACTOR_KEY);
Schedulers.removeExecutorServiceDecorator(SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY);
Hooks.removeQueueWrapper(SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY);
Schedulers.resetOnScheduleHook(SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY);
}
public static void setup(ConfigurableApplicationContext context) {

1
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/IgnoreAutoConfiguredSkipPatternsIntegrationTests.java
View File

@@ -62,6 +62,7 @@ public class IgnoreAutoConfiguredSkipPatternsIntegrationTests {
@After
public void clearSpans() {
this.spans.clear();
tracer.withSpanInScope(null);
}
@Test

2
tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/Issue866Configuration.java
View File

@@ -40,7 +40,7 @@ public class Issue866Configuration {
public static TestHook hook;
@Bean
HookRegisteringBeanDefinitionRegistryPostProcessor overridingProcessorForTests(
static HookRegisteringBeanDefinitionRegistryPostProcessor overridingProcessorForTests(
ConfigurableApplicationContext context) {
log.info(
"Registering a HookRegisteringBeanDefinitionRegistryPostProcessor for context ["

6
tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberSpringBootTests.java
View File

@@ -25,6 +25,7 @@ import brave.propagation.TraceContext;
import brave.sampler.Sampler;
import org.awaitility.Awaitility;
import org.junit.Test;
import org.junit.jupiter.api.BeforeAll;
import org.junit.runner.RunWith;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
@@ -58,6 +59,11 @@ public class ScopePassingSpanSubscriberSpringBootTests {
TraceContext context2 = TraceContext.newBuilder().traceId(1).spanId(2).sampled(true)
.build();
@BeforeAll
static void setup() {
TraceReactorAutoConfigurationAccessorConfiguration.close();
}
@Test
public void should_pass_tracing_info_when_using_reactor() {
final AtomicReference<TraceContext> spanInOperation = new AtomicReference<>();

3
tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java
View File

@@ -148,7 +148,8 @@ public class ScopePassingSpanSubscriberTests {
// prevent should_not_scope_scalar_subscribe from being interfered with.
Hooks.resetOnEachOperator(SLEUTH_TRACE_REACTOR_KEY);
Hooks.resetOnLastOperator(SLEUTH_TRACE_REACTOR_KEY);
Schedulers.removeExecutorServiceDecorator(SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY);
Hooks.removeQueueWrapper(SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY);
Schedulers.resetOnScheduleHook(SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY);
}
@After

16
tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java
View File

@@ -16,16 +16,13 @@
package org.springframework.cloud.sleuth.instrument.reactor;
import java.io.IOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.Hooks;
import reactor.core.scheduler.Schedulers;
import org.springframework.context.ConfigurableApplicationContext;
import static org.springframework.cloud.sleuth.instrument.reactor.TraceReactorAutoConfiguration.SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY;
import static org.springframework.cloud.sleuth.instrument.reactor.TraceReactorAutoConfiguration.TraceReactorConfiguration.SLEUTH_TRACE_REACTOR_KEY;
/**
* @author Marcin Grzejszczak
*/
@@ -42,9 +39,12 @@ public final class TraceReactorAutoConfigurationAccessorConfiguration {
if (log.isTraceEnabled()) {
log.trace("Cleaning up hooks");
}
Hooks.resetOnEachOperator(SLEUTH_TRACE_REACTOR_KEY);
Hooks.resetOnLastOperator(SLEUTH_TRACE_REACTOR_KEY);
Schedulers.removeExecutorServiceDecorator(SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY);
try {
new HookRegisteringBeanDefinitionRegistryPostProcessor(null).close();
}
catch (IOException e) {
throw new IllegalStateException(e);
}
}
public static void setup(ConfigurableApplicationContext context) {

39
tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/sample/FlatMapTests.java
View File

@@ -30,6 +30,7 @@ import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.BeforeEach;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
@@ -74,14 +75,35 @@ public class FlatMapTests {
Issue866Configuration.hook = null;
}
@BeforeEach
void before() {
TraceReactorAutoConfigurationAccessorConfiguration.close();
}
@Test
public void should_work_with_flat_maps() {
public void should_work_with_flat_maps_on_hooks_instrumentation() {
// given
ConfigurableApplicationContext context = new SpringApplicationBuilder(
FlatMapTests.TestConfiguration.class, Issue866Configuration.class)
.web(WebApplicationType.REACTIVE)
.properties("server.port=0", "spring.jmx.enabled=false",
"spring.application.name=TraceWebFluxTests",
"spring.application.name=TraceWebFluxOnHooksTests",
"security.basic.enabled=false",
"management.security.enabled=false")
.run();
assertReactorTracing(context);
}
@Test
public void should_work_with_flat_maps_on_each_operator_instrumentation() {
// given
ConfigurableApplicationContext context = new SpringApplicationBuilder(
FlatMapTests.TestConfiguration.class, Issue866Configuration.class)
.web(WebApplicationType.REACTIVE)
.properties("server.port=0", "spring.jmx.enabled=false",
"spring.sleuth.reactor.decorate-hooks=false",
"spring.sleuth.reactor.decorate-on-each=true",
"spring.application.name=TraceWebFluxOnEachTests",
"security.basic.enabled=false",
"management.security.enabled=false")
.run();
@@ -95,20 +117,23 @@ public class FlatMapTests {
FlatMapTests.TestConfiguration.class, Issue866Configuration.class)
.web(WebApplicationType.REACTIVE)
.properties("server.port=0", "spring.jmx.enabled=false",
"spring.sleuth.reactor.decorate-hooks=false",
"spring.sleuth.reactor.decorate-on-each=false",
"spring.application.name=TraceWebFlux2Tests",
"spring.application.name=TraceWebFluxOnLastTests",
"security.basic.enabled=false",
"management.security.enabled=false")
.run();
assertReactorTracing(context);
try {
System.setProperty("spring.sleuth.reactor.decorate-on-each", "true");
System.setProperty("spring.sleuth.reactor.decorate-hooks", "false");
System.setProperty("spring.sleuth.reactor.decorate-on-each", "false");
// trigger context refreshed
context.getBean(ContextRefresher.class).refresh();
assertReactorTracing(context);
}
finally {
System.clearProperty("spring.sleuth.reactor.decorate-hooks");
System.clearProperty("spring.sleuth.reactor.decorate-on-each");
}
}
@@ -196,11 +221,11 @@ public class FlatMapTests {
RouterFunction<ServerResponse> handlers(Tracer tracer,
RequestSender requestSender) {
return route(GET("/noFlatMap"), request -> {
LOGGER.info("noFlatMap");
LOGGER.info("noFlatMap [" + request + "]");
Flux<Integer> one = requestSender.getAll().map(String::length);
return ServerResponse.ok().body(one, Integer.class);
}).andRoute(GET("/withFlatMap"), request -> {
LOGGER.info("withFlatMap");
LOGGER.info("withFlatMap [" + request + "]");
Flux<Integer> one = requestSender.getAll().map(String::length);
Flux<Integer> response = one
.flatMap(size -> requestSender.getAll().doOnEach(
@@ -211,7 +236,7 @@ public class FlatMapTests {
});
return ServerResponse.ok().body(response, Integer.class);
}).andRoute(GET("/foo"), request -> {
LOGGER.info("foo");
LOGGER.info("foo [" + request + "]");
this.spanInFoo = tracer.currentSpan();
return ServerResponse.ok().body(Flux.just(1), Integer.class);
});

2
tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/resources/application.yml
View File

@@ -1,3 +1,3 @@
logging.level.org.springframework.cloud: DEBUG
logging.level.org.springframework.cloud: TRACE
logging.level.com.netflix.discovery.InstanceInfoReplicator: ERROR
logging.level.org.springframework.cloud.sleuth.instrument.web.client.feign: TRACE