From 1e1e1f04bfd28b65109fe8f3fdd5ad4af7981950 Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Thu, 26 Mar 2020 17:10:01 +0800 Subject: [PATCH] Finalizes reactor-netty HttpClient implementations (#1586) --- benchmarks/pom.xml | 2 +- pom.xml | 9 +- .../client/HttpClientBeanPostProcessor.java | 40 +++++---- spring-cloud-sleuth-dependencies/pom.xml | 2 +- .../ITSpringConfiguredReactorClient.java | 13 ++- .../ReactorNettyHttpClientBraveTests.java | 86 ++++++++----------- .../web/client/WebClientBraveTests.java | 2 +- 7 files changed, 80 insertions(+), 74 deletions(-) diff --git a/benchmarks/pom.xml b/benchmarks/pom.xml index 96fed1067..d326c1d79 100644 --- a/benchmarks/pom.xml +++ b/benchmarks/pom.xml @@ -33,7 +33,7 @@ 1.8 1.8 2.2.5.RELEASE - 5.10.1 + 5.10.2 3.14.6 diff --git a/pom.xml b/pom.xml index 9d03785f1..e5c405d6c 100644 --- a/pom.xml +++ b/pom.xml @@ -132,6 +132,13 @@ + + + io.projectreactor.netty + reactor-netty + 0.9.6.RELEASE + true + org.springframework.cloud spring-cloud-sleuth-dependencies @@ -257,7 +264,7 @@ Horsham.SR3 2.2.3.BUILD-SNAPSHOT 2.2.3.BUILD-SNAPSHOT - 5.10.1 + 5.10.2 2.1.7.RELEASE 2.2.1.RELEASE false 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 fc26b271b..83bbcf3dc 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 @@ -59,9 +59,13 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { // propagation of the current span as a reactor context property. // This done in mapConnect, added last so that it is setup first. // https://projectreactor.io/docs/core/release/reference/#_simple_context_examples + + // In our case, we treat a normal response no differently than one in + // preparation of a redirect follow-up. + TracingDoOnResponse doOnResponse = new TracingDoOnResponse(httpTracing); return ((HttpClient) bean) .doOnResponseError(new TracingDoOnErrorResponse(httpTracing)) - .doOnResponse(new TracingDoOnResponse(httpTracing)) + .doOnRedirect(doOnResponse).doOnResponse(doOnResponse) .doOnRequestError(new TracingDoOnErrorRequest(httpTracing)) .doOnRequest(new TracingDoOnRequest(httpTracing)) .mapConnect(new TracingMapConnect(() -> { @@ -146,16 +150,13 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { return; // Somehow TracingMapConnect was not invoked.. skip out } - // This might be re-entrant on auto-redirect or connection retry: - // See reactor/reactor-netty#1000 for follow-ups. + // All completion hooks clear this reference. If somehow this has a span upon + // re-entry, the state model in reactor-netty has changed and we need to + // update this code! Span span = pendingSpan.getAndSet(null); if (span != null) { - // Retry from a connect fail wouldn't have parsed the request, leading to - // an empty span with no data if we finished it. An auto-redirect would - // have parsed the request, but we have no idea which status code it - // finished with. Since we can't see the preceding request state, we - // abandon its span in favor of the next. - span.abandon(); + assert false : "span exists when it shouldn't!"; + span.abandon(); // abandon instead of break } // Start a new client span with the appropriate parent @@ -187,7 +188,6 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { @Override public void accept(HttpClientResponse response, Connection connection) { - // TODO: is there a way to read the request at response time? handle(response.currentContext(), response, null); } @@ -276,7 +276,7 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { @Override public String path() { - return "/" + delegate.path(); // TODO: reactor/reactor-netty#999 + return delegate.fullPath(); } @Override @@ -300,18 +300,28 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { final HttpClientResponse delegate; + HttpClientRequestWrapper request; + HttpClientResponseWrapper(HttpClientResponse delegate) { this.delegate = delegate; } @Override - public String method() { - return delegate.method().name(); + public Object unwrap() { + return delegate; } @Override - public Object unwrap() { - return delegate; + public HttpClientRequestWrapper request() { + if (request == null) { + if (delegate instanceof HttpClientRequest) { + request = new HttpClientRequestWrapper((HttpClientRequest) delegate); + } + else { + assert false : "We expect the response to be the same reference as the request"; + } + } + return request; } @Override diff --git a/spring-cloud-sleuth-dependencies/pom.xml b/spring-cloud-sleuth-dependencies/pom.xml index a995370e0..afd822bec 100644 --- a/spring-cloud-sleuth-dependencies/pom.xml +++ b/spring-cloud-sleuth-dependencies/pom.xml @@ -31,7 +31,7 @@ spring-cloud-sleuth-dependencies Spring Cloud Sleuth Dependencies - 5.10.1 + 5.10.2 0.35.1 3.4.1 diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ITSpringConfiguredReactorClient.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ITSpringConfiguredReactorClient.java index 5ac284c46..1f11e86bf 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ITSpringConfiguredReactorClient.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ITSpringConfiguredReactorClient.java @@ -21,6 +21,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import brave.http.HttpTracing; +import brave.propagation.CurrentTraceContext; import brave.test.http.ITHttpAsyncClient; import io.netty.channel.ChannelOption; import io.netty.handler.timeout.ReadTimeoutHandler; @@ -50,17 +51,20 @@ abstract class ITSpringConfiguredReactorClient /** * @param componentClasses configure instrumentation given {@linkplain URI baseUrl}, - * {@link HttpClient} and {@link HttpTracing} bindings exist. + * {@link HttpClient}, {@link HttpTracing} and {@link CurrentTraceContext} bindings + * exist. */ ITSpringConfiguredReactorClient(Class... componentClasses) { this.componentClasses = componentClasses; } @Override - final protected AnnotationConfigApplicationContext newClient(int port) { + protected AnnotationConfigApplicationContext newClient(int port) { AnnotationConfigApplicationContext result = new AnnotationConfigApplicationContext(); URI baseUrl = URI.create("http://127.0.0.1:" + server.getPort()); result.registerBean(HttpTracing.class, () -> httpTracing); + result.registerBean(CurrentTraceContext.class, + () -> httpTracing.tracing().currentTraceContext()); result.registerBean(HttpClient.class, () -> testHttpClient(baseUrl)); result.registerBean(URI.class, () -> baseUrl); result.register(componentClasses); @@ -74,7 +78,7 @@ abstract class ITSpringConfiguredReactorClient .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000) .doOnConnected(conn -> conn .addHandler(new ReadTimeoutHandler(1, TimeUnit.SECONDS)))) - .followRedirect(true); + .disableRetry(true).followRedirect(true); } @Override @@ -155,7 +159,8 @@ abstract class ITSpringConfiguredReactorClient latch.await(); assertThat(server.getRequestCount()).isOne(); - assertThat(takeSpan().tags()).containsKey("error"); + + takeClientSpanWithError("CANCELLED"); } } diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java index b52d80fe7..af509e1c2 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java @@ -16,17 +16,50 @@ package org.springframework.cloud.sleuth.instrument.web.client; -import org.junit.Ignore; -import org.junit.Test; +import org.junit.After; +import org.reactivestreams.Subscriber; import reactor.core.publisher.Mono; import reactor.netty.ByteBufFlux; import reactor.netty.http.client.HttpClient; import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.cloud.sleuth.instrument.reactor.TraceReactorAutoConfiguration; +import org.springframework.cloud.sleuth.instrument.reactor.TraceReactorAutoConfigurationAccessorConfiguration; import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.web.reactive.function.client.WebClient; +/** + * This tests the reactor {@link HttpClient} in isolation of {@link WebClient} as it could + * be used directly. + */ public class ReactorNettyHttpClientBraveTests extends ITSpringConfiguredReactorClient { + /** + * This borrows hooks from {@link TraceReactorAutoConfiguration} to ensure that the + * invocation trace context is set in scope for hooks like {@link Subscriber#onNext}. + * + *

+ * We do this implicitly until + * issue 1036. + * Until then, there's no known way to directly instrument the + * {@code Mono} created in + * {@code reactor.netty.http.client.MonoConnect$MonoHttpConnect} with + * {@code ScopePassingSpanSubscriber}. While this looks like cheating the test, Sleuth + * will always setup these hooks anyway unless "spring.sleuth.reactor.enabled=false". + */ + @Override + protected AnnotationConfigApplicationContext newClient(int port) { + TraceReactorAutoConfigurationAccessorConfiguration.close(); + AnnotationConfigApplicationContext context = super.newClient(port); + TraceReactorAutoConfigurationAccessorConfiguration.setup(context); + return context; + } + + @After + public void cleanupHooks() { + TraceReactorAutoConfigurationAccessorConfiguration.close(); + } + /** * This uses Spring to instrument the {@link HttpClient} using a * {@link BeanPostProcessor}. @@ -35,55 +68,6 @@ public class ReactorNettyHttpClientBraveTests extends ITSpringConfiguredReactorC super(HttpClientBeanPostProcessor.class); } - @Test - @Ignore("TODO: NPE reading context: consider integrating TracingMapConnect with ScopePassingSpanSubscriber") - @Override - public void callbackContextIsFromInvocationTime() { - } - - @Test - @Ignore("TODO: reactor/reactor-netty#1000") - @Override - public void redirect() { - } - - @Test - @Ignore("TODO: reactor/reactor-netty#1000") - @Override - public void supportsPortableCustomization() { - } - - @Test - @Ignore("TODO: reactor/reactor-netty#1000") - @Override - @Deprecated - public void supportsDeprecatedPortableCustomization() { - } - - @Test - @Ignore("TODO: reactor/reactor-netty#1000") - @Override - public void post() { - } - - @Test - @Ignore("TODO: reactor/reactor-netty#1000") - @Override - public void customSampler() { - } - - @Test - @Ignore("TODO: reactor/reactor-netty#1000") - @Override - public void httpPathTagExcludesQueryParams() { - } - - @Test - @Ignore("HttpClient has no function to retrieve the wire request from the response") - @Override - public void readsRequestAtResponseTime() { - } - @Override Mono postMono(AnnotationConfigApplicationContext context, String pathIncludingQuery, String body) { diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java index 59c247885..27f5e9d24 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java @@ -67,7 +67,7 @@ public class WebClientBraveTests extends ITSpringConfiguredReactorClient { } @Test - @Ignore("TODO: reactor/reactor-netty#1000") + @Ignore("WebClient is blind to the implementation of redirects") @Override public void redirect() { }