From 4acbaf38f320e2c2ec8b57256b9c99c04109942c Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Fri, 31 Jan 2020 15:48:19 +0800 Subject: [PATCH 1/7] Refactors some of the reactor code (#1541) --- .../reactor/LazySpanSubscriber.java | 1 + .../instrument/reactor/ReactorSleuth.java | 29 +++++++------------ .../reactor/SpanSubscriptionProvider.java | 17 +++++------ .../SpanSubscriptionProviderTests.java | 12 ++++++-- 4 files changed, 29 insertions(+), 30 deletions(-) diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java index 576c6ffc0..667080201 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java @@ -29,6 +29,7 @@ import reactor.util.context.Context; * @author Marcin Grzejszczak * @since 2.0.0 */ +// TODO: why are we extending AtomicBoolean and not actually using its methods? final class LazySpanSubscriber extends AtomicBoolean implements SpanSubscription { private final Supplier> supplier; diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java index 10df1a826..a1746e0ce 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java @@ -18,7 +18,6 @@ package org.springframework.cloud.sleuth.instrument.reactor; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import java.util.function.BooleanSupplier; import java.util.function.Function; import brave.Tracing; @@ -67,24 +66,21 @@ public abstract class ReactorSleuth { log.trace("Scope passing operator [" + beanFactory + "]"); } - // Adapt if lazy bean factory - BooleanSupplier isActive = beanFactory instanceof ConfigurableApplicationContext - ? ((ConfigurableApplicationContext) beanFactory)::isActive : () -> true; - return Operators.liftPublisher((p, sub) -> { - // if Flux/Mono #just, #empty, #error + // While supply of scalar types may be deferred, we don't currently scope + // production of values in a trace context. This prevents excessive overhead + // when using constant results such as Flux/Mono #just, #empty, #error if (p instanceof Fuseable.ScalarCallable) { return sub; } - Scannable scannable = Scannable.from(p); - // rest of the logic unchanged... - if (isActive.getAsBoolean()) { + + if (beanFactory instanceof ConfigurableApplicationContext + && ((ConfigurableApplicationContext) beanFactory).isActive()) { if (log.isTraceEnabled()) { log.trace("Spring Context [" + beanFactory + "] already refreshed. Creating a scope " + "passing span subscriber with Reactor Context " + "[" - + sub.currentContext() + "] and name [" + scannable.name() - + "]"); + + sub.currentContext() + "] and name [" + name(sub) + "]"); } return scopePassingSpanSubscription(beanFactory, sub); @@ -92,18 +88,15 @@ public abstract class ReactorSleuth { if (log.isTraceEnabled()) { log.trace("Spring Context [" + beanFactory + "] is not yet refreshed, falling back to lazy span subscriber. Reactor Context is [" - + sub.currentContext() + "] and name is [" + scannable.name() - + "]"); + + sub.currentContext() + "] and name is [" + name(sub) + "]"); } return new LazySpanSubscriber<>( - lazyScopePassingSpanSubscription(beanFactory, scannable, sub)); + new SpanSubscriptionProvider<>(beanFactory, sub)); }); } - static SpanSubscriptionProvider lazyScopePassingSpanSubscription( - BeanFactory beanFactory, Scannable scannable, CoreSubscriber sub) { - return new SpanSubscriptionProvider<>(beanFactory, sub, sub.currentContext(), - scannable.name()); + static String name(CoreSubscriber sub) { + return Scannable.from(sub).name(); } private static Map CACHE = new ConcurrentHashMap<>(); diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java index e90b34836..2c79e5975 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java @@ -22,11 +22,13 @@ import brave.propagation.CurrentTraceContext; import brave.propagation.TraceContext; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.reactivestreams.Subscriber; +import reactor.core.CoreSubscriber; import reactor.util.context.Context; import org.springframework.beans.factory.BeanFactory; +import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.name; + /** * Supplier to lazily start a {@link SpanSubscription}. * @@ -39,23 +41,20 @@ final class SpanSubscriptionProvider implements Supplier> final BeanFactory beanFactory; - final Subscriber subscriber; + final CoreSubscriber subscriber; final Context context; - final String name; - private volatile CurrentTraceContext currentTraceContext; - SpanSubscriptionProvider(BeanFactory beanFactory, Subscriber subscriber, - Context context, String name) { + SpanSubscriptionProvider(BeanFactory beanFactory, + CoreSubscriber subscriber) { this.beanFactory = beanFactory; this.subscriber = subscriber; - this.context = context; - this.name = name; + this.context = subscriber.currentContext(); if (log.isTraceEnabled()) { log.trace("Spring context [" + beanFactory + "], Reactor context [" + context - + "], name [" + name + "]"); + + "], name [" + name(subscriber) + "]"); } } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java index 6838e4f3a..2146941f5 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java @@ -19,6 +19,7 @@ package org.springframework.cloud.sleuth.instrument.reactor; import org.assertj.core.api.BDDAssertions; import org.junit.Test; import org.mockito.BDDMockito; +import reactor.core.CoreSubscriber; import reactor.util.context.Context; import org.springframework.beans.factory.BeanFactory; @@ -27,13 +28,18 @@ public class SpanSubscriptionProviderTests { @Test public void should_return_default_tracing_instance_when_exception_thrown_upon_bean_retrieval() { + CoreSubscriber subscriber = BDDMockito.mock(CoreSubscriber.class); + BDDMockito.when(subscriber.currentContext()).thenReturn(Context.empty()); + BeanFactory beanFactory = BDDMockito.mock(BeanFactory.class); + BDDMockito.when(beanFactory.getBean(BDDMockito.any(Class.class))) .thenThrow(new IllegalStateException()); - SpanSubscriptionProvider provider = new SpanSubscriptionProvider(beanFactory, - null, Context.empty(), "example"); - SpanSubscription spanSubscription = provider.get(); + SpanSubscriptionProvider provider = new SpanSubscriptionProvider<>( + beanFactory, subscriber); + + SpanSubscription spanSubscription = provider.get(); BDDAssertions.then(spanSubscription).isNotNull(); } From 241f536aefdebb50e6d428f9fa403bcf8a41ed30 Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Mon, 3 Feb 2020 19:26:38 +0800 Subject: [PATCH 2/7] Uses ConfigurableApplicationContext consistently in Reactor (#1545) Before, we used ConfigurableApplicationContext or BeanFactory eventhough we already had a reference to ConfigurableApplicationContext. This uses the latter consistently, avoiding a state condition that caused more code. This also corrects some misnamed tests and adjusts them to verify only what they are responsible for. --- .../reactor/LazySpanSubscriber.java | 76 ---- .../instrument/reactor/ReactorSleuth.java | 47 +-- .../reactor/ScopePassingSpanSubscriber.java | 6 + .../reactor/SpanSubscriptionProvider.java | 91 ----- .../TraceReactorAutoConfiguration.java | 30 +- .../TraceWebClientAutoConfiguration.java | 5 +- .../TraceWebClientBeanPostProcessor.java | 33 +- .../SpanSubscriptionProviderTests.java | 47 --- .../TraceWebClientBeanPostProcessorTest.java | 8 +- ...ePassingSpanSubscriberSpringBootTests.java | 204 +++++++++++ .../ScopePassingSpanSubscriberTests.java | 129 ++++++- .../reactor/SpanSubscriberTests.java | 333 ------------------ 12 files changed, 384 insertions(+), 625 deletions(-) delete mode 100644 spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java delete mode 100644 spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java delete mode 100644 spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java create mode 100644 tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberSpringBootTests.java delete mode 100644 tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java deleted file mode 100644 index 667080201..000000000 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java +++ /dev/null @@ -1,76 +0,0 @@ -/* - * Copyright 2013-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.sleuth.instrument.reactor; - -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.function.Supplier; - -import org.reactivestreams.Subscription; -import reactor.util.context.Context; - -/** - * A lazy representation of the {@link SpanSubscription}. - * - * @param of what subscription returns - * @author Marcin Grzejszczak - * @since 2.0.0 - */ -// TODO: why are we extending AtomicBoolean and not actually using its methods? -final class LazySpanSubscriber extends AtomicBoolean implements SpanSubscription { - - private final Supplier> supplier; - - LazySpanSubscriber(Supplier> supplier) { - this.supplier = supplier; - } - - @Override - public void onSubscribe(Subscription subscription) { - this.supplier.get().onSubscribe(subscription); - } - - @Override - public void request(long n) { - this.supplier.get().request(n); - } - - @Override - public void cancel() { - this.supplier.get().cancel(); - } - - @Override - public void onNext(T o) { - this.supplier.get().onNext(o); - } - - @Override - public void onError(Throwable throwable) { - this.supplier.get().onError(throwable); - } - - @Override - public void onComplete() { - this.supplier.get().onComplete(); - } - - @Override - public Context currentContext() { - return this.supplier.get().currentContext(); - } - -} diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java index a1746e0ce..73a212a16 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java @@ -32,7 +32,6 @@ import reactor.core.Scannable; import reactor.core.publisher.Operators; import reactor.util.context.Context; -import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ConfigurableApplicationContext; /** @@ -55,15 +54,20 @@ public abstract class ReactorSleuth { * {@link reactor.core.publisher.Hooks#onLastOperator(Function)} or * {@link reactor.core.publisher.Hooks#onLastOperator(Function)}. The Span operator * pointcut will pass the Scope of the Span without ever creating any new spans. - * @param beanFactory - {@link BeanFactory} + * @param springContext the Spring context. * @param an arbitrary type that is left unchanged by the span operator * @return a new lazy span operator pointcut */ - @SuppressWarnings("unchecked") + // Much of Boot assumes that the Spring context will be a + // ConfigurableApplicationContext, rooted in SpringApplication's + // requirement for it to be so. Previous versions of Reactor + // instrumentation injected both BeanFactory and also + // ConfigurableApplicationContext. This chooses the more narrow + // signature as it is simpler than explaining instanceof checks. public static Function, ? extends Publisher> scopePassingSpanOperator( - BeanFactory beanFactory) { + ConfigurableApplicationContext springContext) { if (log.isTraceEnabled()) { - log.trace("Scope passing operator [" + beanFactory + "]"); + log.trace("Scope passing operator [" + springContext + "]"); } return Operators.liftPublisher((p, sub) -> { @@ -74,24 +78,23 @@ public abstract class ReactorSleuth { return sub; } - if (beanFactory instanceof ConfigurableApplicationContext - && ((ConfigurableApplicationContext) beanFactory).isActive()) { + if (!springContext.isActive()) { if (log.isTraceEnabled()) { - log.trace("Spring Context [" + beanFactory - + "] already refreshed. Creating a scope " - + "passing span subscriber with Reactor Context " + "[" - + sub.currentContext() + "] and name [" + name(sub) + "]"); + log.trace("Spring Context [" + springContext + + "] is not yet refreshed. This is unexpected. Reactor Context is [" + + sub.currentContext() + "] and name is [" + name(sub) + "]"); } + assert false; // should never happen, but don't break. + return sub; + } - return scopePassingSpanSubscription(beanFactory, sub); - } if (log.isTraceEnabled()) { - log.trace("Spring Context [" + beanFactory - + "] is not yet refreshed, falling back to lazy span subscriber. Reactor Context is [" - + sub.currentContext() + "] and name is [" + name(sub) + "]"); + log.trace("Spring Context [" + springContext + + "] Creating a scope passing span subscriber with Reactor Context " + + "[" + sub.currentContext() + "] and name [" + name(sub) + "]"); } - return new LazySpanSubscriber<>( - new SpanSubscriptionProvider<>(beanFactory, sub)); + + return scopePassingSpanSubscription(springContext, sub); }); } @@ -99,12 +102,12 @@ public abstract class ReactorSleuth { return Scannable.from(sub).name(); } - private static Map CACHE = new ConcurrentHashMap<>(); + private static Map CACHE = new ConcurrentHashMap<>(); static CoreSubscriber scopePassingSpanSubscription( - BeanFactory beanFactory, CoreSubscriber sub) { - CurrentTraceContext currentTraceContext = CACHE.computeIfAbsent(beanFactory, - beanFactory1 -> beanFactory1.getBean(CurrentTraceContext.class)); + ConfigurableApplicationContext springContext, CoreSubscriber sub) { + CurrentTraceContext currentTraceContext = CACHE.computeIfAbsent(springContext, + springContext1 -> springContext1.getBean(CurrentTraceContext.class)); Context context = sub.currentContext(); TraceContext parent = context.getOrDefault(TraceContext.class, null); diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java index 488e26632..aa76125e4 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java @@ -120,4 +120,10 @@ final class ScopePassingSpanSubscriber implements SpanSubscription, Scanna } } + @Override + public String toString() { + return "ScopePassingSpanSubscriber{" + "subscriber=" + this.subscriber + + ", parent=" + this.parent + "}"; + } + } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java deleted file mode 100644 index 2c79e5975..000000000 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java +++ /dev/null @@ -1,91 +0,0 @@ -/* - * Copyright 2013-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.sleuth.instrument.reactor; - -import java.util.function.Supplier; - -import brave.propagation.CurrentTraceContext; -import brave.propagation.TraceContext; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import reactor.core.CoreSubscriber; -import reactor.util.context.Context; - -import org.springframework.beans.factory.BeanFactory; - -import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.name; - -/** - * Supplier to lazily start a {@link SpanSubscription}. - * - * @param type of returned subscription - * @author Marcin Grzejszczak - */ -final class SpanSubscriptionProvider implements Supplier> { - - private static final Log log = LogFactory.getLog(SpanSubscriptionProvider.class); - - final BeanFactory beanFactory; - - final CoreSubscriber subscriber; - - final Context context; - - private volatile CurrentTraceContext currentTraceContext; - - SpanSubscriptionProvider(BeanFactory beanFactory, - CoreSubscriber subscriber) { - this.beanFactory = beanFactory; - this.subscriber = subscriber; - this.context = subscriber.currentContext(); - if (log.isTraceEnabled()) { - log.trace("Spring context [" + beanFactory + "], Reactor context [" + context - + "], name [" + name(subscriber) + "]"); - } - } - - @Override - public SpanSubscription get() { - return newCoreSubscriber(currentTraceContext()); - } - - SpanSubscription newCoreSubscriber(CurrentTraceContext currentTraceContext) { - TraceContext root = this.context.hasKey(TraceContext.class) - ? this.context.get(TraceContext.class) : currentTraceContext.get(); - return new ScopePassingSpanSubscriber<>(this.subscriber, this.context, - currentTraceContext, root); - } - - private CurrentTraceContext currentTraceContext() { - if (this.currentTraceContext == null) { - try { - this.currentTraceContext = this.beanFactory - .getBean(CurrentTraceContext.class); - } - catch (Exception ex) { - if (log.isDebugEnabled()) { - log.debug( - "Exception occurred while trying to get the currentTraceContext bean. Will return a default instance", - ex); - } - return CurrentTraceContext.Default.create(); - } - } - return this.currentTraceContext; - } - -} diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java index c0c6c6ed4..f28ddba1c 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java @@ -25,8 +25,6 @@ import reactor.core.publisher.Hooks; import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.BeanDefinitionRegistry; @@ -76,14 +74,14 @@ public class TraceReactorAutoConfiguration { private static final Log log = LogFactory.getLog(TraceReactorConfiguration.class); @Autowired - BeanFactory beanFactory; + ConfigurableApplicationContext springContext; @PreDestroy public void cleanupHooks() { if (log.isTraceEnabled()) { log.trace("Cleaning up hooks"); } - SleuthReactorProperties reactorProperties = this.beanFactory + SleuthReactorProperties reactorProperties = this.springContext .getBean(SleuthReactorProperties.class); if (reactorProperties.isDecorateOnEach()) { if (log.isTraceEnabled()) { @@ -175,26 +173,24 @@ class HookRegisteringBeanDefinitionRegistryPostProcessor private static final Log log = LogFactory .getLog(HookRegisteringBeanDefinitionRegistryPostProcessor.class); - private final ConfigurableApplicationContext context; + final ConfigurableApplicationContext springContext; HookRegisteringBeanDefinitionRegistryPostProcessor( - ConfigurableApplicationContext context) { - this.context = context; + ConfigurableApplicationContext springContext) { + this.springContext = springContext; } @Override - public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) - throws BeansException { + public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) { } @Override - public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) - throws BeansException { - setupHooks(beanFactory); + public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) { + setupHooks(this.springContext); } - void setupHooks(BeanFactory beanFactory) { - ConfigurableEnvironment environment = this.context.getEnvironment(); + void setupHooks(ConfigurableApplicationContext springContext) { + ConfigurableEnvironment environment = springContext.getEnvironment(); boolean decorateOnEach = environment.getProperty( "spring.sleuth.reactor.decorate-on-each", Boolean.class, true); if (decorateOnEach) { @@ -202,20 +198,20 @@ class HookRegisteringBeanDefinitionRegistryPostProcessor log.trace("Decorating onEach operator instrumentation"); } Hooks.onEachOperator(SLEUTH_TRACE_REACTOR_KEY, - ReactorSleuth.scopePassingSpanOperator(this.context)); + ReactorSleuth.scopePassingSpanOperator(springContext)); } else { if (log.isTraceEnabled()) { log.trace("Decorating onLast operator instrumentation"); } Hooks.onLastOperator(SLEUTH_TRACE_REACTOR_KEY, - ReactorSleuth.scopePassingSpanOperator(this.context)); + ReactorSleuth.scopePassingSpanOperator(springContext)); } Schedulers.setExecutorServiceDecorator( TraceReactorAutoConfiguration.SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY, (scheduler, scheduledExecutorService) -> new TraceableScheduledExecutorService( - beanFactory, scheduledExecutorService)); + springContext, scheduledExecutorService)); } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java index c78114d84..ef596dddf 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java @@ -43,6 +43,7 @@ import org.springframework.boot.web.client.RestTemplateCustomizer; import org.springframework.cloud.commons.httpclient.HttpClientConfiguration; import org.springframework.cloud.gateway.filter.headers.HttpHeadersFilter; import org.springframework.cloud.sleuth.instrument.web.TraceWebServletAutoConfiguration; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.annotation.Order; @@ -133,8 +134,8 @@ public class TraceWebClientAutoConfiguration { @Bean static TraceWebClientBeanPostProcessor traceWebClientBeanPostProcessor( - BeanFactory beanFactory) { - return new TraceWebClientBeanPostProcessor(beanFactory); + ConfigurableApplicationContext springContext) { + return new TraceWebClientBeanPostProcessor(springContext); } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java index 8cafff570..558f384cf 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java @@ -38,9 +38,8 @@ import reactor.util.annotation.Nullable; import reactor.util.context.Context; import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.BeanPostProcessor; -import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.web.client.RestClientException; import org.springframework.web.reactive.function.client.ClientRequest; @@ -49,6 +48,8 @@ import org.springframework.web.reactive.function.client.ExchangeFilterFunction; import org.springframework.web.reactive.function.client.ExchangeFunction; import org.springframework.web.reactive.function.client.WebClient; +import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.scopePassingSpanOperator; + /** * {@link BeanPostProcessor} to wrap a {@link WebClient} instance into its trace * representation. @@ -58,10 +59,10 @@ import org.springframework.web.reactive.function.client.WebClient; */ final class TraceWebClientBeanPostProcessor implements BeanPostProcessor { - private final BeanFactory beanFactory; + private final ConfigurableApplicationContext springContext; - TraceWebClientBeanPostProcessor(BeanFactory beanFactory) { - this.beanFactory = beanFactory; + TraceWebClientBeanPostProcessor(ConfigurableApplicationContext springContext) { + this.springContext = springContext; } @Override @@ -92,7 +93,7 @@ final class TraceWebClientBeanPostProcessor implements BeanPostProcessor { return functions -> { boolean noneMatch = noneMatchTraceExchangeFunction(functions); if (noneMatch) { - functions.add(new TraceExchangeFilterFunction(this.beanFactory)); + functions.add(new TraceExchangeFilterFunction(this.springContext)); } }; } @@ -134,7 +135,7 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { private static final String CANCELLED_SUBSCRIPTION_ERROR = "CANCELLED"; - final BeanFactory beanFactory; + final ConfigurableApplicationContext springContext; final Function, ? extends Publisher> scopePassingTransformer; @@ -146,14 +147,14 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { TraceContext.Injector injector; - TraceExchangeFilterFunction(BeanFactory beanFactory) { - this.beanFactory = beanFactory; - this.scopePassingTransformer = ReactorSleuth - .scopePassingSpanOperator(beanFactory); + TraceExchangeFilterFunction(ConfigurableApplicationContext springContext) { + this.springContext = springContext; + this.scopePassingTransformer = scopePassingSpanOperator(springContext); } - public static ExchangeFilterFunction create(BeanFactory beanFactory) { - return new TraceExchangeFilterFunction(beanFactory); + public static ExchangeFilterFunction create( + ConfigurableApplicationContext springContext) { + return new TraceExchangeFilterFunction(springContext); } @Override @@ -177,7 +178,7 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { HttpClientHandler handler() { if (this.handler == null) { this.handler = HttpClientHandler - .create(this.beanFactory.getBean(HttpTracing.class)); + .create(this.springContext.getBean(HttpTracing.class)); } return this.handler; } @@ -191,14 +192,14 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { HttpTracing httpTracing() { if (this.httpTracing == null) { - this.httpTracing = this.beanFactory.getBean(HttpTracing.class); + this.httpTracing = this.springContext.getBean(HttpTracing.class); } return this.httpTracing; } TraceContext.Injector injector() { if (this.injector == null) { - this.injector = this.beanFactory.getBean(HttpTracing.class).tracing() + this.injector = this.springContext.getBean(HttpTracing.class).tracing() .propagation().injector(SETTER); } return this.injector; diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java deleted file mode 100644 index 2146941f5..000000000 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java +++ /dev/null @@ -1,47 +0,0 @@ -/* - * Copyright 2013-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.sleuth.instrument.reactor; - -import org.assertj.core.api.BDDAssertions; -import org.junit.Test; -import org.mockito.BDDMockito; -import reactor.core.CoreSubscriber; -import reactor.util.context.Context; - -import org.springframework.beans.factory.BeanFactory; - -public class SpanSubscriptionProviderTests { - - @Test - public void should_return_default_tracing_instance_when_exception_thrown_upon_bean_retrieval() { - CoreSubscriber subscriber = BDDMockito.mock(CoreSubscriber.class); - BDDMockito.when(subscriber.currentContext()).thenReturn(Context.empty()); - - BeanFactory beanFactory = BDDMockito.mock(BeanFactory.class); - - BDDMockito.when(beanFactory.getBean(BDDMockito.any(Class.class))) - .thenThrow(new IllegalStateException()); - - SpanSubscriptionProvider provider = new SpanSubscriptionProvider<>( - beanFactory, subscriber); - - SpanSubscription spanSubscription = provider.get(); - - BDDAssertions.then(spanSubscription).isNotNull(); - } - -} diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java index db74c2a13..4bf00805c 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java @@ -22,7 +22,7 @@ import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; -import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.web.reactive.function.client.WebClient; /** @@ -32,12 +32,12 @@ import org.springframework.web.reactive.function.client.WebClient; public class TraceWebClientBeanPostProcessorTest { @Mock - BeanFactory beanFactory; + ConfigurableApplicationContext springContext; @Test public void should_add_filter_only_once_to_web_client() { TraceWebClientBeanPostProcessor processor = new TraceWebClientBeanPostProcessor( - this.beanFactory); + this.springContext); WebClient client = WebClient.create(); client = (WebClient) processor.postProcessAfterInitialization(client, "foo"); @@ -53,7 +53,7 @@ public class TraceWebClientBeanPostProcessorTest { @Test public void should_add_filter_only_once_to_web_client_via_builder() { TraceWebClientBeanPostProcessor processor = new TraceWebClientBeanPostProcessor( - this.beanFactory); + this.springContext); WebClient.Builder builder = WebClient.builder(); builder = (WebClient.Builder) processor.postProcessAfterInitialization(builder, diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberSpringBootTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberSpringBootTests.java new file mode 100644 index 000000000..ef0a6bbd8 --- /dev/null +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberSpringBootTests.java @@ -0,0 +1,204 @@ +/* + * Copyright 2013-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.instrument.reactor; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicReference; + +import brave.propagation.CurrentTraceContext; +import brave.propagation.CurrentTraceContext.Scope; +import brave.propagation.TraceContext; +import brave.sampler.Sampler; +import org.awaitility.Awaitility; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.BDDAssertions.then; + +/** + * Like {@link ScopePassingSpanSubscriberTests}, except this tests wiring with spring boot + * config. + */ +@RunWith(SpringRunner.class) +@SpringBootTest(classes = ScopePassingSpanSubscriberSpringBootTests.Config.class, + webEnvironment = SpringBootTest.WebEnvironment.NONE) +public class ScopePassingSpanSubscriberSpringBootTests { + + @Autowired + CurrentTraceContext currentTraceContext; + + TraceContext context = TraceContext.newBuilder().traceId(1).spanId(1).sampled(true) + .build(); + + TraceContext context2 = TraceContext.newBuilder().traceId(1).spanId(2).sampled(true) + .build(); + + @Test + public void should_pass_tracing_info_when_using_reactor() { + final AtomicReference spanInOperation = new AtomicReference<>(); + Publisher traced = Flux.just(1, 2, 3); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Flux.from(traced).map(d -> d + 1).map(d -> d + 1).map((d) -> { + spanInOperation.set(this.currentTraceContext.get()); + return d + 1; + }).map(d -> d + 1).subscribe(d -> { + }); + } + + then(this.currentTraceContext.get()).isNull(); + then(spanInOperation.get()).isEqualTo(context); + } + + @Test + public void should_support_reactor_fusion_optimization() { + final AtomicReference spanInOperation = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Mono.just(1).flatMap(d -> Flux.just(d + 1).collectList().map(p -> p.get(0))) + .map(d -> d + 1).map((d) -> { + spanInOperation.set(this.currentTraceContext.get()); + return d + 1; + }).map(d -> d + 1).subscribe(d -> { + }); + } + + then(this.currentTraceContext.get()).isNull(); + then(spanInOperation.get()).isEqualTo(context); + } + + @Test + public void should_pass_tracing_info_when_using_reactor_async() { + final AtomicReference spanInOperation = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.1") + .map(d -> d + 1).map(d -> d + 1) + .publishOn(Schedulers.newSingle("secondThread")).log("reactor.2") + .map((d) -> { + spanInOperation.set(this.currentTraceContext.get()); + return d + 1; + }).map(d -> d + 1).blockLast(); + + Awaitility.await() + .untilAsserted(() -> then(spanInOperation.get()).isEqualTo(context)); + then(this.currentTraceContext.get()).isEqualTo(context); + } + + then(this.currentTraceContext.get()).isNull(); + + try (Scope ws = this.currentTraceContext.newScope(context2)) { + Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.") + .map(d -> d + 1).map(d -> d + 1).map((d) -> { + spanInOperation.set(this.currentTraceContext.get()); + return d + 1; + }).map(d -> d + 1).blockLast(); + + then(this.currentTraceContext.get()).isEqualTo(context2); + then(spanInOperation.get()).isEqualTo(context2); + } + + then(this.currentTraceContext.get()).isNull(); + } + + @Test + public void onlyConsidersContextDuringSubscribe() { + Mono fromMono = Mono.fromCallable(this.currentTraceContext::get); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + then(fromMono.map(context -> context).block()).isNotNull(); + } + } + + @Test + public void checkTraceIdDuringZipOperation() { + final AtomicReference spanInOperation = new AtomicReference<>(); + final AtomicReference spanInZipOperation = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Mono.fromCallable(this.currentTraceContext::get).map(span -> span) + .doOnNext(spanInOperation::set) + .zipWith(Mono.fromCallable(this.currentTraceContext::get) + .map(span -> span).doOnNext(spanInZipOperation::set)) + .block(); + } + + then(spanInZipOperation).hasValue(context); + then(spanInOperation).hasValue(context); + } + + // #646 + @Test + public void should_work_for_mono_just_with_flat_map() { + try (Scope ws = this.currentTraceContext.newScope(context)) { + Mono.just("value1") + .flatMap(request -> Mono.just("value2").then(Mono.just("foo"))) + .map(a -> "qwe").block(); + } + } + + // #1030 + @Test + public void checkTraceIdFromSubscriberContext() { + final AtomicReference spanInSubscriberContext = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Mono.subscriberContext().map(context -> this.currentTraceContext.get()) + .doOnNext(spanInSubscriberContext::set).block(); + } + + then(spanInSubscriberContext).hasValue(context); // ok here + } + + @Test + public void should_pass_tracing_info_into_inner_publishers() { + final AtomicReference spanInOperation = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Flux.range(0, 5) + .flatMap(it -> Mono.delay(Duration.ofMillis(1)) + .map(context -> this.currentTraceContext.get()) + .doOnNext(spanInOperation::set)) + .blockFirst(); + } + + then(spanInOperation.get()).isEqualTo(context); + } + + @EnableAutoConfiguration + @Configuration + static class Config { + + @Bean + Sampler sampler() { + return Sampler.ALWAYS_SAMPLE; + } + + } + +} diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java index 6c0bf89c6..b809a743e 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java @@ -16,32 +16,59 @@ package org.springframework.cloud.sleuth.instrument.reactor; +import java.util.Objects; +import java.util.function.Function; + import brave.propagation.CurrentTraceContext; import brave.propagation.CurrentTraceContext.Scope; import brave.propagation.TraceContext; +import org.assertj.core.presentation.StandardRepresentation; +import org.awaitility.Awaitility; +import org.junit.After; import org.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.BDDMockito; -import org.mockito.junit.MockitoJUnitRunner; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; import reactor.core.CoreSubscriber; import reactor.core.publisher.BaseSubscriber; +import reactor.core.publisher.Mono; import reactor.util.context.Context; -import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.BDDAssertions.then; +import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.scopePassingSpanOperator; /** * @author Marcin Grzejszczak */ -@RunWith(MockitoJUnitRunner.class) public class ScopePassingSpanSubscriberTests { - CurrentTraceContext currentTraceContext = CurrentTraceContext.Default.create(); + static { + // AssertJ will recognise QueueSubscription implements queue and try to invoke + // iterator. That's not allowed, and will cause an exception + // Fuseable$QueueSubscription.NOT_SUPPORTED_MESSAGE. + // This ensures AssertJ uses normal toString. + StandardRepresentation.registerFormatterForType(ScopePassingSpanSubscriber.class, + Objects::toString); + } + + final CurrentTraceContext currentTraceContext = CurrentTraceContext.Default.create(); TraceContext context = TraceContext.newBuilder().traceId(1).spanId(1).sampled(true) .build(); + TraceContext context2 = TraceContext.newBuilder().traceId(1).spanId(2).sampled(true) + .build(); + + AnnotationConfigApplicationContext springContext = new AnnotationConfigApplicationContext(); + + @After + public void close() { + springContext.close(); + } + @Test public void should_propagate_current_context() { ScopePassingSpanSubscriber subscriber = new ScopePassingSpanSubscriber<>(null, @@ -53,28 +80,96 @@ public class ScopePassingSpanSubscriberTests { @Test public void should_set_empty_context_when_context_is_null() { ScopePassingSpanSubscriber subscriber = new ScopePassingSpanSubscriber<>(null, - null, this.currentTraceContext, null); + Context.empty(), this.currentTraceContext, null); then(subscriber.currentContext().isEmpty()).isTrue(); } @Test public void should_put_current_span_to_context() { - try (Scope ws = this.currentTraceContext.newScope(context)) { - CoreSubscriber subscriber = ReactorSleuth.scopePassingSpanSubscription( - beanFactory(), new BaseSubscriber() { - }); + try (Scope ws = this.currentTraceContext.newScope(context2)) { + CoreSubscriber subscriber = new ScopePassingSpanSubscriber<>( + new BaseSubscriber() { + }, Context.empty(), currentTraceContext, context); then(subscriber.currentContext().get(TraceContext.class)).isEqualTo(context); } - } - private BeanFactory beanFactory() { - BeanFactory beanFactory = BDDMockito.mock(BeanFactory.class); - BDDMockito.given(beanFactory.getBean(CurrentTraceContext.class)) - .willReturn(this.currentTraceContext); - return beanFactory; + @Test + public void should_not_trace_scalar_flows() { + springContext.registerBean(CurrentTraceContext.class, () -> currentTraceContext); + springContext.refresh(); + + Function, ? extends Publisher> transformer = scopePassingSpanOperator( + this.springContext); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Subscriber assertNoSpanSubscriber = new CoreSubscriber() { + @Override + public void onSubscribe(Subscription s) { + s.request(Long.MAX_VALUE); + assertThat(s).isNotInstanceOf(ScopePassingSpanSubscriber.class); + } + + @Override + public void onNext(Object o) { + + } + + @Override + public void onError(Throwable t) { + + } + + @Override + public void onComplete() { + + } + }; + + Subscriber assertSpanSubscriber = new CoreSubscriber() { + @Override + public void onSubscribe(Subscription s) { + s.request(Long.MAX_VALUE); + assertThat(s).isInstanceOf(ScopePassingSpanSubscriber.class); + } + + @Override + public void onNext(Object o) { + + } + + @Override + public void onError(Throwable t) { + + } + + @Override + public void onComplete() { + + } + }; + transformer.apply(Mono.just(1).hide()).subscribe(assertSpanSubscriber); + + transformer.apply(Mono.just(1)).subscribe(assertNoSpanSubscriber); + + transformer.apply(Mono.error(new Exception()).hide()) + .subscribe(assertSpanSubscriber); + + transformer.apply(Mono.error(new Exception())) + .subscribe(assertNoSpanSubscriber); + + transformer.apply(Mono.empty().hide()) + .subscribe(assertSpanSubscriber); + + transformer.apply(Mono.empty()).subscribe(assertNoSpanSubscriber); + + } + + Awaitility.await().untilAsserted(() -> { + then(this.currentTraceContext.get()).isNull(); + }); } } diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java deleted file mode 100644 index 9a8278f6c..000000000 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java +++ /dev/null @@ -1,333 +0,0 @@ -/* - * Copyright 2013-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.sleuth.instrument.reactor; - -import java.time.Duration; -import java.util.concurrent.atomic.AtomicReference; -import java.util.function.Function; - -import brave.Span; -import brave.Tracer; -import brave.sampler.Sampler; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.awaitility.Awaitility; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; -import reactor.core.CoreSubscriber; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.core.scheduler.Schedulers; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.test.context.junit4.SpringRunner; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.BDDAssertions.then; - -@RunWith(SpringRunner.class) -@SpringBootTest(classes = SpanSubscriberTests.Config.class, - webEnvironment = SpringBootTest.WebEnvironment.NONE) -public class SpanSubscriberTests { - - private static final Log log = LogFactory.getLog(SpanSubscriberTests.class); - - @Autowired - Tracer tracer; - - @Autowired - ConfigurableApplicationContext factory; - - @Test - public void should_pass_tracing_info_when_using_reactor() { - Span span = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - Publisher traced = Flux.just(1, 2, 3); - log.info("Hello"); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - Flux.from(traced).map(d -> d + 1).map(d -> d + 1).map((d) -> { - spanInOperation.set(this.tracer.currentSpan()); - return d + 1; - }).map(d -> d + 1).subscribe(System.out::println); - } - finally { - span.finish(); - } - - then(this.tracer.currentSpan()).isNull(); - then(spanInOperation.get().context().spanId()).isEqualTo(span.context().spanId()); - } - - @Test - public void should_support_reactor_fusion_optimization() { - Span span = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - log.info("Hello"); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - Mono.just(1).flatMap(d -> Flux.just(d + 1).collectList().map(p -> p.get(0))) - .map(d -> d + 1).map((d) -> { - spanInOperation.set(this.tracer.currentSpan()); - return d + 1; - }).map(d -> d + 1).subscribe(System.out::println); - } - finally { - span.finish(); - } - - then(this.tracer.currentSpan()).isNull(); - then(spanInOperation.get().context().spanId()).isEqualTo(span.context().spanId()); - } - - @Test - public void should_not_trace_scalar_flows() { - Span span = this.tracer.nextSpan().name("foo").start(); - log.info("Hello"); - - // Disable global hooks for local hook testing - TraceReactorAutoConfigurationAccessorConfiguration.close(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - - Function, ? extends Publisher> transformer = ReactorSleuth - .scopePassingSpanOperator(this.factory); - - Subscriber assertNoSpanSubscriber = new CoreSubscriber() { - @Override - public void onSubscribe(Subscription s) { - s.request(Long.MAX_VALUE); - assertThat(s).isNotInstanceOf(ScopePassingSpanSubscriber.class); - } - - @Override - public void onNext(Object o) { - - } - - @Override - public void onError(Throwable t) { - - } - - @Override - public void onComplete() { - - } - }; - - Subscriber assertSpanSubscriber = new CoreSubscriber() { - @Override - public void onSubscribe(Subscription s) { - s.request(Long.MAX_VALUE); - assertThat(s).isInstanceOf(ScopePassingSpanSubscriber.class); - } - - @Override - public void onNext(Object o) { - - } - - @Override - public void onError(Throwable t) { - - } - - @Override - public void onComplete() { - - } - }; - transformer.apply(Mono.just(1).hide()).subscribe(assertSpanSubscriber); - - transformer.apply(Mono.just(1)).subscribe(assertNoSpanSubscriber); - - transformer.apply(Mono.error(new Exception()).hide()) - .subscribe(assertSpanSubscriber); - - transformer.apply(Mono.error(new Exception())) - .subscribe(assertNoSpanSubscriber); - - transformer.apply(Mono.empty().hide()) - .subscribe(assertSpanSubscriber); - - transformer.apply(Mono.empty()).subscribe(assertNoSpanSubscriber); - - } - finally { - span.finish(); - } - - Awaitility.await().untilAsserted(() -> { - then(this.tracer.currentSpan()).isNull(); - }); - - TraceReactorAutoConfigurationAccessorConfiguration.setup(this.factory); - } - - @Test - public void should_pass_tracing_info_when_using_reactor_async() { - Span span = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - log.info("Hello"); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.1") - .map(d -> d + 1).map(d -> d + 1) - .publishOn(Schedulers.newSingle("secondThread")).log("reactor.2") - .map((d) -> { - spanInOperation.set(this.tracer.currentSpan()); - return d + 1; - }).map(d -> d + 1).blockLast(); - - Awaitility.await().untilAsserted(() -> { - then(spanInOperation.get().context().traceId()) - .isEqualTo(span.context().traceId()); - }); - then(this.tracer.currentSpan()).isEqualTo(span); - } - finally { - span.finish(); - } - - then(this.tracer.currentSpan()).isNull(); - Span foo2 = this.tracer.nextSpan().name("foo").start(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(foo2)) { - Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.") - .map(d -> d + 1).map(d -> d + 1).map((d) -> { - spanInOperation.set(this.tracer.currentSpan()); - return d + 1; - }).map(d -> d + 1).blockLast(); - - then(this.tracer.currentSpan()).isEqualTo(foo2); - // parent cause there's an async span in the meantime - then(spanInOperation.get().context().traceId()) - .isEqualTo(foo2.context().traceId()); - } - finally { - foo2.finish(); - } - - then(this.tracer.currentSpan()).isNull(); - } - - @Test - public void checkSequenceOfOperations() { - Span parentSpan = this.tracer.nextSpan().name("foo").start(); - log.info("Hello"); - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(parentSpan)) { - final Long spanId = Mono.fromCallable(this.tracer::currentSpan) - .map(span -> span.context().spanId()).block(); - then(spanId).isNotNull(); - - final Long secondSpanId = Mono.fromCallable(this.tracer::currentSpan) - .map(span -> span.context().spanId()).block(); - then(secondSpanId).isEqualTo(spanId); // different trace ids here - } - } - - @Test - public void checkTraceIdDuringZipOperation() { - Span initSpan = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - final AtomicReference spanInZipOperation = new AtomicReference<>(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) { - Mono.fromCallable(this.tracer::currentSpan) - .map(span -> span.context().spanId()).doOnNext(spanInOperation::set) - .zipWith(Mono.fromCallable(this.tracer::currentSpan) - .map(span -> span.context().spanId()) - .doOnNext(spanInZipOperation::set)) - .block(); - } - - then(spanInZipOperation).hasValue(initSpan.context().spanId()); // ok here - then(spanInOperation).hasValue(initSpan.context().spanId()); // Expecting - // - // to have value: - // <1L> but did - // not. - } - - // #646 - @Test - public void should_work_for_mono_just_with_flat_map() { - Span initSpan = this.tracer.nextSpan().name("foo").start(); - log.info("Hello"); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) { - Mono.just("value1") - .flatMap(request -> Mono.just("value2").then(Mono.just("foo"))) - .map(a -> "qwe").block(); - } - } - - // #1030 - @Test - public void checkTraceIdFromSubscriberContext() { - Span initSpan = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInSubscriberContext = new AtomicReference<>(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) { - Mono.subscriberContext() - .map(context -> this.tracer.currentSpan().context().spanId()) - .doOnNext(spanInSubscriberContext::set).block(); - } - - then(spanInSubscriberContext).hasValue(initSpan.context().spanId()); // ok here - } - - @Test - public void should_pass_tracing_info_into_inner_publishers() { - Span span = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - Flux.range(0, 5) - .flatMap(it -> Mono.delay(Duration.ofMillis(1)) - .map(context -> this.tracer.currentSpan()) - .doOnNext(spanInOperation::set)) - .blockFirst(); - } - finally { - span.finish(); - } - - then(spanInOperation.get().context().spanId()).isEqualTo(span.context().spanId()); - } - - @EnableAutoConfiguration - @Configuration - static class Config { - - @Bean - Sampler sampler() { - return Sampler.ALWAYS_SAMPLE; - } - - } - -} From 36aed2b88844c374e6e71f26e1f40d965f152b16 Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Tue, 4 Feb 2020 09:07:06 +0800 Subject: [PATCH 3/7] Extracts LazyBean and backports reactor code to use it (#1542) There are numerous places where code defensively guards access to BeanFactory methods. This centralizes the code, first using in reactor as that's the more performance sensitive. This also weaves in feedback from #1541 given by @simonbasle and @robotmrv --- .../sleuth/instrument/reactor/LazyBean.java | 67 ++++++++++++++++++ .../instrument/reactor/ReactorSleuth.java | 68 +++++++++++-------- .../reactor/ScopePassingSpanSubscriber.java | 4 +- .../instrument/reactor/SpanSubscriber.java | 2 +- .../TraceReactorAutoConfiguration.java | 9 +-- .../instrument/reactor/LazyBeanTests.java | 44 ++++++++++++ 6 files changed, 159 insertions(+), 35 deletions(-) create mode 100644 spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBean.java create mode 100644 spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBeanTests.java diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBean.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBean.java new file mode 100644 index 000000000..364d18648 --- /dev/null +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBean.java @@ -0,0 +1,67 @@ +/* + * Copyright 2013-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.instrument.reactor; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.lang.Nullable; + +/** + * Avoids calling the expensive {@link ConfigurableApplicationContext#getBean(Class)} many + * times or throwing an exception. + */ +final class LazyBean { + + // spring-jcl uses commons-logging, so do we. + private static final Log log = LogFactory.getLog(LazyBean.class); + + final ConfigurableApplicationContext springContext; + + final Class requiredType; + + T value; + + LazyBean(ConfigurableApplicationContext springContext, Class requiredType) { + this.springContext = springContext; + this.requiredType = requiredType; + } + + /** + * Attempts to provision from the underlying bean factory, if not already provisioned. + * @return the bean value or null if there was an exception getting it. + */ + @Nullable + T get() { + if (this.value != null) { + return this.value; + } + + try { + this.value = springContext.getBean(requiredType); + } + catch (Exception ex) { + if (log.isDebugEnabled()) { + log.debug("Spring context [" + springContext + "] error getting [" + + requiredType + "].", ex); + } + } + return this.value; + } + +} diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java index 73a212a16..7f095b779 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java @@ -16,8 +16,6 @@ package org.springframework.cloud.sleuth.instrument.reactor; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; import java.util.function.Function; import brave.Tracing; @@ -70,10 +68,14 @@ public abstract class ReactorSleuth { log.trace("Scope passing operator [" + springContext + "]"); } + // keep a reference outside the lambda so that any caching will be visible to + // all publishers + LazyBean lazyCurrentTraceContext = new LazyBean<>( + springContext, CurrentTraceContext.class); + return Operators.liftPublisher((p, sub) -> { - // While supply of scalar types may be deferred, we don't currently scope - // production of values in a trace context. This prevents excessive overhead - // when using constant results such as Flux/Mono #just, #empty, #error + // We don't scope scalar results as they happen in an instant. This prevents + // excessive overhead when using Flux/Mono #just, #empty, #error, etc. if (p instanceof Fuseable.ScalarCallable) { return sub; } @@ -88,13 +90,36 @@ public abstract class ReactorSleuth { return sub; } + Context context = sub.currentContext(); + if (log.isTraceEnabled()) { - log.trace("Spring Context [" + springContext - + "] Creating a scope passing span subscriber with Reactor Context " - + "[" + sub.currentContext() + "] and name [" + name(sub) + "]"); + log.trace("Spring context [" + springContext + "], Reactor context [" + + context + "], name [" + name(sub) + "]"); } - return scopePassingSpanSubscription(springContext, sub); + // Try to get the current trace context bean, lenient when there are problems + CurrentTraceContext currentTraceContext = lazyCurrentTraceContext.get(); + if (currentTraceContext == null) { + if (log.isTraceEnabled()) { + log.trace("Spring Context [" + springContext + + "] did not return a CurrentTraceContext. Reactor Context is [" + + sub.currentContext() + "] and name is [" + name(sub) + "]"); + } + assert false; // should never happen, but don't break. + return sub; + } + + TraceContext parent = traceContext(context, currentTraceContext); + if (parent == null) { + return sub; // no need to scope a null parent + } + + if (log.isTraceEnabled()) { + log.trace("Creating a scope passing span subscriber with Reactor Context " + + "[" + context + "] and name [" + name(sub) + "]"); + } + return new ScopePassingSpanSubscriber<>(sub, context, currentTraceContext, + parent); }); } @@ -102,25 +127,14 @@ public abstract class ReactorSleuth { return Scannable.from(sub).name(); } - private static Map CACHE = new ConcurrentHashMap<>(); - - static CoreSubscriber scopePassingSpanSubscription( - ConfigurableApplicationContext springContext, CoreSubscriber sub) { - CurrentTraceContext currentTraceContext = CACHE.computeIfAbsent(springContext, - springContext1 -> springContext1.getBean(CurrentTraceContext.class)); - Context context = sub.currentContext(); - - TraceContext parent = context.getOrDefault(TraceContext.class, null); - if (parent == null) { - parent = currentTraceContext.get(); - } - if (parent != null) { - return new ScopePassingSpanSubscriber<>(sub, context, currentTraceContext, - parent); - } - else { - return sub; // no need to trace + /** + * Like {@link CurrentTraceContext#get()}, except it first checks the reactor context. + */ + static TraceContext traceContext(Context context, CurrentTraceContext fallback) { + if (context.hasKey(TraceContext.class)) { + return context.get(TraceContext.class); } + return fallback.get(); } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java index aa76125e4..2e24dc278 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java @@ -54,8 +54,7 @@ final class ScopePassingSpanSubscriber implements SpanSubscription, Scanna this.subscriber = subscriber; this.currentTraceContext = currentTraceContext; this.parent = parent; - this.context = ctx != null && parent != null ? ctx.put(TraceContext.class, parent) - : ctx != null ? ctx : Context.empty(); + this.context = parent != null ? ctx.put(TraceContext.class, parent) : ctx; if (log.isTraceEnabled()) { log.trace("Parent span [" + parent + "], context [" + this.context + "]"); } @@ -81,7 +80,6 @@ final class ScopePassingSpanSubscriber implements SpanSubscription, Scanna try (Scope scope = this.currentTraceContext.maybeScope(this.parent)) { this.s.cancel(); } - } @Override diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java index 61ad068fc..b5d0014d9 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java @@ -109,7 +109,7 @@ final class SpanSubscriber extends AtomicBoolean implements SpanSubscription< // no additional cleaning is required cause we operate on scopes if (log.isTraceEnabled()) { log.trace("Request after cleaning. Current span [{}]", - this.currentTraceContext.get()); + this.span.context()); } } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java index f28ddba1c..a61790ef3 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java @@ -45,6 +45,7 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.env.ConfigurableEnvironment; +import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.scopePassingSpanOperator; import static org.springframework.cloud.sleuth.instrument.reactor.TraceReactorAutoConfiguration.TraceReactorConfiguration.SLEUTH_TRACE_REACTOR_KEY; /** @@ -154,14 +155,14 @@ class HooksRefresher implements ApplicationListener log.trace("Decorating onEach operator instrumentation"); } Hooks.onEachOperator(SLEUTH_TRACE_REACTOR_KEY, - ReactorSleuth.scopePassingSpanOperator(this.context)); + scopePassingSpanOperator(this.context)); } else { if (log.isTraceEnabled()) { log.trace("Decorating onLast operator instrumentation"); } Hooks.onLastOperator(SLEUTH_TRACE_REACTOR_KEY, - ReactorSleuth.scopePassingSpanOperator(this.context)); + scopePassingSpanOperator(this.context)); } } @@ -198,14 +199,14 @@ class HookRegisteringBeanDefinitionRegistryPostProcessor log.trace("Decorating onEach operator instrumentation"); } Hooks.onEachOperator(SLEUTH_TRACE_REACTOR_KEY, - ReactorSleuth.scopePassingSpanOperator(springContext)); + scopePassingSpanOperator(this.springContext)); } else { if (log.isTraceEnabled()) { log.trace("Decorating onLast operator instrumentation"); } Hooks.onLastOperator(SLEUTH_TRACE_REACTOR_KEY, - ReactorSleuth.scopePassingSpanOperator(springContext)); + scopePassingSpanOperator(this.springContext)); } Schedulers.setExecutorServiceDecorator( TraceReactorAutoConfiguration.SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY, diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBeanTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBeanTests.java new file mode 100644 index 000000000..3f9694ba6 --- /dev/null +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBeanTests.java @@ -0,0 +1,44 @@ +/* + * Copyright 2013-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.instrument.reactor; + +import brave.propagation.CurrentTraceContext; +import org.junit.Test; + +import org.springframework.context.ConfigurableApplicationContext; + +import static org.assertj.core.api.BDDAssertions.then; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class LazyBeanTests { + + @Test + public void should_return_null_when_exception_thrown_upon_bean_retrieval() { + ConfigurableApplicationContext springContext = mock( + ConfigurableApplicationContext.class); + + when(springContext.getBean(CurrentTraceContext.class)) + .thenThrow(new IllegalStateException()); + + LazyBean provider = new LazyBean<>(springContext, + CurrentTraceContext.class); + + then(provider.get()).isNull(); + } + +} From af04d9d43167119edeba692b853ec0ae6edaceff Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Tue, 4 Feb 2020 10:16:55 +0800 Subject: [PATCH 4/7] Corrects HookRegisteringBeanDefinitionRegistryPostProcessor.setupHooks (#1546) Probably due to code drift, `HookRegisteringBeanDefinitionRegistryPostProcessor.setupHooks` was used as an object method, which led to a confusing no-op chain in tests. This ensures it is used consistently as a static method, and also corrects a bean accidentally marked static for testing. --- .../reactor/TraceReactorAutoConfiguration.java | 9 ++++----- ...aceReactorAutoConfigurationAccessorConfiguration.java | 4 +--- ...aceReactorAutoConfigurationAccessorConfiguration.java | 4 +--- 3 files changed, 6 insertions(+), 11 deletions(-) diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java index a61790ef3..1f913ec08 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java @@ -101,9 +101,8 @@ public class TraceReactorAutoConfiguration { } @Bean - // for tests @ConditionalOnMissingBean - static HookRegisteringBeanDefinitionRegistryPostProcessor traceHookRegisteringBeanDefinitionRegistryPostProcessor( + HookRegisteringBeanDefinitionRegistryPostProcessor traceHookRegisteringBeanDefinitionRegistryPostProcessor( ConfigurableApplicationContext context) { if (log.isTraceEnabled()) { log.trace( @@ -190,7 +189,7 @@ class HookRegisteringBeanDefinitionRegistryPostProcessor setupHooks(this.springContext); } - void setupHooks(ConfigurableApplicationContext springContext) { + static void setupHooks(ConfigurableApplicationContext springContext) { ConfigurableEnvironment environment = springContext.getEnvironment(); boolean decorateOnEach = environment.getProperty( "spring.sleuth.reactor.decorate-on-each", Boolean.class, true); @@ -199,14 +198,14 @@ class HookRegisteringBeanDefinitionRegistryPostProcessor log.trace("Decorating onEach operator instrumentation"); } Hooks.onEachOperator(SLEUTH_TRACE_REACTOR_KEY, - scopePassingSpanOperator(this.springContext)); + scopePassingSpanOperator(springContext)); } else { if (log.isTraceEnabled()) { log.trace("Decorating onLast operator instrumentation"); } Hooks.onLastOperator(SLEUTH_TRACE_REACTOR_KEY, - scopePassingSpanOperator(this.springContext)); + scopePassingSpanOperator(springContext)); } Schedulers.setExecutorServiceDecorator( TraceReactorAutoConfiguration.SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY, diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java index 7cbf973f6..763574221 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java @@ -51,9 +51,7 @@ public final class TraceReactorAutoConfigurationAccessorConfiguration { if (log.isTraceEnabled()) { log.trace("Setting up hooks"); } - TraceReactorAutoConfiguration.TraceReactorConfiguration - .traceHookRegisteringBeanDefinitionRegistryPostProcessor(context) - .setupHooks(context); + HookRegisteringBeanDefinitionRegistryPostProcessor.setupHooks(context); } } diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java index 7cbf973f6..763574221 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java @@ -51,9 +51,7 @@ public final class TraceReactorAutoConfigurationAccessorConfiguration { if (log.isTraceEnabled()) { log.trace("Setting up hooks"); } - TraceReactorAutoConfiguration.TraceReactorConfiguration - .traceHookRegisteringBeanDefinitionRegistryPostProcessor(context) - .setupHooks(context); + HookRegisteringBeanDefinitionRegistryPostProcessor.setupHooks(context); } } From fc47d61883964831d79e3b47e863b6bc0376c1d9 Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Tue, 4 Feb 2020 13:23:54 +0800 Subject: [PATCH 5/7] bumps to latest brave --- benchmarks/pom.xml | 2 +- pom.xml | 2 +- spring-cloud-sleuth-dependencies/pom.xml | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/benchmarks/pom.xml b/benchmarks/pom.xml index 535a21d6f..3c1f204f4 100644 --- a/benchmarks/pom.xml +++ b/benchmarks/pom.xml @@ -33,7 +33,7 @@ 1.8 1.8 2.3.0.BUILD-SNAPSHOT - 5.9.2 + 5.9.3 3.14.6 diff --git a/pom.xml b/pom.xml index 6ba2f17b2..38a06d5d7 100644 --- a/pom.xml +++ b/pom.xml @@ -257,7 +257,7 @@ Horsham.SR1 2.2.2.BUILD-SNAPSHOT 2.2.2.BUILD-SNAPSHOT - 5.9.2 + 5.9.3 2.1.7.RELEASE 2.2.2.BUILD-SNAPSHOT false diff --git a/spring-cloud-sleuth-dependencies/pom.xml b/spring-cloud-sleuth-dependencies/pom.xml index 56225537b..acb206410 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.9.2 + 5.9.3 0.35.1 3.4.1 From ac43895b75760d07bc06c13876eeaabda4a08778 Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Wed, 5 Feb 2020 13:21:37 +0100 Subject: [PATCH 6/7] Ignores stream spans for channels with name starting with channel* ; fixes gh-1549 --- .../instrument/messaging/SleuthMessagingProperties.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java index 740aa1452..08806b453 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java @@ -57,9 +57,9 @@ public class SleuthMessagingProperties { /** * An array of patterns against which channel names will be matched. * @see org.springframework.integration.config.GlobalChannelInterceptor#patterns() - * Defaults to any channel name not matching the Hystrix Stream channel name. + * Defaults to any channel name not matching the Hystrix Stream and functional Stream channel names. */ - private String[] patterns = new String[] { "!hystrixStreamOutput*", "*" }; + private String[] patterns = new String[] { "!hystrixStreamOutput*", "*", "!channel*"}; /** * Enable Spring Integration sleuth instrumentation. From 415c809fd095449dfb5207c64b3dce71c4b20ff4 Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Wed, 5 Feb 2020 13:36:15 +0100 Subject: [PATCH 7/7] Added better remote name resolution for messaging; fixes gh-1481 --- .../messaging/TracingChannelInterceptor.java | 103 ++++----- .../TracingChannelInterceptorTest.java | 202 +++++++++--------- 2 files changed, 152 insertions(+), 153 deletions(-) diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java index 07d3efd1c..bef74c433 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2019 the original author or authors. + * Copyright 2013-2020 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. @@ -59,8 +59,7 @@ import org.springframework.util.ClassUtils; * * @author Marcin Grzejszczak */ -public final class TracingChannelInterceptor extends ChannelInterceptorAdapter - implements ExecutorChannelInterceptor { +public final class TracingChannelInterceptor extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor { /** * Name of the class in Spring Cloud Stream that is a direct channel. @@ -109,25 +108,22 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter @Autowired TracingChannelInterceptor(Tracing tracing) { - this(tracing, MessageHeaderPropagation.INSTANCE, - MessageHeaderPropagation.INSTANCE); + this(tracing, MessageHeaderPropagation.INSTANCE, MessageHeaderPropagation.INSTANCE); } - TracingChannelInterceptor(Tracing tracing, - Propagation.Setter setter, + TracingChannelInterceptor(Tracing tracing, Propagation.Setter setter, Propagation.Getter getter) { this.tracing = tracing; this.tracer = tracing.tracer(); this.threadLocalSpan = ThreadLocalSpan.create(this.tracer); this.injector = tracing.propagation().injector(setter); this.extractor = tracing.propagation().extractor(getter); - this.integrationObjectSupportPresent = ClassUtils.isPresent( - "org.springframework.integration.context.IntegrationObjectSupport", null); - this.hasDirectChannelClass = ClassUtils - .isPresent("org.springframework.integration.channel.DirectChannel", null); - this.directWithAttributesChannelClass = ClassUtils - .isPresent(STREAM_DIRECT_CHANNEL, null) - ? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null; + this.integrationObjectSupportPresent = ClassUtils + .isPresent("org.springframework.integration.context.IntegrationObjectSupport", null); + this.hasDirectChannelClass = ClassUtils.isPresent("org.springframework.integration.channel.DirectChannel", + null); + this.directWithAttributesChannelClass = ClassUtils.isPresent(STREAM_DIRECT_CHANNEL, null) + ? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null; } public static TracingChannelInterceptor create(Tracing tracing) { @@ -170,12 +166,11 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter MessageHeaderAccessor headers = mutableHeaderAccessor(retrievedMessage); TraceContextOrSamplingFlags extracted = this.extractor.extract(headers); Span span = this.threadLocalSpan.next(extracted); - MessageHeaderPropagation.removeAnyTraceHeaders(headers, - this.tracing.propagation().keys()); + MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys()); this.injector.inject(span.context(), headers); if (!span.isNoop()) { span.kind(Span.Kind.PRODUCER).name("send").start(); - span.remoteServiceName(REMOTE_SERVICE_NAME); + span.remoteServiceName(toRemoteServiceName(headers)); addTags(message, span, channel); } if (log.isDebugEnabled()) { @@ -188,24 +183,31 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter return outputMessage; } - private Message outputMessage(Message originalMessage, - Message retrievedMessage, MessageHeaderAccessor additionalHeaders) { - MessageHeaderAccessor headers = MessageHeaderAccessor - .getMutableAccessor(originalMessage); + private String toRemoteServiceName(MessageHeaderAccessor headers) { + for (String key : headers.getMessageHeaders().keySet()) { + if (key.startsWith("kafka_")) { + return "kafka"; + } + else if (key.startsWith("amqp_")) { + return "rabbitmq"; + } + } + return REMOTE_SERVICE_NAME; + } + + private Message outputMessage(Message originalMessage, Message retrievedMessage, + MessageHeaderAccessor additionalHeaders) { + MessageHeaderAccessor headers = MessageHeaderAccessor.getMutableAccessor(originalMessage); if (originalMessage instanceof ErrorMessage) { ErrorMessage errorMessage = (ErrorMessage) originalMessage; - headers.copyHeaders(MessageHeaderPropagation.propagationHeaders( - additionalHeaders.getMessageHeaders(), + headers.copyHeaders(MessageHeaderPropagation.propagationHeaders(additionalHeaders.getMessageHeaders(), this.tracing.propagation().keys())); - return new ErrorMessage(errorMessage.getPayload(), - isWebSockets(headers) ? headers.getMessageHeaders() - : new MessageHeaders(headers.getMessageHeaders()), - errorMessage.getOriginalMessage()); + return new ErrorMessage(errorMessage.getPayload(), isWebSockets(headers) ? headers.getMessageHeaders() + : new MessageHeaders(headers.getMessageHeaders()), errorMessage.getOriginalMessage()); } headers.copyHeaders(additionalHeaders.getMessageHeaders()); return new GenericMessage<>(retrievedMessage.getPayload(), - isWebSockets(headers) ? headers.getMessageHeaders() - : new MessageHeaders(headers.getMessageHeaders())); + isWebSockets(headers) ? headers.getMessageHeaders() : new MessageHeaders(headers.getMessageHeaders())); } private boolean isWebSockets(MessageHeaderAccessor headerAccessor) { @@ -215,8 +217,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter private boolean isDirectChannel(MessageChannel channel) { Class targetClass = AopUtils.getTargetClass(channel); - boolean directChannel = this.hasDirectChannelClass - && DirectChannel.class.isAssignableFrom(targetClass); + boolean directChannel = this.hasDirectChannelClass && DirectChannel.class.isAssignableFrom(targetClass); if (!directChannel) { return false; } @@ -231,8 +232,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter } @Override - public void afterSendCompletion(Message message, MessageChannel channel, - boolean sent, Exception ex) { + public void afterSendCompletion(Message message, MessageChannel channel, boolean sent, Exception ex) { if (emptyMessage(message)) { return; } @@ -240,8 +240,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter afterMessageHandled(message, channel, null, ex); } if (log.isDebugEnabled()) { - log.debug("Will finish the current span after completion " - + this.tracer.currentSpan()); + log.debug("Will finish the current span after completion " + this.tracer.currentSpan()); } finishSpan(ex); } @@ -258,12 +257,11 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter MessageHeaderAccessor headers = mutableHeaderAccessor(message); TraceContextOrSamplingFlags extracted = this.extractor.extract(headers); Span span = this.threadLocalSpan.next(extracted); - MessageHeaderPropagation.removeAnyTraceHeaders(headers, - this.tracing.propagation().keys()); + MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys()); this.injector.inject(span.context(), headers); if (!span.isNoop()) { span.kind(Span.Kind.CONSUMER).name("receive").start(); - span.remoteServiceName(REMOTE_SERVICE_NAME); + span.remoteServiceName(toRemoteServiceName(headers)); addTags(message, span, channel); } if (log.isDebugEnabled()) { @@ -272,21 +270,19 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter headers.setImmutable(); if (message instanceof ErrorMessage) { ErrorMessage errorMessage = (ErrorMessage) message; - return new ErrorMessage(errorMessage.getPayload(), - headers.getMessageHeaders(), errorMessage.getOriginalMessage()); + return new ErrorMessage(errorMessage.getPayload(), headers.getMessageHeaders(), + errorMessage.getOriginalMessage()); } return new GenericMessage<>(message.getPayload(), headers.getMessageHeaders()); } @Override - public void afterReceiveCompletion(Message message, MessageChannel channel, - Exception ex) { + public void afterReceiveCompletion(Message message, MessageChannel channel, Exception ex) { if (emptyMessage(message)) { return; } if (log.isDebugEnabled()) { - log.debug("Will finish the current span after receive completion " - + this.tracer.currentSpan()); + log.debug("Will finish the current span after receive completion " + this.tracer.currentSpan()); } finishSpan(ex); } @@ -296,8 +292,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter * context. It then creates a span for the handler, placing it in scope. */ @Override - public Message beforeHandle(Message message, MessageChannel channel, - MessageHandler handler) { + public Message beforeHandle(Message message, MessageChannel channel, MessageHandler handler) { if (emptyMessage(message)) { return message; } @@ -312,34 +307,28 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter consumerSpan.finish(); } // create and scope a span for the message processor - this.threadLocalSpan - .next(TraceContextOrSamplingFlags.create(consumerSpan.context())) - .name("handle").start(); + this.threadLocalSpan.next(TraceContextOrSamplingFlags.create(consumerSpan.context())).name("handle").start(); // remove any trace headers, but don't re-inject as we are synchronously // processing the // message and can rely on scoping to access this span later. - MessageHeaderPropagation.removeAnyTraceHeaders(headers, - this.tracing.propagation().keys()); + MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys()); if (log.isDebugEnabled()) { log.debug("Created a new span in before handle" + consumerSpan); } if (message instanceof ErrorMessage) { - return new ErrorMessage((Throwable) message.getPayload(), - headers.getMessageHeaders()); + return new ErrorMessage((Throwable) message.getPayload(), headers.getMessageHeaders()); } headers.setImmutable(); return new GenericMessage<>(message.getPayload(), headers.getMessageHeaders()); } @Override - public void afterMessageHandled(Message message, MessageChannel channel, - MessageHandler handler, Exception ex) { + public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, Exception ex) { if (emptyMessage(message)) { return; } if (log.isDebugEnabled()) { - log.debug("Will finish the current span after message handled " - + this.tracer.currentSpan()); + log.debug("Will finish the current span after message handled " + this.tracer.currentSpan()); } finishSpan(ex); } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java index 1d8423f6c..10b5c1c91 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java @@ -29,8 +29,10 @@ import org.junit.After; import org.junit.Test; import zipkin2.Span; +import org.springframework.amqp.support.AmqpHeaders; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.kafka.support.KafkaHeaders; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -51,10 +53,11 @@ public class TracingChannelInterceptorTest { List spans = new ArrayList<>(); - ChannelInterceptor interceptor = TracingChannelInterceptor.create(Tracing.newBuilder() - .currentTraceContext(ThreadLocalCurrentTraceContext.newBuilder() - .addScopeDecorator(StrictScopeDecorator.create()).build()) - .spanReporter(this.spans::add).build()); + ChannelInterceptor interceptor = TracingChannelInterceptor + .create(Tracing.newBuilder() + .currentTraceContext(ThreadLocalCurrentTraceContext.newBuilder() + .addScopeDecorator(StrictScopeDecorator.create()).build()) + .spanReporter(this.spans::add).build()); QueueChannel channel = new QueueChannel(); @@ -83,10 +86,9 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", - "X-B3-SpanId", "X-B3-Sampled", "nativeHeaders"); - assertThat(this.spans).hasSize(1).flatExtracting(Span::kind) - .containsExactly(Span.Kind.PRODUCER); + assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", + "nativeHeaders"); + assertThat(this.spans).hasSize(1).flatExtracting(Span::kind).containsExactly(Span.Kind.PRODUCER); } @Test @@ -96,10 +98,9 @@ public class TracingChannelInterceptorTest { this.directChannel.send(MessageBuilder.withPayload("foo").build()); assertThat(this.message).isNotNull(); - assertThat(this.message.getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", - "X-B3-Sampled", "nativeHeaders"); - assertThat(this.spans).flatExtracting(Span::kind).contains(Span.Kind.CONSUMER, - Span.Kind.PRODUCER); + assertThat(this.message.getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", + "nativeHeaders"); + assertThat(this.spans).flatExtracting(Span::kind).contains(Span.Kind.CONSUMER, Span.Kind.PRODUCER); } @Test @@ -108,9 +109,8 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); - assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)) - .containsOnlyKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", - "spanTraceId", "spanId", "spanSampled"); + assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsOnlyKeys("X-B3-TraceId", + "X-B3-SpanId", "X-B3-Sampled", "spanTraceId", "spanId", "spanSampled"); } /** @@ -122,13 +122,11 @@ public class TracingChannelInterceptorTest { public void producerConsidersOldSpanIds() { this.channel.addInterceptor(producerSideOnly(this.interceptor)); - this.channel.send(MessageBuilder.withPayload("foo") - .setHeader("X-B3-TraceId", "000000000000000a") - .setHeader("X-B3-ParentSpanId", "000000000000000a") - .setHeader("X-B3-SpanId", "000000000000000b").build()); + this.channel.send(MessageBuilder.withPayload("foo").setHeader("X-B3-TraceId", "000000000000000a") + .setHeader("X-B3-ParentSpanId", "000000000000000a").setHeader("X-B3-SpanId", "000000000000000b") + .build()); - assertThat(this.channel.receive().getHeaders()).containsEntry("X-B3-ParentSpanId", - "000000000000000b"); + assertThat(this.channel.receive().getHeaders()).containsEntry("X-B3-ParentSpanId", "000000000000000b"); } @Test @@ -142,12 +140,10 @@ public class TracingChannelInterceptorTest { accessor.setNativeHeader("X-B3-ParentSpanId", "000000000000000a"); accessor.setNativeHeader("X-B3-SpanId", "000000000000000b"); - this.channel.send(MessageBuilder.withPayload("foo") - .copyHeaders(accessor.toMessageHeaders()).build()); + this.channel.send(MessageBuilder.withPayload("foo").copyHeaders(accessor.toMessageHeaders()).build()); - assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)) - .containsEntry("X-B3-ParentSpanId", - Collections.singletonList("000000000000000b")); + assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsEntry("X-B3-ParentSpanId", + Collections.singletonList("000000000000000b")); } /** @@ -160,10 +156,9 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", - "X-B3-SpanId", "X-B3-Sampled", "nativeHeaders"); - assertThat(this.spans).hasSize(1).flatExtracting(Span::kind) - .containsExactly(Span.Kind.CONSUMER); + assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", + "nativeHeaders"); + assertThat(this.spans).hasSize(1).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER); } @Test @@ -172,9 +167,8 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); - assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)) - .containsOnlyKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", - "spanTraceId", "spanId", "spanSampled"); + assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsOnlyKeys("X-B3-TraceId", + "X-B3-SpanId", "X-B3-Sampled", "spanTraceId", "spanId", "spanSampled"); } @Test @@ -186,10 +180,9 @@ public class TracingChannelInterceptorTest { channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", - "X-B3-SpanId", "X-B3-Sampled", "nativeHeaders"); - assertThat(this.spans).flatExtracting(Span::kind) - .containsExactly(Span.Kind.CONSUMER, null); + assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", + "nativeHeaders"); + assertThat(this.spans).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER, null); } /** @@ -206,8 +199,7 @@ public class TracingChannelInterceptorTest { channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", - "X-B3-SpanId", "X-B3-Sampled"); + assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled"); } @Test @@ -219,8 +211,8 @@ public class TracingChannelInterceptorTest { channel.send(MessageBuilder.withPayload("foo").build()); - assertThat((Map) messages.get(0).getHeaders().get(NATIVE_HEADERS)) - .doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled"); + assertThat((Map) messages.get(0).getHeaders().get(NATIVE_HEADERS)).doesNotContainKeys("X-B3-TraceId", + "X-B3-SpanId", "X-B3-Sampled"); } @Test @@ -230,8 +222,8 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); this.channel.receive(); - assertThat(this.spans).flatExtracting(Span::kind) - .containsExactlyInAnyOrder(Span.Kind.CONSUMER, Span.Kind.PRODUCER); + assertThat(this.spans).flatExtracting(Span::kind).containsExactlyInAnyOrder(Span.Kind.CONSUMER, + Span.Kind.PRODUCER); } @Test @@ -243,8 +235,7 @@ public class TracingChannelInterceptorTest { channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(this.spans).flatExtracting(Span::kind) - .containsExactly(Span.Kind.CONSUMER, null, Span.Kind.PRODUCER); + assertThat(this.spans).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER, null, Span.Kind.PRODUCER); } @Test @@ -255,54 +246,41 @@ public class TracingChannelInterceptorTest { Map errorChannelHeaders = new HashMap<>(); errorChannelHeaders.put(MessageHeaders.REPLY_CHANNEL, errorsReplyChannel); errorChannelHeaders.put(MessageHeaders.ERROR_CHANNEL, errorsReplyChannel); - this.channel - .send(new ErrorMessage( - new MessagingException(MessageBuilder.withPayload("hi") - .setHeader(TraceMessageHeaders.TRACE_ID_NAME, - "000000000000000a") - .setHeader(TraceMessageHeaders.SPAN_ID_NAME, - "000000000000000a") - .setReplyChannel(deadReplyChannel) - .setErrorChannel(deadReplyChannel).build()), - errorChannelHeaders)); + this.channel.send(new ErrorMessage( + new MessagingException(MessageBuilder.withPayload("hi") + .setHeader(TraceMessageHeaders.TRACE_ID_NAME, "000000000000000a") + .setHeader(TraceMessageHeaders.SPAN_ID_NAME, "000000000000000a") + .setReplyChannel(deadReplyChannel).setErrorChannel(deadReplyChannel).build()), + errorChannelHeaders)); this.message = this.channel.receive(); assertThat(this.message).isNotNull(); - String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, - String.class); + String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, String.class); assertThat(spanId).isNotNull(); - String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, - String.class); + String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, String.class); assertThat(traceId).isEqualTo("000000000000000a"); assertThat(spanId).isNotEqualTo("000000000000000a"); assertThat(this.spans).hasSize(2); - assertThat(this.message.getHeaders().getReplyChannel()) - .isSameAs(errorsReplyChannel); - assertThat(this.message.getHeaders().getErrorChannel()) - .isSameAs(errorsReplyChannel); + assertThat(this.message.getHeaders().getReplyChannel()).isSameAs(errorsReplyChannel); + assertThat(this.message.getHeaders().getErrorChannel()).isSameAs(errorsReplyChannel); } @Test public void errorMessageOriginalMessageRetained() { this.channel.addInterceptor(this.interceptor); - Message originalMessage = MessageBuilder.withPayload("Hello") - .setHeader("header", "value").build(); - Message failedMessage = MessageBuilder.fromMessage(originalMessage) - .removeHeader("header").build(); - this.channel.send(new ErrorMessage(new MessagingException(failedMessage), - originalMessage.getHeaders(), originalMessage)); + Message originalMessage = MessageBuilder.withPayload("Hello").setHeader("header", "value").build(); + Message failedMessage = MessageBuilder.fromMessage(originalMessage).removeHeader("header").build(); + this.channel.send( + new ErrorMessage(new MessagingException(failedMessage), originalMessage.getHeaders(), originalMessage)); this.message = this.channel.receive(); assertThat(this.message).isNotNull(); - assertThat(this.message).isInstanceOfSatisfying(ErrorMessage.class, - errorMessage -> { - assertThat(errorMessage.getOriginalMessage()) - .isSameAs(originalMessage); - assertThat(errorMessage.getHeaders().get("header")) - .isEqualTo("value"); - }); + assertThat(this.message).isInstanceOfSatisfying(ErrorMessage.class, errorMessage -> { + assertThat(errorMessage.getOriginalMessage()).isSameAs(originalMessage); + assertThat(errorMessage.getHeaders().get("header")).isEqualTo("value"); + }); } @Test @@ -311,22 +289,60 @@ public class TracingChannelInterceptorTest { Map errorChannelHeaders = new HashMap<>(); errorChannelHeaders.put(TraceMessageHeaders.TRACE_ID_NAME, "000000000000000a"); errorChannelHeaders.put(TraceMessageHeaders.SPAN_ID_NAME, "000000000000000a"); - this.channel.send(new ErrorMessage(new MessagingException("exception"), - errorChannelHeaders)); + this.channel.send(new ErrorMessage(new MessagingException("exception"), errorChannelHeaders)); this.message = this.channel.receive(); assertThat(this.message).isNotNull(); - String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, - String.class); + String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, String.class); assertThat(spanId).isNotNull(); - String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, - String.class); + String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, String.class); assertThat(traceId).isEqualTo("000000000000000a"); assertThat(spanId).isNotEqualTo("000000000000000a"); assertThat(this.spans).hasSize(2); } + @Test + public void should_store_kafka_as_remote_service_name_when_kafka_header_is_present() { + ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(); + channel.addInterceptor(this.interceptor); + List> messages = new ArrayList<>(); + channel.subscribe(messages::add); + + Map headers = new HashMap<>(); + headers.put(KafkaHeaders.MESSAGE_KEY, "hello"); + channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers))); + + assertThat(this.spans).flatExtracting(Span::remoteServiceName).contains("kafka"); + } + + @Test + public void should_store_rabbitmq_as_remote_service_name_when_rabbit_header_is_present() { + ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(); + channel.addInterceptor(this.interceptor); + List> messages = new ArrayList<>(); + channel.subscribe(messages::add); + + Map headers = new HashMap<>(); + headers.put(AmqpHeaders.RECEIVED_ROUTING_KEY, "hello"); + channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers))); + + assertThat(this.spans).flatExtracting(Span::remoteServiceName).contains("rabbitmq"); + } + + @Test + public void should_store_broker_as_remote_service_name_when_no_special_headers_were_found() { + ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(); + channel.addInterceptor(this.interceptor); + List> messages = new ArrayList<>(); + channel.subscribe(messages::add); + + Map headers = new HashMap<>(); + channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers))); + + assertThat(this.spans).flatExtracting(Span::remoteServiceName).containsOnly("broker", null); + } + ChannelInterceptor producerSideOnly(ChannelInterceptor delegate) { return new ChannelInterceptorAdapter() { @Override @@ -335,8 +351,7 @@ public class TracingChannelInterceptorTest { } @Override - public void afterSendCompletion(Message message, MessageChannel channel, - boolean sent, Exception ex) { + public void afterSendCompletion(Message message, MessageChannel channel, boolean sent, Exception ex) { delegate.afterSendCompletion(message, channel, sent, ex); } }; @@ -350,29 +365,24 @@ public class TracingChannelInterceptorTest { } @Override - public void afterReceiveCompletion(Message message, MessageChannel channel, - Exception ex) { + public void afterReceiveCompletion(Message message, MessageChannel channel, Exception ex) { delegate.afterReceiveCompletion(message, channel, ex); } }; } ExecutorChannelInterceptor executorSideOnly(ChannelInterceptor delegate) { - class ExecutorSideOnly extends ChannelInterceptorAdapter - implements ExecutorChannelInterceptor { + class ExecutorSideOnly extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor { @Override - public Message beforeHandle(Message message, MessageChannel channel, - MessageHandler handler) { - return ((ExecutorChannelInterceptor) delegate).beforeHandle(message, - channel, handler); + public Message beforeHandle(Message message, MessageChannel channel, MessageHandler handler) { + return ((ExecutorChannelInterceptor) delegate).beforeHandle(message, channel, handler); } @Override - public void afterMessageHandled(Message message, MessageChannel channel, - MessageHandler handler, Exception ex) { - ((ExecutorChannelInterceptor) delegate).afterMessageHandled(message, - channel, handler, ex); + public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, + Exception ex) { + ((ExecutorChannelInterceptor) delegate).afterMessageHandled(message, channel, handler, ex); } }