SpringCloud/spring-cloud-sleuth
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Uses ConfigurableApplicationContext consistently in Reactor (#1545)

Browse Source 
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.
This commit is contained in:
Adrian Cole
2020-02-03 19:26:38 +08:00
committed by GitHub
parent 4acbaf38f3
commit 241f536aef
12 changed files with 384 additions and 625 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java
View File

@@ -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 <T> 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<T> extends AtomicBoolean implements SpanSubscription<T> {
private final Supplier<SpanSubscription<T>> supplier;
LazySpanSubscriber(Supplier<SpanSubscription<T>> 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();
}
}

47
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java
View File

@@ -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 <T> 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 <T> Function<? super Publisher<T>, ? extends Publisher<T>> 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<BeanFactory, CurrentTraceContext> CACHE = new ConcurrentHashMap<>();
private static Map<ConfigurableApplicationContext, CurrentTraceContext> CACHE = new ConcurrentHashMap<>();
static <T> CoreSubscriber<? super T> scopePassingSpanSubscription(
BeanFactory beanFactory, CoreSubscriber<? super T> sub) {
CurrentTraceContext currentTraceContext = CACHE.computeIfAbsent(beanFactory,
beanFactory1 -> beanFactory1.getBean(CurrentTraceContext.class));
ConfigurableApplicationContext springContext, CoreSubscriber<? super T> sub) {
CurrentTraceContext currentTraceContext = CACHE.computeIfAbsent(springContext,
springContext1 -> springContext1.getBean(CurrentTraceContext.class));
Context context = sub.currentContext();
TraceContext parent = context.getOrDefault(TraceContext.class, null);

6
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java
View File

@@ -120,4 +120,10 @@ final class ScopePassingSpanSubscriber<T> implements SpanSubscription<T>, Scanna
}
}
@Override
public String toString() {
return "ScopePassingSpanSubscriber{" + "subscriber=" + this.subscriber
+ ", parent=" + this.parent + "}";
}
}

91
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java
View File

@@ -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 <T> type of returned subscription
* @author Marcin Grzejszczak
*/
final class SpanSubscriptionProvider<T> implements Supplier<SpanSubscription<T>> {
private static final Log log = LogFactory.getLog(SpanSubscriptionProvider.class);
final BeanFactory beanFactory;
final CoreSubscriber<? super T> subscriber;
final Context context;
private volatile CurrentTraceContext currentTraceContext;
SpanSubscriptionProvider(BeanFactory beanFactory,
CoreSubscriber<? super T> 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<T> get() {
return newCoreSubscriber(currentTraceContext());
}
SpanSubscription<T> 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;
}
}

30
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java
View File

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

5
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java
View File

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

33
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java
View File

@@ -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<? super Publisher<DataBuffer>, ? extends Publisher<DataBuffer>> scopePassingTransformer;
@@ -146,14 +147,14 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction {
TraceContext.Injector<ClientRequest.Builder> 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<brave.http.HttpClientRequest, brave.http.HttpClientResponse> 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<ClientRequest.Builder> 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;

47
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java
View File

@@ -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<String> 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<String> provider = new SpanSubscriptionProvider<>(
beanFactory, subscriber);
SpanSubscription<String> spanSubscription = provider.get();
BDDAssertions.then(spanSubscription).isNotNull();
}
}

8
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java
View File

@@ -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,

204
tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberSpringBootTests.java Normal file
View File

@@ -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<TraceContext> spanInOperation = new AtomicReference<>();
Publisher<Integer> 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<TraceContext> 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<TraceContext> 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<TraceContext> 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<TraceContext> spanInOperation = new AtomicReference<>();
final AtomicReference<TraceContext> 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<TraceContext> 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<TraceContext> 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;
}
}
}

129
tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java
View File

@@ -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<Object>() {
});
try (Scope ws = this.currentTraceContext.newScope(context2)) {
CoreSubscriber<?> subscriber = new ScopePassingSpanSubscriber<>(
new BaseSubscriber<Object>() {
}, 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<? super Publisher<Integer>, ? extends Publisher<Integer>> transformer = scopePassingSpanOperator(
this.springContext);
try (Scope ws = this.currentTraceContext.newScope(context)) {
Subscriber<Object> assertNoSpanSubscriber = new CoreSubscriber<Object>() {
@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<Object> assertSpanSubscriber = new CoreSubscriber<Object>() {
@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.<Integer>error(new Exception()).hide())
.subscribe(assertSpanSubscriber);
transformer.apply(Mono.error(new Exception()))
.subscribe(assertNoSpanSubscriber);
transformer.apply(Mono.<Integer>empty().hide())
.subscribe(assertSpanSubscriber);
transformer.apply(Mono.empty()).subscribe(assertNoSpanSubscriber);
}
Awaitility.await().untilAsserted(() -> {
then(this.currentTraceContext.get()).isNull();
});
}
}

333
tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java
View File

@@ -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<Span> spanInOperation = new AtomicReference<>();
Publisher<Integer> 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<Span> 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<? super Publisher<Integer>, ? extends Publisher<Integer>> transformer = ReactorSleuth
.scopePassingSpanOperator(this.factory);
Subscriber<Object> assertNoSpanSubscriber = new CoreSubscriber<Object>() {
@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<Object> assertSpanSubscriber = new CoreSubscriber<Object>() {
@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.<Integer>error(new Exception()).hide())
.subscribe(assertSpanSubscriber);
transformer.apply(Mono.error(new Exception()))
.subscribe(assertNoSpanSubscriber);
transformer.apply(Mono.<Integer>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<Span> 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<Long> spanInOperation = new AtomicReference<>();
final AtomicReference<Long> 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
// <AtomicReference[null]>
// 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<Long> 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<Span> 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;
}
}
}