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.
This commit is contained in:
@@ -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<CurrentTraceContext> lazyCurrentTraceContext = new LazyBean<>(
|
||||
springContext, CurrentTraceContext.class);
|
||||
LazyBean<CurrentTraceContext> 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
|
||||
|
||||
@@ -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> 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<Mono<? extends Connection>, Bootstrap, Mono<? extends Connection>> {
|
||||
|
||||
private final BeanFactory beanFactory;
|
||||
final LazyBean<HttpTracing> httpTracing;
|
||||
|
||||
private Tracer tracer;
|
||||
Tracer tracer;
|
||||
|
||||
TracingMapConnect(BeanFactory beanFactory) {
|
||||
this.beanFactory = beanFactory;
|
||||
TracingMapConnect(LazyBean<HttpTracing> 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<HttpClientRequest, Connection> {
|
||||
|
||||
final BeanFactory beanFactory;
|
||||
|
||||
HttpTracing httpTracing;
|
||||
final LazyBean<HttpTracing> httpTracing;
|
||||
|
||||
List<String> propagationKeys;
|
||||
|
||||
HttpClientHandler<brave.http.HttpClientRequest, brave.http.HttpClientResponse> handler;
|
||||
|
||||
TracingDoOnRequest(BeanFactory beanFactory) {
|
||||
this.beanFactory = beanFactory;
|
||||
TracingDoOnRequest(LazyBean<HttpTracing> httpTracing) {
|
||||
this.httpTracing = httpTracing;
|
||||
}
|
||||
|
||||
static TracingDoOnRequest create(BeanFactory beanFactory) {
|
||||
return new TracingDoOnRequest(beanFactory);
|
||||
static TracingDoOnRequest create(LazyBean<HttpTracing> httpTracing) {
|
||||
return new TracingDoOnRequest(httpTracing);
|
||||
}
|
||||
|
||||
private HttpTracing httpTracing() {
|
||||
if (this.httpTracing == null) {
|
||||
this.httpTracing = this.beanFactory.getBean(HttpTracing.class);
|
||||
}
|
||||
return this.httpTracing;
|
||||
}
|
||||
|
||||
private List<String> propagationKeys() {
|
||||
List<String> propagationKeys() {
|
||||
if (this.propagationKeys == null) {
|
||||
this.propagationKeys = httpTracing().tracing().propagation().keys();
|
||||
this.propagationKeys = httpTracing.get().tracing().propagation().keys();
|
||||
}
|
||||
return this.propagationKeys;
|
||||
}
|
||||
|
||||
private HttpClientHandler<brave.http.HttpClientRequest, brave.http.HttpClientResponse> handler() {
|
||||
HttpClientHandler<brave.http.HttpClientRequest, brave.http.HttpClientResponse> 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<HttpClientResponse, Connection> {
|
||||
|
||||
TracingDoOnResponse(BeanFactory beanFactory) {
|
||||
super(beanFactory);
|
||||
TracingDoOnResponse(LazyBean<HttpTracing> httpTracing) {
|
||||
super(httpTracing);
|
||||
}
|
||||
|
||||
static TracingDoOnResponse create(BeanFactory beanFactory) {
|
||||
return new TracingDoOnResponse(beanFactory);
|
||||
static TracingDoOnResponse create(LazyBean<HttpTracing> httpTracing) {
|
||||
return new TracingDoOnResponse(httpTracing);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -165,12 +159,12 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor {
|
||||
private static class TracingDoOnErrorRequest extends AbstractTracingDoOnHandler
|
||||
implements BiConsumer<HttpClientRequest, Throwable> {
|
||||
|
||||
TracingDoOnErrorRequest(BeanFactory beanFactory) {
|
||||
super(beanFactory);
|
||||
TracingDoOnErrorRequest(LazyBean<HttpTracing> httpTracing) {
|
||||
super(httpTracing);
|
||||
}
|
||||
|
||||
static TracingDoOnErrorRequest create(BeanFactory beanFactory) {
|
||||
return new TracingDoOnErrorRequest(beanFactory);
|
||||
static TracingDoOnErrorRequest create(LazyBean<HttpTracing> httpTracing) {
|
||||
return new TracingDoOnErrorRequest(httpTracing);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -183,12 +177,12 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor {
|
||||
private static class TracingDoOnErrorResponse extends AbstractTracingDoOnHandler
|
||||
implements BiConsumer<HttpClientResponse, Throwable> {
|
||||
|
||||
TracingDoOnErrorResponse(BeanFactory beanFactory) {
|
||||
super(beanFactory);
|
||||
TracingDoOnErrorResponse(LazyBean<HttpTracing> httpTracing) {
|
||||
super(httpTracing);
|
||||
}
|
||||
|
||||
static TracingDoOnErrorResponse create(BeanFactory beanFactory) {
|
||||
return new TracingDoOnErrorResponse(beanFactory);
|
||||
static TracingDoOnErrorResponse create(LazyBean<HttpTracing> 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> httpTracing;
|
||||
|
||||
HttpClientHandler<brave.http.HttpClientRequest, brave.http.HttpClientResponse> handler;
|
||||
|
||||
AbstractTracingDoOnHandler(BeanFactory beanFactory) {
|
||||
this.beanFactory = beanFactory;
|
||||
AbstractTracingDoOnHandler(LazyBean<HttpTracing> httpTracing) {
|
||||
this.httpTracing = httpTracing;
|
||||
}
|
||||
|
||||
private HttpTracing httpTracing() {
|
||||
if (this.httpTracing == null) {
|
||||
this.httpTracing = this.beanFactory.getBean(HttpTracing.class);
|
||||
}
|
||||
return this.httpTracing;
|
||||
}
|
||||
|
||||
private HttpClientHandler<brave.http.HttpClientRequest, brave.http.HttpClientResponse> handler() {
|
||||
HttpClientHandler<brave.http.HttpClientRequest, brave.http.HttpClientResponse> handler() {
|
||||
if (this.handler == null) {
|
||||
this.handler = HttpClientHandler.create(httpTracing());
|
||||
this.handler = HttpClientHandler.create(httpTracing.get());
|
||||
}
|
||||
return this.handler;
|
||||
}
|
||||
|
||||
@@ -162,8 +162,8 @@ public class TraceWebClientAutoConfiguration {
|
||||
|
||||
@Bean
|
||||
static HttpClientBeanPostProcessor httpClientBeanPostProcessor(
|
||||
BeanFactory beanFactory) {
|
||||
return new HttpClientBeanPostProcessor(beanFactory);
|
||||
ConfigurableApplicationContext springContext) {
|
||||
return new HttpClientBeanPostProcessor(springContext);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
*
|
||||
* <p>
|
||||
* Note: This is an internal class to sleuth and must not be used by external code.
|
||||
*/
|
||||
final class LazyBean<T> {
|
||||
public final class LazyBean<T> {
|
||||
|
||||
public static <T> LazyBean<T> create(ConfigurableApplicationContext springContext,
|
||||
Class<T> 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<T> {
|
||||
* @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;
|
||||
}
|
||||
@@ -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.
|
||||
*
|
||||
* <p>
|
||||
* <em>Note:</em> {@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<Span> 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<String> 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<Span> spans() {
|
||||
return new LinkedBlockingQueue<>();
|
||||
}
|
||||
|
||||
@Bean
|
||||
Reporter<zipkin2.Span> spanReporter(BlockingQueue<Span> spans) {
|
||||
return spans::add;
|
||||
}
|
||||
|
||||
@Bean
|
||||
HttpClient reactorHttpClient() {
|
||||
return HttpClient.create();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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;
|
||||
Reference in New Issue
Block a user