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

Merge branch '2.2.x'

Browse Source
This commit is contained in:
Marcin Grzejszczak
2020-02-05 13:36:38 +01:00
parent a6d7bd3784 415c809fd0
commit 2b427e1240
19 changed files with 664 additions and 788 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java
View File

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

103
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java
View File

@@ -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<MessageHeaderAccessor, String> setter,
TracingChannelInterceptor(Tracing tracing, Propagation.Setter<MessageHeaderAccessor, String> setter,
Propagation.Getter<MessageHeaderAccessor, String> 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);
}

67
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBean.java Normal file
View File

@@ -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<T> {
// spring-jcl uses commons-logging, so do we.
private static final Log log = LogFactory.getLog(LazyBean.class);
final ConfigurableApplicationContext springContext;
final Class<T> requiredType;
T value;
LazyBean(ConfigurableApplicationContext springContext, Class<T> 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;
}
}

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

@@ -1,75 +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
*/
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();
}
}

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

@@ -16,9 +16,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;
@@ -33,7 +30,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;
/**
@@ -56,75 +52,89 @@ 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 + "]");
}
// Adapt if lazy bean factory
BooleanSupplier isActive = beanFactory instanceof ConfigurableApplicationContext
? ((ConfigurableApplicationContext) beanFactory)::isActive : () -> true;
// keep a reference outside the lambda so that any caching will be visible to
// all publishers
LazyBean<CurrentTraceContext> lazyCurrentTraceContext = new LazyBean<>(
springContext, CurrentTraceContext.class);
return Operators.liftPublisher((p, sub) -> {
// if 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;
}
Scannable scannable = Scannable.from(p);
// rest of the logic unchanged...
if (isActive.getAsBoolean()) {
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()
+ "]");
}
return scopePassingSpanSubscription(beanFactory, sub);
if (!springContext.isActive()) {
if (log.isTraceEnabled()) {
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;
}
Context context = sub.currentContext();
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()
+ "]");
log.trace("Spring context [" + springContext + "], Reactor context ["
+ context + "], name [" + name(sub) + "]");
}
return new LazySpanSubscriber<>(
lazyScopePassingSpanSubscription(beanFactory, scannable, 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);
});
}
static <T> SpanSubscriptionProvider<T> lazyScopePassingSpanSubscription(
BeanFactory beanFactory, Scannable scannable, CoreSubscriber<? super T> sub) {
return new SpanSubscriptionProvider<>(beanFactory, sub, sub.currentContext(),
scannable.name());
static String name(CoreSubscriber<?> sub) {
return Scannable.from(sub).name();
}
private static Map<BeanFactory, 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));
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();
}
}

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

@@ -54,8 +54,7 @@ final class ScopePassingSpanSubscriber<T> implements SpanSubscription<T>, 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<T> implements SpanSubscription<T>, Scanna
try (Scope scope = this.currentTraceContext.maybeScope(this.parent)) {
this.s.cancel();
}
}
@Override
@@ -120,4 +118,10 @@ final class ScopePassingSpanSubscriber<T> implements SpanSubscription<T>, Scanna
}
}
@Override
public String toString() {
return "ScopePassingSpanSubscriber{" + "subscriber=" + this.subscriber
+ ", parent=" + this.parent + "}";
}
}

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

@@ -109,7 +109,7 @@ final class SpanSubscriber<T> 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());
}
}
}

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

@@ -1,92 +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 org.reactivestreams.Subscriber;
import reactor.util.context.Context;
import org.springframework.beans.factory.BeanFactory;
/**
* 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 Subscriber<? super T> subscriber;
final Context context;
final String name;
private volatile CurrentTraceContext currentTraceContext;
SpanSubscriptionProvider(BeanFactory beanFactory, Subscriber<? super T> subscriber,
Context context, String name) {
this.beanFactory = beanFactory;
this.subscriber = subscriber;
this.context = context;
this.name = name;
if (log.isTraceEnabled()) {
log.trace("Spring context [" + beanFactory + "], Reactor context [" + context
+ "], name [" + name + "]");
}
}
@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;
}
}

38
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;
@@ -47,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;
/**
@@ -76,14 +75,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()) {
@@ -102,9 +101,8 @@ public class TraceReactorAutoConfiguration {
}
@Bean
// for tests
@ConditionalOnMissingBean
static HookRegisteringBeanDefinitionRegistryPostProcessor traceHookRegisteringBeanDefinitionRegistryPostProcessor(
HookRegisteringBeanDefinitionRegistryPostProcessor traceHookRegisteringBeanDefinitionRegistryPostProcessor(
ConfigurableApplicationContext context) {
if (log.isTraceEnabled()) {
log.trace(
@@ -156,14 +154,14 @@ class HooksRefresher implements ApplicationListener<RefreshScopeRefreshedEvent>
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));
}
}
@@ -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();
static 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));
scopePassingSpanOperator(springContext));
}
else {
if (log.isTraceEnabled()) {
log.trace("Decorating onLast operator instrumentation");
}
Hooks.onLastOperator(SLEUTH_TRACE_REACTOR_KEY,
ReactorSleuth.scopePassingSpanOperator(this.context));
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;

202
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java
View File

@@ -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<Span> 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<String, Object> 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<String, Object> 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<Message<?>> messages = new ArrayList<>();
channel.subscribe(messages::add);
Map<String, Object> 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<Message<?>> messages = new ArrayList<>();
channel.subscribe(messages::add);
Map<String, Object> 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<Message<?>> messages = new ArrayList<>();
channel.subscribe(messages::add);
Map<String, Object> 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);
}
}

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

@@ -16,26 +16,29 @@
package org.springframework.cloud.sleuth.instrument.reactor;
import org.assertj.core.api.BDDAssertions;
import brave.propagation.CurrentTraceContext;
import org.junit.Test;
import org.mockito.BDDMockito;
import reactor.util.context.Context;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ConfigurableApplicationContext;
public class SpanSubscriptionProviderTests {
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_default_tracing_instance_when_exception_thrown_upon_bean_retrieval() {
BeanFactory beanFactory = BDDMockito.mock(BeanFactory.class);
BDDMockito.when(beanFactory.getBean(BDDMockito.any(Class.class)))
public void should_return_null_when_exception_thrown_upon_bean_retrieval() {
ConfigurableApplicationContext springContext = mock(
ConfigurableApplicationContext.class);
when(springContext.getBean(CurrentTraceContext.class))
.thenThrow(new IllegalStateException());
SpanSubscriptionProvider provider = new SpanSubscriptionProvider(beanFactory,
null, Context.empty(), "example");
SpanSubscription spanSubscription = provider.get();
LazyBean<CurrentTraceContext> provider = new LazyBean<>(springContext,
CurrentTraceContext.class);
BDDAssertions.then(spanSubscription).isNotNull();
then(provider.get()).isNull();
}
}

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

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

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

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

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