From b594ad5b6f1d78628b37747af1f96c1bf02ae02a Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Fri, 2 Jun 2023 19:49:21 +0200 Subject: [PATCH] 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 --- pom.xml | 1 + spring-cloud-sleuth-autoconfigure/pom.xml | 5 + .../BraveReactorNettyAutoConfiguration.java | 85 +++++++ ...itional-spring-configuration-metadata.json | 6 + .../main/resources/META-INF/spring.factories | 1 + ...aveReactorNettyAutoConfigurationTests.java | 70 ++++++ .../netty/TracingChannelInboundHandler.java | 207 ++++++++++++++++++ .../netty/TracingChannelOutboundHandler.java | 202 +++++++++++++++++ .../reactor/netty/TracingHandlerUtil.java | 45 ++++ .../pom.xml | 20 ++ .../web/client/reactor/netty/LoggingTest.java | 121 ++++++++++ .../reactor/netty/ReuseTraceIdTest.java | 195 +++++++++++++++++ .../src/test/resources/application.yml | 8 +- 13 files changed, 962 insertions(+), 4 deletions(-) create mode 100644 spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/reactor/netty/BraveReactorNettyAutoConfiguration.java create mode 100644 spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/reactor/netty/BraveReactorNettyAutoConfigurationTests.java create mode 100644 spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingChannelInboundHandler.java create mode 100644 spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingChannelOutboundHandler.java create mode 100644 spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingHandlerUtil.java create mode 100644 tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/web/client/reactor/netty/LoggingTest.java create mode 100644 tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/web/client/reactor/netty/ReuseTraceIdTest.java diff --git a/pom.xml b/pom.xml index 5d3ffadd7..dfd9b9208 100644 --- a/pom.xml +++ b/pom.xml @@ -77,6 +77,7 @@ 10.0.6 2.8.0 1.6.21 + 2.35.0 true diff --git a/spring-cloud-sleuth-autoconfigure/pom.xml b/spring-cloud-sleuth-autoconfigure/pom.xml index 00fcce42e..016297993 100644 --- a/spring-cloud-sleuth-autoconfigure/pom.xml +++ b/spring-cloud-sleuth-autoconfigure/pom.xml @@ -338,6 +338,11 @@ reactor-kafka true + + io.projectreactor.netty + reactor-netty-http-brave + true + io.github.lognet diff --git a/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/reactor/netty/BraveReactorNettyAutoConfiguration.java b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/reactor/netty/BraveReactorNettyAutoConfiguration.java new file mode 100644 index 000000000..3293e78b9 --- /dev/null +++ b/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)); + }); + } + + } + +} diff --git a/spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json index 622184304..1e07fb69a 100644 --- a/spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json +++ b/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 } ] } diff --git a/spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/spring.factories b/spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/spring.factories index 040841570..d9ce63527 100644 --- a/spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/spring.factories +++ b/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 diff --git a/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/reactor/netty/BraveReactorNettyAutoConfigurationTests.java b/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/reactor/netty/BraveReactorNettyAutoConfigurationTests.java new file mode 100644 index 000000000..7b9fd1b72 --- /dev/null +++ b/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)); + } + +} diff --git a/spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingChannelInboundHandler.java b/spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingChannelInboundHandler.java new file mode 100644 index 000000000..879968d55 --- /dev/null +++ b/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. + *

+ * 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_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; + } + +} diff --git a/spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingChannelOutboundHandler.java b/spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingChannelOutboundHandler.java new file mode 100644 index 000000000..4f9db8030 --- /dev/null +++ b/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. + *

+ * 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_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; + } + +} diff --git a/spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingHandlerUtil.java b/spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingHandlerUtil.java new file mode 100644 index 000000000..139e72b17 --- /dev/null +++ b/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 pendingSpan = ReactorSleuth.getPendingSpan(ctxView); + if (pendingSpan != null) { + Span span = pendingSpan.get(); + if (span != null) { + return BraveTraceContext.toBrave(span.context()); + } + } + return null; + } + +} diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/pom.xml b/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/pom.xml index 22a70ba37..8194f10d9 100644 --- a/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/pom.xml +++ b/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/pom.xml @@ -76,6 +76,26 @@ org.awaitility awaitility + + io.projectreactor.netty + reactor-netty-http-brave + + + com.squareup.okhttp3 + mockwebserver + test + + + com.squareup.okhttp3 + okhttp + ${okhttp.version} + test + + + com.github.tomakehurst + wiremock-jre8-standalone + ${wiremock.version} + diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/web/client/reactor/netty/LoggingTest.java b/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/web/client/reactor/netty/LoggingTest.java new file mode 100644 index 000000000..f5ac5103d --- /dev/null +++ b/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 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 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(); + } + + } + +} diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/web/client/reactor/netty/ReuseTraceIdTest.java b/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/web/client/reactor/netty/ReuseTraceIdTest.java new file mode 100644 index 000000000..92c97d7bb --- /dev/null +++ b/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 traceIds = Collections.newSetFromMap(new ConcurrentHashMap<>()); + + private final Set previousTraceIds = Collections.newSetFromMap(new ConcurrentHashMap<>()); + + private final Map failedTraceIds = new ConcurrentHashMap<>(); + + private final Set 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(); + } + + } + +} diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/src/test/resources/application.yml b/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/src/test/resources/application.yml index b1f8a8d34..b5b0f2487 100644 --- a/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/src/test/resources/application.yml +++ b/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