Add support for Reactor Netty Brave additional logging (#2295)

* Add support for Reactor Netty Brave additional logging

with this change a new feature (opt-in via a property) has been added to add more observability on the Reactor Netty side.

WARNING: Turning this on can lead to a significant drop in performance. Use with caution

---------

Co-authored-by: Jonatan Ivanov <jonatan.ivanov@gmail.com>

pom.xml

@@ -77,6 +77,7 @@
 		<tomcat-jdbc.version>10.0.6</tomcat-jdbc.version>
 		<commons-dbcp2.version>2.8.0</commons-dbcp2.version>
 		<kotlin.version>1.6.21</kotlin.version>
+		<wiremock.version>2.35.0</wiremock.version>
 
 		<!-- Until we switch it to true in sc-build -->
 		<javadoc.failOnError>true</javadoc.failOnError>

spring-cloud-sleuth-autoconfigure/pom.xml

@@ -338,6 +338,11 @@
 			<artifactId>reactor-kafka</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>io.projectreactor.netty</groupId>
+			<artifactId>reactor-netty-http-brave</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<!-- GRPC Optional Dependencies -->
 		<dependency>
 			<groupId>io.github.lognet</groupId>

spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/reactor/netty/BraveReactorNettyAutoConfiguration.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2013-2021 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.autoconfig.brave.instrument.reactor.netty;
+
+import brave.http.HttpTracing;
+import brave.propagation.CurrentTraceContext;
+import reactor.netty.NettyPipeline;
+import reactor.netty.http.brave.ReactorNettyHttpTracing;
+
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.web.embedded.netty.NettyServerCustomizer;
+import org.springframework.cloud.gateway.config.HttpClientCustomizer;
+import org.springframework.cloud.sleuth.autoconfig.brave.BraveAutoConfiguration;
+import org.springframework.cloud.sleuth.brave.instrument.reactor.netty.TracingChannelInboundHandler;
+import org.springframework.cloud.sleuth.brave.instrument.reactor.netty.TracingChannelOutboundHandler;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration
+ * Auto-configuration} to enable additional tracing with Reactor Netty.
+ *
+ * @author Marcin Grzejszczak
+ * @since 3.1.9
+ */
+@Configuration(proxyBeanMethods = false)
+@ConditionalOnClass({ HttpTracing.class, ReactorNettyHttpTracing.class })
+@AutoConfigureAfter(BraveAutoConfiguration.class)
+public class BraveReactorNettyAutoConfiguration {
+
+	@Configuration(proxyBeanMethods = false)
+	@ConditionalOnProperty("spring.sleuth.reactor.netty.debug.enabled")
+	@ConditionalOnBean(HttpTracing.class)
+	static class DebugReactorNettyConfiguration {
+
+		static final String INBOUND_NAME = NettyPipeline.LEFT + "customTracingChannelInboundHandler";
+
+		static final String OUTBOUND_NAME = NettyPipeline.RIGHT + "customTracingChannelOutboundHandler";
+
+		@Bean
+		public NettyServerCustomizer tracingNettyServerCustomizer(HttpTracing httpTracing) {
+			return server -> ReactorNettyHttpTracing.create(httpTracing).decorateHttpServer(server)
+					.doOnChannelInit((obs, ch, addr) -> {
+						CurrentTraceContext currentTraceContext = httpTracing.tracing().currentTraceContext();
+
+						String oldNameInboundHandler = NettyPipeline.LEFT + "tracingChannelInboundHandler";
+						ch.pipeline().remove(oldNameInboundHandler);
+						ch.pipeline().addFirst(INBOUND_NAME, new TracingChannelInboundHandler(currentTraceContext));
+
+						String oldNameOutboundHandler = NettyPipeline.RIGHT + "tracingChannelOutboundHandler";
+						ch.pipeline().replace(oldNameOutboundHandler, OUTBOUND_NAME,
+								new TracingChannelOutboundHandler(currentTraceContext));
+					});
+		}
+
+		@Bean
+		public HttpClientCustomizer tracingHttpClientCustomizer(HttpTracing httpTracing) {
+			return client -> client.doOnChannelInit((obs, ch, addr) -> {
+				CurrentTraceContext currentTraceContext = httpTracing.tracing().currentTraceContext();
+				ch.pipeline().addFirst(INBOUND_NAME, new TracingChannelInboundHandler(currentTraceContext));
+				ch.pipeline().addBefore(NettyPipeline.ReactiveBridge, OUTBOUND_NAME,
+						new TracingChannelOutboundHandler(currentTraceContext));
+			});
+		}
+
+	}
+
+}

spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -214,6 +214,12 @@
       "type": "java.lang.Boolean",
       "description": "Enable Spring Vault instrumentation.",
       "defaultValue": true
+    },
+    {
+      "name": "spring.sleuth.reactor.netty.debug.enabled",
+      "type": "java.lang.Boolean",
+      "description": "WARNING: Use with caution, can lead to serious performance issues. Enable additional instrumentation for Reactor Netty.",
+      "defaultValue": false
     }
   ]
 }

spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/spring.factories

@@ -43,6 +43,7 @@ org.springframework.cloud.sleuth.autoconfig.brave.instrument.messaging.BraveKafk
 org.springframework.cloud.sleuth.autoconfig.brave.instrument.messaging.BraveMessagingAutoConfiguration,\
 org.springframework.cloud.sleuth.autoconfig.brave.instrument.opentracing.BraveOpentracingAutoConfiguration,\
 org.springframework.cloud.sleuth.autoconfig.brave.instrument.redis.BraveRedisAutoConfiguration,\
+org.springframework.cloud.sleuth.autoconfig.brave.instrument.reactor.netty.BraveReactorNettyAutoConfiguration,\
 org.springframework.cloud.sleuth.autoconfig.brave.instrument.mongodb.BraveMongoDbAutoConfiguration,\
 org.springframework.cloud.sleuth.autoconfig.zipkin2.ZipkinAutoConfiguration
 # Environment Post Processor

spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/reactor/netty/BraveReactorNettyAutoConfigurationTests.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2013-2023 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.autoconfig.brave.instrument.reactor.netty;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.FilteredClassLoader;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.boot.web.embedded.netty.NettyServerCustomizer;
+import org.springframework.cloud.gateway.config.HttpClientCustomizer;
+import org.springframework.cloud.sleuth.autoconfig.brave.BraveAutoConfiguration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class BraveReactorNettyAutoConfigurationTests {
+
+	@Test
+	void should_not_auto_configure_brave_reactor_netty_by_default() {
+		new ApplicationContextRunner()
+				.withConfiguration(
+						AutoConfigurations.of(BraveAutoConfiguration.class, BraveReactorNettyAutoConfiguration.class))
+				.run(context -> assertThat(context).doesNotHaveBean(NettyServerCustomizer.class)
+						.doesNotHaveBean(HttpClientCustomizer.class));
+	}
+
+	@Test
+	void should_not_auto_configure_brave_reactor_netty_when_no_http_tracing_on_classpath() {
+		new ApplicationContextRunner().withPropertyValues("spring.sleuth.reactor.netty.debug.enabled=true")
+				.withClassLoader(new FilteredClassLoader("brave.http.HttpTracing"))
+				.withConfiguration(
+						AutoConfigurations.of(BraveAutoConfiguration.class, BraveReactorNettyAutoConfiguration.class))
+				.run(context -> assertThat(context).doesNotHaveBean(NettyServerCustomizer.class)
+						.doesNotHaveBean(HttpClientCustomizer.class));
+	}
+
+	@Test
+	void should_not_auto_configure_brave_reactor_netty_when_no_reactor_netty_brave_on_classpath() {
+		new ApplicationContextRunner().withPropertyValues("spring.sleuth.reactor.netty.debug.enabled=true")
+				.withClassLoader(new FilteredClassLoader("reactor.netty.http.brave.ReactorNettyHttpTracing"))
+				.withConfiguration(
+						AutoConfigurations.of(BraveAutoConfiguration.class, BraveReactorNettyAutoConfiguration.class))
+				.run(context -> assertThat(context).doesNotHaveBean(NettyServerCustomizer.class)
+						.doesNotHaveBean(HttpClientCustomizer.class));
+	}
+
+	@Test
+	void should_auto_configure_brave_reactor_netty_when_property_set() {
+		new ApplicationContextRunner().withPropertyValues("spring.sleuth.reactor.netty.debug.enabled=true")
+				.withConfiguration(
+						AutoConfigurations.of(BraveAutoConfiguration.class, BraveReactorNettyAutoConfiguration.class))
+				.run(context -> assertThat(context).hasSingleBean(NettyServerCustomizer.class)
+						.hasSingleBean(HttpClientCustomizer.class));
+	}
+
+}

spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingChannelInboundHandler.java

@@ -0,0 +1,207 @@
+/*
+ * Copyright 2013-2023 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.brave.instrument.reactor.netty;
+
+import brave.Span;
+import brave.propagation.CurrentTraceContext;
+import brave.propagation.TraceContext;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.util.AttributeKey;
+import reactor.netty.Connection;
+import reactor.netty.ConnectionObserver;
+import reactor.netty.channel.ChannelOperations;
+import reactor.netty.http.client.HttpClientResponse;
+
+/**
+ * {@link ChannelInboundHandlerAdapter} that wraps all events in scope.
+ * <p>
+ * WARNING: Using this feature can lead to serious performance issues. This should be only
+ * used for debugging purposes.
+ *
+ * @since 3.1.9
+ */
+public class TracingChannelInboundHandler extends ChannelInboundHandlerAdapter {
+
+	static final AttributeKey<Span> SPAN_ATTRIBUTE_KEY = AttributeKey.valueOf(Span.class.getName());
+
+	final CurrentTraceContext currentTraceContext;
+
+	/**
+	 * Creates a new instance of {@link TracingChannelInboundHandler}.
+	 * @param currentTraceContext current trace context
+	 */
+	public TracingChannelInboundHandler(CurrentTraceContext currentTraceContext) {
+		this.currentTraceContext = currentTraceContext;
+	}
+
+	@Override
+	public void channelRegistered(ChannelHandlerContext ctx) {
+		if (instrumentOperation(ctx, () -> ctx.fireChannelRegistered())) {
+			return;
+		}
+
+		ctx.fireChannelRegistered();
+	}
+
+	@Override
+	public void channelUnregistered(ChannelHandlerContext ctx) {
+		if (instrumentOperation(ctx, () -> ctx.fireChannelUnregistered())) {
+			return;
+		}
+
+		ctx.fireChannelUnregistered();
+	}
+
+	@Override
+	public void channelActive(ChannelHandlerContext ctx) {
+		if (instrumentOperation(ctx, () -> ctx.fireChannelActive())) {
+			return;
+		}
+
+		ctx.fireChannelActive();
+	}
+
+	@Override
+	public void channelInactive(ChannelHandlerContext ctx) {
+		if (instrumentOperation(ctx, () -> ctx.fireChannelInactive())) {
+			return;
+		}
+
+		ctx.fireChannelInactive();
+	}
+
+	@Override
+	public void channelRead(ChannelHandlerContext ctx, Object msg) {
+		if (instrumentOperation(ctx, () -> ctx.fireChannelRead(msg))) {
+			return;
+		}
+
+		ctx.fireChannelRead(msg);
+	}
+
+	@Override
+	public void channelReadComplete(ChannelHandlerContext ctx) {
+		if (instrumentOperation(ctx, () -> ctx.fireChannelReadComplete())) {
+			return;
+		}
+
+		ctx.fireChannelReadComplete();
+	}
+
+	@Override
+	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
+		if (instrumentOperation(ctx, () -> ctx.fireUserEventTriggered(evt))) {
+			return;
+		}
+
+		ctx.fireUserEventTriggered(evt);
+	}
+
+	@Override
+	public void channelWritabilityChanged(ChannelHandlerContext ctx) {
+		if (instrumentOperation(ctx, () -> ctx.fireChannelWritabilityChanged())) {
+			return;
+		}
+
+		ctx.fireChannelWritabilityChanged();
+	}
+
+	@Override
+	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+		if (instrumentOperation(ctx, () -> ctx.fireExceptionCaught(cause))) {
+			return;
+		}
+
+		ctx.fireExceptionCaught(cause);
+	}
+
+	@Override
+	public boolean isSharable() {
+		return false;
+	}
+
+	@Override
+	public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+		if (instrumentOperation(ctx, () -> {
+			try {
+				super.handlerAdded(ctx);
+			}
+			catch (Exception e) {
+				throw new RuntimeException(e);
+			}
+		})) {
+			return;
+		}
+
+		super.handlerAdded(ctx);
+	}
+
+	@Override
+	public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+		if (instrumentOperation(ctx, () -> {
+			try {
+				super.handlerRemoved(ctx);
+			}
+			catch (Exception e) {
+				throw new RuntimeException(e);
+			}
+		})) {
+			return;
+		}
+
+		super.handlerRemoved(ctx);
+	}
+
+	boolean instrumentOperation(ChannelHandlerContext ctx, Runnable operation) {
+		Span span = ctx.channel().attr(SPAN_ATTRIBUTE_KEY).get();
+		if (span != null) {
+			try (CurrentTraceContext.Scope scope = currentTraceContext.maybeScope(span.context())) {
+				operation.run();
+			}
+			return true;
+		}
+		else {
+			Connection conn = Connection.from(ctx.channel());
+			if (conn instanceof ConnectionObserver) {
+				TraceContext parent = ((ConnectionObserver) conn).currentContext().getOrDefault(TraceContext.class,
+						null);
+				if (parent != null) {
+					try (CurrentTraceContext.Scope scope = currentTraceContext.maybeScope(parent)) {
+						operation.run();
+					}
+					return true;
+				}
+			}
+			else {
+				ChannelOperations<?, ?> ops = conn.as(ChannelOperations.class);
+				if (ops instanceof HttpClientResponse) {
+					TraceContext parent = TracingHandlerUtil
+							.traceContext(((HttpClientResponse) ops).currentContextView());
+					if (parent != null) {
+						try (CurrentTraceContext.Scope scope = currentTraceContext.maybeScope(parent)) {
+							operation.run();
+						}
+						return true;
+					}
+				}
+			}
+		}
+		return false;
+	}
+
+}

spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingChannelOutboundHandler.java

@@ -0,0 +1,202 @@
+/*
+ * Copyright 2013-2023 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.brave.instrument.reactor.netty;
+
+import java.net.SocketAddress;
+
+import brave.Span;
+import brave.propagation.CurrentTraceContext;
+import brave.propagation.TraceContext;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.util.AttributeKey;
+import reactor.netty.Connection;
+import reactor.netty.ConnectionObserver;
+import reactor.netty.channel.ChannelOperations;
+import reactor.netty.http.client.HttpClientRequest;
+
+/**
+ * {@link ChannelOutboundHandlerAdapter} that wraps all events in scope.
+ * <p>
+ * WARNING: Using this feature can lead to serious performance issues. This should be only
+ * used for debugging purposes.
+ *
+ * @since 3.1.9
+ */
+public class TracingChannelOutboundHandler extends ChannelOutboundHandlerAdapter {
+
+	static final AttributeKey<Span> SPAN_ATTRIBUTE_KEY = AttributeKey.valueOf(Span.class.getName());
+
+	final CurrentTraceContext currentTraceContext;
+
+	/**
+	 * Creates a new instance of {@link TracingChannelOutboundHandler}.
+	 * @param currentTraceContext current trace context
+	 */
+	public TracingChannelOutboundHandler(CurrentTraceContext currentTraceContext) {
+		this.currentTraceContext = currentTraceContext;
+	}
+
+	@Override
+	public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
+		if (instrumentOperation(ctx, () -> ctx.bind(localAddress, promise))) {
+			return;
+		}
+
+		ctx.bind(localAddress, promise);
+	}
+
+	@Override
+	public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress,
+			ChannelPromise promise) {
+		if (instrumentOperation(ctx, () -> ctx.connect(remoteAddress, localAddress, promise))) {
+			return;
+		}
+
+		ctx.connect(remoteAddress, localAddress, promise);
+	}
+
+	@Override
+	public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) {
+		if (instrumentOperation(ctx, () -> ctx.disconnect(promise))) {
+			return;
+		}
+
+		ctx.disconnect(promise);
+	}
+
+	@Override
+	public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
+		if (instrumentOperation(ctx, () -> ctx.close(promise))) {
+			return;
+		}
+
+		ctx.close(promise);
+	}
+
+	@Override
+	public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) {
+		if (instrumentOperation(ctx, () -> ctx.deregister(promise))) {
+			return;
+		}
+
+		ctx.deregister(promise);
+	}
+
+	@Override
+	public void read(ChannelHandlerContext ctx) {
+		if (instrumentOperation(ctx, () -> ctx.read())) {
+			return;
+		}
+
+		ctx.read();
+	}
+
+	@Override
+	public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+		if (instrumentOperation(ctx, () -> ctx.write(msg, promise))) {
+			return;
+		}
+
+		ctx.write(msg, promise);
+	}
+
+	@Override
+	public void flush(ChannelHandlerContext ctx) {
+		if (instrumentOperation(ctx, () -> ctx.flush())) {
+			return;
+		}
+
+		ctx.flush();
+	}
+
+	@Override
+	public boolean isSharable() {
+		return false;
+	}
+
+	@Override
+	public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+		if (instrumentOperation(ctx, () -> {
+			try {
+				super.handlerAdded(ctx);
+			}
+			catch (Exception e) {
+				throw new RuntimeException(e);
+			}
+		})) {
+			return;
+		}
+
+		super.handlerAdded(ctx);
+	}
+
+	@Override
+	public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+		if (instrumentOperation(ctx, () -> {
+			try {
+				super.handlerRemoved(ctx);
+			}
+			catch (Exception e) {
+				throw new RuntimeException(e);
+			}
+		})) {
+			return;
+		}
+
+		super.handlerRemoved(ctx);
+	}
+
+	boolean instrumentOperation(ChannelHandlerContext ctx, Runnable operation) {
+		Span span = ctx.channel().attr(SPAN_ATTRIBUTE_KEY).get();
+		if (span != null) {
+			try (CurrentTraceContext.Scope scope = currentTraceContext.maybeScope(span.context())) {
+				operation.run();
+			}
+			return true;
+		}
+		else {
+			Connection conn = Connection.from(ctx.channel());
+			if (conn instanceof ConnectionObserver) {
+				TraceContext parent = ((ConnectionObserver) conn).currentContext().getOrDefault(TraceContext.class,
+						null);
+				if (parent != null) {
+					try (CurrentTraceContext.Scope scope = currentTraceContext.maybeScope(parent)) {
+						operation.run();
+					}
+					return true;
+				}
+			}
+			else {
+				ChannelOperations<?, ?> ops = conn.as(ChannelOperations.class);
+				if (ops instanceof HttpClientRequest) {
+					TraceContext traceContext = TracingHandlerUtil
+							.traceContext(((HttpClientRequest) ops).currentContextView());
+					if (traceContext != null) {
+						try (CurrentTraceContext.Scope scope = currentTraceContext.maybeScope(traceContext)) {
+							operation.run();
+						}
+						return true;
+					}
+				}
+			}
+		}
+		return false;
+	}
+
+}

spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingHandlerUtil.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2013-2023 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.brave.instrument.reactor.netty;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import brave.propagation.TraceContext;
+import reactor.util.context.ContextView;
+
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.brave.bridge.BraveTraceContext;
+import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth;
+
+final class TracingHandlerUtil {
+
+	private TracingHandlerUtil() {
+		throw new IllegalStateException("Can't instantiate a utility class");
+	}
+
+	static TraceContext traceContext(ContextView ctxView) {
+		AtomicReference<Span> pendingSpan = ReactorSleuth.getPendingSpan(ctxView);
+		if (pendingSpan != null) {
+			Span span = pendingSpan.get();
+			if (span != null) {
+				return BraveTraceContext.toBrave(span.context());
+			}
+		}
+		return null;
+	}
+
+}

tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/pom.xml

@@ -76,6 +76,26 @@
 			<groupId>org.awaitility</groupId>
 			<artifactId>awaitility</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>io.projectreactor.netty</groupId>
+			<artifactId>reactor-netty-http-brave</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>com.squareup.okhttp3</groupId>
+			<artifactId>mockwebserver</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>com.squareup.okhttp3</groupId>
+			<artifactId>okhttp</artifactId>
+			<version>${okhttp.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>com.github.tomakehurst</groupId>
+			<artifactId>wiremock-jre8-standalone</artifactId>
+			<version>${wiremock.version}</version>
+		</dependency>
 	</dependencies>
 
 </project>

tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/web/client/reactor/netty/LoggingTest.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2013-2023 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.brave.instrument.web.client.reactor.netty;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.client.TestRestTemplate;
+import org.springframework.cloud.gateway.filter.NettyRoutingFilter;
+import org.springframework.cloud.gateway.route.RouteLocator;
+import org.springframework.cloud.gateway.route.builder.RouteLocatorBuilder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.HttpStatus;
+
+import static org.assertj.core.api.BDDAssertions.then;
+
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT,
+		properties = { "spring.sleuth.reactor.netty.debug.enabled=true",
+				"spring.sleuth.reactor.instrumentation-type=decorate_queues", "spring.sleuth.sampler.probability=1.0",
+				"server.port=8554" })
+class LoggingTest {
+
+	private static final String TRACE_ID = "traceId";
+
+	private static final String SPAN_ID = "spanId";
+
+	private static final String ID = "1231231231231231";
+
+	private ListAppender<ILoggingEvent> appender;
+
+	@Autowired
+	private TestRestTemplate rest;
+
+	@BeforeEach
+	void init() {
+		rest.getRestTemplate().setInterceptors(Collections.singletonList((request, body, execution) -> {
+			request.getHeaders().add("x-b3-traceid", ID);
+			request.getHeaders().add("x-b3-spanid", ID);
+			request.getHeaders().add("Foo-Bar-Id", "123");
+			return execution.execute(request, body);
+		}));
+		appender = new ListAppender<>();
+		appender.start();
+		Logger root = (Logger) LoggerFactory.getLogger(org.slf4j.Logger.ROOT_LOGGER_NAME);
+		root.setLevel(Level.TRACE);
+		((Logger) LoggerFactory.getLogger("reactor.netty.http.client.HttpClientConnect")).addAppender(appender);
+		((Logger) LoggerFactory.getLogger(NettyRoutingFilter.class)).addAppender(appender);
+	}
+
+	@Test
+	void should_properly_fill_out_mdc_context() {
+		then(this.rest.getForEntity("http://localhost:8554/headers", String.class).getStatusCode())
+				.isEqualTo(HttpStatus.OK);
+		then(appender.list).as("No logs to process").isNotEmpty();
+		appender.list.forEach(le -> {
+			Map<String, String> mdc = le.getMDCPropertyMap();
+			then(mdc).as("TraceId does not exist for record: " + le).containsKey(TRACE_ID);
+			then(mdc.get(TRACE_ID)).as("TraceId did not match").isEqualTo(ID);
+			then(mdc).as("SpanId does not exist for record: " + le).containsKey(SPAN_ID);
+			then(mdc).as("SpanId did not match").isEqualTo(SPAN_ID);
+		});
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableAutoConfiguration
+	static class Config {
+
+		MockWebServer server = new MockWebServer();
+
+		@PostConstruct
+		void setup() throws IOException {
+			server.start();
+			server.enqueue(new MockResponse());
+		}
+
+		@Bean
+		RouteLocator builder(RouteLocatorBuilder builder) {
+			return builder.routes().route("test_route",
+					r -> r.path("/headers/**").uri("http://localhost:" + server.getPort() + "/foo")).build();
+		}
+
+		@PreDestroy
+		void cleanup() throws IOException {
+			server.close();
+		}
+
+	}
+
+}

tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/web/client/reactor/netty/ReuseTraceIdTest.java

@@ -0,0 +1,195 @@
+/*
+ * Copyright 2013-2023 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.brave.instrument.web.client.reactor.netty;
+
+import java.net.URL;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+
+import brave.Span;
+import brave.Tracer;
+import brave.Tracer.SpanInScope;
+import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.client.WireMock;
+import com.github.tomakehurst.wiremock.extension.responsetemplating.ResponseTemplateTransformer;
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.Response;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.gateway.route.RouteLocator;
+import org.springframework.cloud.gateway.route.builder.RouteLocatorBuilder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.options;
+import static org.junit.jupiter.api.Assertions.fail;
+
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT,
+		properties = { "spring.sleuth.reactor.netty.debug.enabled=true",
+				"spring.sleuth.reactor.instrumentation-type=decorate_queues", "spring.sleuth.sampler.probability=1.0",
+				"server.port=8553" })
+class ReuseTraceIdTest {
+
+	private static final String TOKEN = "eyJhbGciOiJSUzI1NiIsImtpZCI6ImdpUFkxeHZYb0taTVN3eDcvV1dHSUpQQjByTSJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiYWRtaW4iOnRydWUsImV4cCI6MTQ4OTA4MTA3OTc5OTB9.K-Bp_9lAW2W9PjmXGdGwrEID-dvamTmZwwndcQvO8mJJgW_PIo76_BYO-Ncstb_vPnxdG2Re9X_kEjve-i5cymdIkoYczPlryg-QxgLa1ZUt0-7FkLhWbGxghNLxk2k5vdcS07OwOG6wdDhoEvv_49h05p4FKG2Re5dskJIXRKzvmBYddqJrBDJsfYT0UGB94oVKnOQtI7mEB4Q1XpKz5NqYMN_HWZqEF5MHBBLbsIxCpHD6zOeJNppl7BSywFyJMRp-eSBwKlsR3eSX_jMDuM13Eaf3h3yd2pKDIxtValh822xaL9GnDq-YeAxmQrEg8o6tN_r1WRcoS8-cgqw3Ng";
+
+	@Autowired
+	private Tracer tracer;
+
+	private final Set<String> traceIds = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+	private final Set<String> previousTraceIds = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+	private final Map<String, Boolean> failedTraceIds = new ConcurrentHashMap<>();
+
+	private final Set<Exception> exceptions = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+	private final OkHttpClient client = new OkHttpClient.Builder().readTimeout(10, TimeUnit.SECONDS).build();
+
+	@Test
+	void should_not_reuse_traces() throws InterruptedException {
+		traceIds.clear();
+		failedTraceIds.clear();
+		exceptions.clear();
+		int threads = 10;
+		ExecutorService pool = Executors.newFixedThreadPool(threads);
+		for (int i = 0; i < threads; i++) {
+			pool.submit(this::testLoop);
+		}
+		pool.shutdown();
+		pool.awaitTermination(1, TimeUnit.MINUTES);
+		if (!failedTraceIds.isEmpty()) {
+			System.out.println("The following trace ids failed:");
+			failedTraceIds.forEach((id, flag) -> System.out.println("id=" + id + ", previous=" + flag));
+			System.out.println("Previous trace ids:");
+			previousTraceIds.forEach(System.out::println);
+			System.out.println("trace ids: " + traceIds.size());
+			System.out.println("failed trace ids: " + failedTraceIds.size());
+			System.out.println("previous ids: " + previousTraceIds.size());
+			fail();
+		}
+		if (!exceptions.isEmpty()) {
+			System.out.println("The following exceptions occurred:");
+			exceptions.forEach(Exception::printStackTrace);
+			fail();
+		}
+	}
+
+	private void testLoop() {
+		Random rand = new Random();
+		int iterations = 100;
+		for (int i = 0; i < iterations; i++) {
+			testOneTrace();
+			try {
+				Thread.sleep(rand.nextInt(100));
+			}
+			catch (InterruptedException e) {
+				return;
+			}
+		}
+	}
+
+	private void testOneTrace() {
+		Span span = tracer.newTrace();
+		try (SpanInScope ws = tracer.withSpanInScope(span)) {
+			String traceId = tracer.currentSpan().context().traceIdString();
+			traceIds.add(traceId);
+			System.out.println("testing traceId= " + traceId);
+			String spanId = tracer.currentSpan().context().spanIdString();
+			URL url = new URL("http://localhost:8553/headers");
+			Request request = new Request.Builder().url(url).header("X-B3-TraceId", traceId)
+					.header("X-B3-SpanId", spanId).header("Authorization", "Bearer " + TOKEN)
+					.header("Content-Type", "application/json").build();
+
+			Call call = client.newCall(request);
+			try (Response clientResponse = call.execute()) {
+				int status = clientResponse.code();
+				if (status != 200) {
+					throw new RuntimeException("Request failed with status " + status);
+				}
+
+				String responseTraceId = clientResponse.headers().get("X-DOX-TraceId");
+				if (!traceId.equals(responseTraceId)) {
+					System.out.println("error for traceId= " + traceId);
+					boolean previous = traceIds.contains(responseTraceId);
+					if (previous) {
+						previousTraceIds.add(responseTraceId);
+						System.out.println("received for " + traceId + " a previous trace id " + responseTraceId);
+					}
+					else {
+						System.out.println("received an unknown trace id: " + responseTraceId);
+					}
+					failedTraceIds.put(traceId, previous);
+				}
+			}
+		}
+		catch (Exception ex) {
+			ex.printStackTrace();
+			exceptions.add(ex);
+		}
+		finally {
+			span.finish();
+		}
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableAutoConfiguration
+	static class Config {
+
+		WireMockServer wireMockServer = new WireMockServer(
+				options().dynamicPort().extensions(new ResponseTemplateTransformer(false))) {
+			{
+				start();
+			}
+		};
+
+		@PostConstruct
+		void setup() {
+			wireMockServer.stubFor(WireMock.get("/headers")
+					.willReturn(WireMock.aResponse().withHeader("x-dox-traceId", "{{request.headers.x-b3-traceid}}")
+							.withTransformers("response-template")));
+		}
+
+		@PreDestroy
+		void clean() {
+			wireMockServer.shutdown();
+		}
+
+		@Bean
+		RouteLocator builder(RouteLocatorBuilder builder) {
+			return builder.routes()
+					.route("test_route",
+							r -> r.path("/headers/**").uri("http://localhost:" + wireMockServer.port() + "/headers"))
+					.build();
+		}
+
+	}
+
+}

tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/src/test/resources/application.yml

@@ -1,5 +1,5 @@
-logging.level.org.springframework.cloud: DEBUG
-logging.level.com.netflix.discovery.InstanceInfoReplicator: ERROR
-logging.level.org.springframework.cloud.sleuth.instrument.web.client.feign: TRACE
+logging.level:
+  org.springframework.cloud: TRACE
+  reactor.netty: TRACE
 
-spring.autoconfigure.exclude: org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration, org.springframework.boot.autoconfigure.orm.jpa.HibernateJpaAutoConfiguration, org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration, org.springframework.boot.autoconfigure.data.web.SpringDataWebAutoConfiguration, org.springframework.cloud.gateway.config.GatewayAutoConfiguration, org.springframework.cloud.gateway.config.GatewayClassPathWarningAutoConfiguration, org.springframework.cloud.gateway.config.GatewayMetricsAutoConfiguration
+spring.autoconfigure.exclude: org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration, org.springframework.boot.autoconfigure.orm.jpa.HibernateJpaAutoConfiguration, org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration, org.springframework.boot.autoconfigure.data.web.SpringDataWebAutoConfiguration, org.springframework.cloud.gateway.config.GatewayClassPathWarningAutoConfiguration, org.springframework.cloud.gateway.config.GatewayMetricsAutoConfiguration