Finalizes reactor-netty HttpClient implementations (#1586)

benchmarks/pom.xml

@@ -33,7 +33,7 @@
 		<maven.compiler.target>1.8</maven.compiler.target>
 		<maven.compiler.source>1.8</maven.compiler.source>
 		<spring-boot.version>2.2.5.RELEASE</spring-boot.version>
-		<brave.version>5.10.1</brave.version>
+		<brave.version>5.10.2</brave.version>
 		<okhttp.version>3.14.6</okhttp.version>
 	</properties>
 

pom.xml

@@ -132,6 +132,13 @@
 
 	<dependencyManagement>
 		<dependencies>
+			<!-- force latest reactor-netty -->
+			<dependency>
+				<groupId>io.projectreactor.netty</groupId>
+				<artifactId>reactor-netty</artifactId>
+				<version>0.9.6.RELEASE</version>
+				<optional>true</optional>
+			</dependency>
 			<dependency>
 				<groupId>org.springframework.cloud</groupId>
 				<artifactId>spring-cloud-sleuth-dependencies</artifactId>
@@ -257,7 +264,7 @@
 		<spring-cloud-stream.version>Horsham.SR3</spring-cloud-stream.version>
 		<spring-cloud-netflix.version>2.2.3.BUILD-SNAPSHOT</spring-cloud-netflix.version>
 		<spring-cloud-openfeign.version>2.2.3.BUILD-SNAPSHOT</spring-cloud-openfeign.version>
-		<brave.version>5.10.1</brave.version>
+		<brave.version>5.10.2</brave.version>
 		<spring-security-boot-autoconfigure.version>2.1.7.RELEASE</spring-security-boot-autoconfigure.version>
 		<spring-cloud-aws.version>2.2.1.RELEASE</spring-cloud-aws.version>
 		<disable.nohttp.checks>false</disable.nohttp.checks>

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

spring-cloud-sleuth-dependencies/pom.xml

@@ -31,7 +31,7 @@
 	<name>spring-cloud-sleuth-dependencies</name>
 	<description>Spring Cloud Sleuth Dependencies</description>
 	<properties>
-		<brave.version>5.10.1</brave.version>
+		<brave.version>5.10.2</brave.version>
 		<brave.opentracing.version>0.35.1</brave.opentracing.version>
 		<grpc.spring.boot.version>3.4.1</grpc.spring.boot.version>
 	</properties>

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");
 	}
 
 }

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}.
+	 *
+	 * <p>
+	 * We do this implicitly until
+	 * <a href="https://github.com/reactor/reactor-netty/issues/1036">issue 1036</a>.
+	 * Until then, there's no known way to directly instrument the
+	 * {@code Mono<Connection>} 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<Integer> postMono(AnnotationConfigApplicationContext context,
 			String pathIncludingQuery, String body) {

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() {
 	}