From c2cd689ec7aa0193d44354d8742e2fda82cf140b Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Thu, 6 Feb 2020 13:36:57 +0800 Subject: [PATCH] Reduces the amount of times getBean is invoked from reactor-netty (#1551) This consolidates access to the `HttpTracing` bean so that the overhead of netty `HttpClient` is reduced. It also pulls the reactor-netty test into its own file. --- .../instrument/reactor/ReactorSleuth.java | 5 +- .../client/HttpClientBeanPostProcessor.java | 101 ++++++------- .../TraceWebClientAutoConfiguration.java | 4 +- .../reactor => internal}/LazyBean.java | 14 +- ...ReactorNettyHttpClientSpringBootTests.java | 138 ++++++++++++++++++ .../client/integration/WebClientTests.java | 34 ----- .../reactor => internal}/LazyBeanTests.java | 2 +- 7 files changed, 198 insertions(+), 100 deletions(-) rename spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/{instrument/reactor => internal}/LazyBean.java (83%) create mode 100644 spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientSpringBootTests.java rename spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/{instrument/reactor => internal}/LazyBeanTests.java (95%) diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java index 7f095b779..fa013b7c8 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java @@ -30,6 +30,7 @@ import reactor.core.Scannable; import reactor.core.publisher.Operators; import reactor.util.context.Context; +import org.springframework.cloud.sleuth.internal.LazyBean; import org.springframework.context.ConfigurableApplicationContext; /** @@ -70,8 +71,8 @@ public abstract class ReactorSleuth { // keep a reference outside the lambda so that any caching will be visible to // all publishers - LazyBean lazyCurrentTraceContext = new LazyBean<>( - springContext, CurrentTraceContext.class); + LazyBean lazyCurrentTraceContext = LazyBean + .create(springContext, CurrentTraceContext.class); return Operators.liftPublisher((p, sub) -> { // We don't scope scalar results as they happen in an instant. This prevents diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java index dcf21694e..5558b5e99 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java @@ -33,26 +33,29 @@ import reactor.netty.http.client.HttpClientRequest; import reactor.netty.http.client.HttpClientResponse; import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.cloud.sleuth.internal.LazyBean; +import org.springframework.context.ConfigurableApplicationContext; class HttpClientBeanPostProcessor implements BeanPostProcessor { - private final BeanFactory beanFactory; + final ConfigurableApplicationContext springContext; - HttpClientBeanPostProcessor(BeanFactory beanFactory) { - this.beanFactory = beanFactory; + HttpClientBeanPostProcessor(ConfigurableApplicationContext springContext) { + this.springContext = springContext; } @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + LazyBean httpTracing = LazyBean.create(this.springContext, + HttpTracing.class); if (bean instanceof HttpClient) { - return ((HttpClient) bean).mapConnect(new TracingMapConnect(this.beanFactory)) - .doOnRequest(TracingDoOnRequest.create(this.beanFactory)) - .doOnRequestError(TracingDoOnErrorRequest.create(this.beanFactory)) - .doOnResponse(TracingDoOnResponse.create(this.beanFactory)) - .doOnResponseError(TracingDoOnErrorResponse.create(this.beanFactory)); + return ((HttpClient) bean).mapConnect(new TracingMapConnect(httpTracing)) + .doOnRequest(TracingDoOnRequest.create(httpTracing)) + .doOnRequestError(TracingDoOnErrorRequest.create(httpTracing)) + .doOnResponse(TracingDoOnResponse.create(httpTracing)) + .doOnResponseError(TracingDoOnErrorResponse.create(httpTracing)); } return bean; } @@ -60,12 +63,12 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { private static class TracingMapConnect implements BiFunction, Bootstrap, Mono> { - private final BeanFactory beanFactory; + final LazyBean httpTracing; - private Tracer tracer; + Tracer tracer; - TracingMapConnect(BeanFactory beanFactory) { - this.beanFactory = beanFactory; + TracingMapConnect(LazyBean httpTracing) { + this.httpTracing = httpTracing; } @Override @@ -77,7 +80,7 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { private Tracer tracer() { if (this.tracer == null) { - this.tracer = this.beanFactory.getBean(Tracer.class); + this.tracer = this.httpTracing.get().tracing().tracer(); } return this.tracer; } @@ -87,39 +90,30 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { private static class TracingDoOnRequest implements BiConsumer { - final BeanFactory beanFactory; - - HttpTracing httpTracing; + final LazyBean httpTracing; List propagationKeys; HttpClientHandler handler; - TracingDoOnRequest(BeanFactory beanFactory) { - this.beanFactory = beanFactory; + TracingDoOnRequest(LazyBean httpTracing) { + this.httpTracing = httpTracing; } - static TracingDoOnRequest create(BeanFactory beanFactory) { - return new TracingDoOnRequest(beanFactory); + static TracingDoOnRequest create(LazyBean httpTracing) { + return new TracingDoOnRequest(httpTracing); } - private HttpTracing httpTracing() { - if (this.httpTracing == null) { - this.httpTracing = this.beanFactory.getBean(HttpTracing.class); - } - return this.httpTracing; - } - - private List propagationKeys() { + List propagationKeys() { if (this.propagationKeys == null) { - this.propagationKeys = httpTracing().tracing().propagation().keys(); + this.propagationKeys = httpTracing.get().tracing().propagation().keys(); } return this.propagationKeys; } - private HttpClientHandler handler() { + HttpClientHandler handler() { if (this.handler == null) { - this.handler = HttpClientHandler.create(httpTracing()); + this.handler = HttpClientHandler.create(httpTracing.get()); } return this.handler; } @@ -147,12 +141,12 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { private static class TracingDoOnResponse extends AbstractTracingDoOnHandler implements BiConsumer { - TracingDoOnResponse(BeanFactory beanFactory) { - super(beanFactory); + TracingDoOnResponse(LazyBean httpTracing) { + super(httpTracing); } - static TracingDoOnResponse create(BeanFactory beanFactory) { - return new TracingDoOnResponse(beanFactory); + static TracingDoOnResponse create(LazyBean httpTracing) { + return new TracingDoOnResponse(httpTracing); } @Override @@ -165,12 +159,12 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { private static class TracingDoOnErrorRequest extends AbstractTracingDoOnHandler implements BiConsumer { - TracingDoOnErrorRequest(BeanFactory beanFactory) { - super(beanFactory); + TracingDoOnErrorRequest(LazyBean httpTracing) { + super(httpTracing); } - static TracingDoOnErrorRequest create(BeanFactory beanFactory) { - return new TracingDoOnErrorRequest(beanFactory); + static TracingDoOnErrorRequest create(LazyBean httpTracing) { + return new TracingDoOnErrorRequest(httpTracing); } @Override @@ -183,12 +177,12 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { private static class TracingDoOnErrorResponse extends AbstractTracingDoOnHandler implements BiConsumer { - TracingDoOnErrorResponse(BeanFactory beanFactory) { - super(beanFactory); + TracingDoOnErrorResponse(LazyBean httpTracing) { + super(httpTracing); } - static TracingDoOnErrorResponse create(BeanFactory beanFactory) { - return new TracingDoOnErrorResponse(beanFactory); + static TracingDoOnErrorResponse create(LazyBean httpTracing) { + return new TracingDoOnErrorResponse(httpTracing); } @Override @@ -200,26 +194,17 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { private static abstract class AbstractTracingDoOnHandler { - final BeanFactory beanFactory; - - HttpTracing httpTracing; + final LazyBean httpTracing; HttpClientHandler handler; - AbstractTracingDoOnHandler(BeanFactory beanFactory) { - this.beanFactory = beanFactory; + AbstractTracingDoOnHandler(LazyBean httpTracing) { + this.httpTracing = httpTracing; } - private HttpTracing httpTracing() { - if (this.httpTracing == null) { - this.httpTracing = this.beanFactory.getBean(HttpTracing.class); - } - return this.httpTracing; - } - - private HttpClientHandler handler() { + HttpClientHandler handler() { if (this.handler == null) { - this.handler = HttpClientHandler.create(httpTracing()); + this.handler = HttpClientHandler.create(httpTracing.get()); } return this.handler; } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java index ef596dddf..bb59752b7 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java @@ -162,8 +162,8 @@ public class TraceWebClientAutoConfiguration { @Bean static HttpClientBeanPostProcessor httpClientBeanPostProcessor( - BeanFactory beanFactory) { - return new HttpClientBeanPostProcessor(beanFactory); + ConfigurableApplicationContext springContext) { + return new HttpClientBeanPostProcessor(springContext); } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBean.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/internal/LazyBean.java similarity index 83% rename from spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBean.java rename to spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/internal/LazyBean.java index 364d18648..8a55e0b31 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBean.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/internal/LazyBean.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.sleuth.instrument.reactor; +package org.springframework.cloud.sleuth.internal; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -25,8 +25,16 @@ import org.springframework.lang.Nullable; /** * Avoids calling the expensive {@link ConfigurableApplicationContext#getBean(Class)} many * times or throwing an exception. + * + *

+ * Note: This is an internal class to sleuth and must not be used by external code. */ -final class LazyBean { +public final class LazyBean { + + public static LazyBean create(ConfigurableApplicationContext springContext, + Class requiredType) { + return new LazyBean<>(springContext, requiredType); + } // spring-jcl uses commons-logging, so do we. private static final Log log = LogFactory.getLog(LazyBean.class); @@ -47,7 +55,7 @@ final class LazyBean { * @return the bean value or null if there was an exception getting it. */ @Nullable - T get() { + public T get() { if (this.value != null) { return this.value; } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientSpringBootTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientSpringBootTests.java new file mode 100644 index 000000000..30b1dc1c8 --- /dev/null +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientSpringBootTests.java @@ -0,0 +1,138 @@ +/* + * 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.client; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +import brave.propagation.B3SinglePropagation; +import brave.propagation.Propagation; +import brave.sampler.Sampler; +import org.junit.After; +import org.junit.Test; +import org.junit.runner.RunWith; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.netty.DisposableServer; +import reactor.netty.http.client.HttpClient; +import reactor.netty.http.server.HttpServer; +import zipkin2.Span; +import zipkin2.reporter.Reporter; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.web.reactive.function.client.WebClient; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * This tests {@link HttpClient} instrumentation performed by + * {@link HttpClientBeanPostProcessor}, as wired by auto-configuration. + * + *

+ * Note: {@link HttpClient} can be an implementation of {@link WebClient}, so + * care should be taken to also test that integration. For example, it would be easy to + * create duplicate client spans for the same request. + */ +@SpringBootTest(classes = ReactorNettyHttpClientSpringBootTests.TestConfiguration.class, + webEnvironment = SpringBootTest.WebEnvironment.NONE) +@RunWith(SpringRunner.class) +public class ReactorNettyHttpClientSpringBootTests { + + DisposableServer disposableServer; + + @Autowired + HttpClient httpClient; + + @Autowired + BlockingQueue spans; + + @After + public void tearDown() { + if (disposableServer != null) { + disposableServer.disposeNow(); + } + this.spans.clear(); + } + + @Test + public void shouldSendTraceContextToServer_rootSpan() throws Exception { + disposableServer = HttpServer.create().port(0) + // this reads the trace context header, b3, returning it in the response + .handle((in, out) -> out + .sendString(Flux.just(in.requestHeaders().get("b3")))) + .bindNow(); + + Mono request = httpClient.port(disposableServer.port()).get().uri("/") + .responseContent().aggregate().asString(); + + String b3SingleHeaderReadByServer = request.block(); + + Span clientSpan = takeClientSpan(); + + assertThat(b3SingleHeaderReadByServer) + .isEqualTo(clientSpan.traceId() + "-" + clientSpan.id() + "-1"); + } + + /** Call this to block until a span was reported */ + Span takeClientSpan() throws InterruptedException { + Span result = spans.poll(1, TimeUnit.SECONDS); + assertThat(result).withFailMessage("Span was not reported").isNotNull(); + assertThat(result.kind()).isEqualTo(Span.Kind.CLIENT); + return result; + } + + @Configuration + @EnableAutoConfiguration + static class TestConfiguration { + + @Bean + Propagation.Factory propagationFactory() { + return B3SinglePropagation.FACTORY; + } + + @Bean + Sampler sampler() { + return Sampler.ALWAYS_SAMPLE; + } + + /** + * Use a blocking queue as it is simpler than wrapping everything in awaitility + */ + @Bean + BlockingQueue spans() { + return new LinkedBlockingQueue<>(); + } + + @Bean + Reporter spanReporter(BlockingQueue spans) { + return spans::add; + } + + @Bean + HttpClient reactorHttpClient() { + return HttpClient.create(); + } + + } + +} diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java index 23a066368..26962aa66 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java @@ -55,8 +55,6 @@ import org.junit.ClassRule; import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; -import reactor.netty.http.client.HttpClient; -import reactor.netty.http.client.HttpClientResponse; import zipkin2.Annotation; import zipkin2.reporter.Reporter; @@ -136,9 +134,6 @@ public class WebClientTests { @Autowired HttpClientBuilder httpClientBuilder; // #845 - @Autowired - HttpClient nettyHttpClient; - @Autowired HttpAsyncClientBuilder httpAsyncClientBuilder; // #845 @@ -278,30 +273,6 @@ public class WebClientTests { then(this.reporter.getSpans()).isNotEmpty(); } - @Test - @SuppressWarnings("unchecked") - public void shouldAttachTraceIdWhenCallingAnotherServiceForNettyHttpClient() - throws Exception { - Span span = this.tracer.nextSpan().name("foo").start(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - HttpClientResponse response = this.nettyHttpClient.get() - .uri("http://localhost:" + this.port).response().block(); - - then(response).isNotNull(); - } - - Awaitility.await().untilAsserted(() -> { - then(this.tracer.currentSpan()).isNull(); - System.out.println("Collected span " + this.reporter.getSpans()); - then(this.reporter.getSpans()).isNotEmpty() - .extracting("traceId", String.class) - // we can have some bizarre spans popping up - .contains(span.context().traceIdString()); - then(this.reporter.getSpans()).extracting("kind.name").contains("CLIENT"); - }); - } - @Test @SuppressWarnings("unchecked") public void shouldAttachTraceIdWhenCallingAnotherServiceForHttpClient() @@ -601,11 +572,6 @@ public class WebClientTests { return new MyRestTemplateCustomizer(); } - @Bean - HttpClient reactorHttpClient() { - return HttpClient.create(); - } - } static class MyRestTemplateCustomizer implements RestTemplateCustomizer { diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBeanTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/internal/LazyBeanTests.java similarity index 95% rename from spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBeanTests.java rename to spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/internal/LazyBeanTests.java index 3f9694ba6..db03fb410 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBeanTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/internal/LazyBeanTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.sleuth.instrument.reactor; +package org.springframework.cloud.sleuth.internal; import brave.propagation.CurrentTraceContext; import org.junit.Test;