diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java index 740aa1452..08806b453 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java @@ -57,9 +57,9 @@ public class SleuthMessagingProperties { /** * An array of patterns against which channel names will be matched. * @see org.springframework.integration.config.GlobalChannelInterceptor#patterns() - * Defaults to any channel name not matching the Hystrix Stream channel name. + * Defaults to any channel name not matching the Hystrix Stream and functional Stream channel names. */ - private String[] patterns = new String[] { "!hystrixStreamOutput*", "*" }; + private String[] patterns = new String[] { "!hystrixStreamOutput*", "*", "!channel*"}; /** * Enable Spring Integration sleuth instrumentation. diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java index 07d3efd1c..bef74c433 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2019 the original author or authors. + * Copyright 2013-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -59,8 +59,7 @@ import org.springframework.util.ClassUtils; * * @author Marcin Grzejszczak */ -public final class TracingChannelInterceptor extends ChannelInterceptorAdapter - implements ExecutorChannelInterceptor { +public final class TracingChannelInterceptor extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor { /** * Name of the class in Spring Cloud Stream that is a direct channel. @@ -109,25 +108,22 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter @Autowired TracingChannelInterceptor(Tracing tracing) { - this(tracing, MessageHeaderPropagation.INSTANCE, - MessageHeaderPropagation.INSTANCE); + this(tracing, MessageHeaderPropagation.INSTANCE, MessageHeaderPropagation.INSTANCE); } - TracingChannelInterceptor(Tracing tracing, - Propagation.Setter setter, + TracingChannelInterceptor(Tracing tracing, Propagation.Setter setter, Propagation.Getter getter) { this.tracing = tracing; this.tracer = tracing.tracer(); this.threadLocalSpan = ThreadLocalSpan.create(this.tracer); this.injector = tracing.propagation().injector(setter); this.extractor = tracing.propagation().extractor(getter); - this.integrationObjectSupportPresent = ClassUtils.isPresent( - "org.springframework.integration.context.IntegrationObjectSupport", null); - this.hasDirectChannelClass = ClassUtils - .isPresent("org.springframework.integration.channel.DirectChannel", null); - this.directWithAttributesChannelClass = ClassUtils - .isPresent(STREAM_DIRECT_CHANNEL, null) - ? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null; + this.integrationObjectSupportPresent = ClassUtils + .isPresent("org.springframework.integration.context.IntegrationObjectSupport", null); + this.hasDirectChannelClass = ClassUtils.isPresent("org.springframework.integration.channel.DirectChannel", + null); + this.directWithAttributesChannelClass = ClassUtils.isPresent(STREAM_DIRECT_CHANNEL, null) + ? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null; } public static TracingChannelInterceptor create(Tracing tracing) { @@ -170,12 +166,11 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter MessageHeaderAccessor headers = mutableHeaderAccessor(retrievedMessage); TraceContextOrSamplingFlags extracted = this.extractor.extract(headers); Span span = this.threadLocalSpan.next(extracted); - MessageHeaderPropagation.removeAnyTraceHeaders(headers, - this.tracing.propagation().keys()); + MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys()); this.injector.inject(span.context(), headers); if (!span.isNoop()) { span.kind(Span.Kind.PRODUCER).name("send").start(); - span.remoteServiceName(REMOTE_SERVICE_NAME); + span.remoteServiceName(toRemoteServiceName(headers)); addTags(message, span, channel); } if (log.isDebugEnabled()) { @@ -188,24 +183,31 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter return outputMessage; } - private Message outputMessage(Message originalMessage, - Message retrievedMessage, MessageHeaderAccessor additionalHeaders) { - MessageHeaderAccessor headers = MessageHeaderAccessor - .getMutableAccessor(originalMessage); + private String toRemoteServiceName(MessageHeaderAccessor headers) { + for (String key : headers.getMessageHeaders().keySet()) { + if (key.startsWith("kafka_")) { + return "kafka"; + } + else if (key.startsWith("amqp_")) { + return "rabbitmq"; + } + } + return REMOTE_SERVICE_NAME; + } + + private Message outputMessage(Message originalMessage, Message retrievedMessage, + MessageHeaderAccessor additionalHeaders) { + MessageHeaderAccessor headers = MessageHeaderAccessor.getMutableAccessor(originalMessage); if (originalMessage instanceof ErrorMessage) { ErrorMessage errorMessage = (ErrorMessage) originalMessage; - headers.copyHeaders(MessageHeaderPropagation.propagationHeaders( - additionalHeaders.getMessageHeaders(), + headers.copyHeaders(MessageHeaderPropagation.propagationHeaders(additionalHeaders.getMessageHeaders(), this.tracing.propagation().keys())); - return new ErrorMessage(errorMessage.getPayload(), - isWebSockets(headers) ? headers.getMessageHeaders() - : new MessageHeaders(headers.getMessageHeaders()), - errorMessage.getOriginalMessage()); + return new ErrorMessage(errorMessage.getPayload(), isWebSockets(headers) ? headers.getMessageHeaders() + : new MessageHeaders(headers.getMessageHeaders()), errorMessage.getOriginalMessage()); } headers.copyHeaders(additionalHeaders.getMessageHeaders()); return new GenericMessage<>(retrievedMessage.getPayload(), - isWebSockets(headers) ? headers.getMessageHeaders() - : new MessageHeaders(headers.getMessageHeaders())); + isWebSockets(headers) ? headers.getMessageHeaders() : new MessageHeaders(headers.getMessageHeaders())); } private boolean isWebSockets(MessageHeaderAccessor headerAccessor) { @@ -215,8 +217,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter private boolean isDirectChannel(MessageChannel channel) { Class targetClass = AopUtils.getTargetClass(channel); - boolean directChannel = this.hasDirectChannelClass - && DirectChannel.class.isAssignableFrom(targetClass); + boolean directChannel = this.hasDirectChannelClass && DirectChannel.class.isAssignableFrom(targetClass); if (!directChannel) { return false; } @@ -231,8 +232,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter } @Override - public void afterSendCompletion(Message message, MessageChannel channel, - boolean sent, Exception ex) { + public void afterSendCompletion(Message message, MessageChannel channel, boolean sent, Exception ex) { if (emptyMessage(message)) { return; } @@ -240,8 +240,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter afterMessageHandled(message, channel, null, ex); } if (log.isDebugEnabled()) { - log.debug("Will finish the current span after completion " - + this.tracer.currentSpan()); + log.debug("Will finish the current span after completion " + this.tracer.currentSpan()); } finishSpan(ex); } @@ -258,12 +257,11 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter MessageHeaderAccessor headers = mutableHeaderAccessor(message); TraceContextOrSamplingFlags extracted = this.extractor.extract(headers); Span span = this.threadLocalSpan.next(extracted); - MessageHeaderPropagation.removeAnyTraceHeaders(headers, - this.tracing.propagation().keys()); + MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys()); this.injector.inject(span.context(), headers); if (!span.isNoop()) { span.kind(Span.Kind.CONSUMER).name("receive").start(); - span.remoteServiceName(REMOTE_SERVICE_NAME); + span.remoteServiceName(toRemoteServiceName(headers)); addTags(message, span, channel); } if (log.isDebugEnabled()) { @@ -272,21 +270,19 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter headers.setImmutable(); if (message instanceof ErrorMessage) { ErrorMessage errorMessage = (ErrorMessage) message; - return new ErrorMessage(errorMessage.getPayload(), - headers.getMessageHeaders(), errorMessage.getOriginalMessage()); + return new ErrorMessage(errorMessage.getPayload(), headers.getMessageHeaders(), + errorMessage.getOriginalMessage()); } return new GenericMessage<>(message.getPayload(), headers.getMessageHeaders()); } @Override - public void afterReceiveCompletion(Message message, MessageChannel channel, - Exception ex) { + public void afterReceiveCompletion(Message message, MessageChannel channel, Exception ex) { if (emptyMessage(message)) { return; } if (log.isDebugEnabled()) { - log.debug("Will finish the current span after receive completion " - + this.tracer.currentSpan()); + log.debug("Will finish the current span after receive completion " + this.tracer.currentSpan()); } finishSpan(ex); } @@ -296,8 +292,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter * context. It then creates a span for the handler, placing it in scope. */ @Override - public Message beforeHandle(Message message, MessageChannel channel, - MessageHandler handler) { + public Message beforeHandle(Message message, MessageChannel channel, MessageHandler handler) { if (emptyMessage(message)) { return message; } @@ -312,34 +307,28 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter consumerSpan.finish(); } // create and scope a span for the message processor - this.threadLocalSpan - .next(TraceContextOrSamplingFlags.create(consumerSpan.context())) - .name("handle").start(); + this.threadLocalSpan.next(TraceContextOrSamplingFlags.create(consumerSpan.context())).name("handle").start(); // remove any trace headers, but don't re-inject as we are synchronously // processing the // message and can rely on scoping to access this span later. - MessageHeaderPropagation.removeAnyTraceHeaders(headers, - this.tracing.propagation().keys()); + MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys()); if (log.isDebugEnabled()) { log.debug("Created a new span in before handle" + consumerSpan); } if (message instanceof ErrorMessage) { - return new ErrorMessage((Throwable) message.getPayload(), - headers.getMessageHeaders()); + return new ErrorMessage((Throwable) message.getPayload(), headers.getMessageHeaders()); } headers.setImmutable(); return new GenericMessage<>(message.getPayload(), headers.getMessageHeaders()); } @Override - public void afterMessageHandled(Message message, MessageChannel channel, - MessageHandler handler, Exception ex) { + public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, Exception ex) { if (emptyMessage(message)) { return; } if (log.isDebugEnabled()) { - log.debug("Will finish the current span after message handled " - + this.tracer.currentSpan()); + log.debug("Will finish the current span after message handled " + this.tracer.currentSpan()); } finishSpan(ex); } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBean.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBean.java new file mode 100644 index 000000000..364d18648 --- /dev/null +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBean.java @@ -0,0 +1,67 @@ +/* + * Copyright 2013-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.instrument.reactor; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.lang.Nullable; + +/** + * Avoids calling the expensive {@link ConfigurableApplicationContext#getBean(Class)} many + * times or throwing an exception. + */ +final class LazyBean { + + // spring-jcl uses commons-logging, so do we. + private static final Log log = LogFactory.getLog(LazyBean.class); + + final ConfigurableApplicationContext springContext; + + final Class requiredType; + + T value; + + LazyBean(ConfigurableApplicationContext springContext, Class requiredType) { + this.springContext = springContext; + this.requiredType = requiredType; + } + + /** + * Attempts to provision from the underlying bean factory, if not already provisioned. + * @return the bean value or null if there was an exception getting it. + */ + @Nullable + T get() { + if (this.value != null) { + return this.value; + } + + try { + this.value = springContext.getBean(requiredType); + } + catch (Exception ex) { + if (log.isDebugEnabled()) { + log.debug("Spring context [" + springContext + "] error getting [" + + requiredType + "].", ex); + } + } + return this.value; + } + +} diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java deleted file mode 100644 index 576c6ffc0..000000000 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java +++ /dev/null @@ -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 of what subscription returns - * @author Marcin Grzejszczak - * @since 2.0.0 - */ -final class LazySpanSubscriber extends AtomicBoolean implements SpanSubscription { - - private final Supplier> supplier; - - LazySpanSubscriber(Supplier> supplier) { - this.supplier = supplier; - } - - @Override - public void onSubscribe(Subscription subscription) { - this.supplier.get().onSubscribe(subscription); - } - - @Override - public void request(long n) { - this.supplier.get().request(n); - } - - @Override - public void cancel() { - this.supplier.get().cancel(); - } - - @Override - public void onNext(T o) { - this.supplier.get().onNext(o); - } - - @Override - public void onError(Throwable throwable) { - this.supplier.get().onError(throwable); - } - - @Override - public void onComplete() { - this.supplier.get().onComplete(); - } - - @Override - public Context currentContext() { - return this.supplier.get().currentContext(); - } - -} diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java index 10df1a826..7f095b779 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java @@ -16,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 an arbitrary type that is left unchanged by the span operator * @return a new lazy span operator pointcut */ - @SuppressWarnings("unchecked") + // Much of Boot assumes that the Spring context will be a + // ConfigurableApplicationContext, rooted in SpringApplication's + // requirement for it to be so. Previous versions of Reactor + // instrumentation injected both BeanFactory and also + // ConfigurableApplicationContext. This chooses the more narrow + // signature as it is simpler than explaining instanceof checks. public static Function, ? extends Publisher> scopePassingSpanOperator( - BeanFactory beanFactory) { + ConfigurableApplicationContext springContext) { if (log.isTraceEnabled()) { - log.trace("Scope passing operator [" + beanFactory + "]"); + log.trace("Scope passing operator [" + springContext + "]"); } - // 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 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 SpanSubscriptionProvider lazyScopePassingSpanSubscription( - BeanFactory beanFactory, Scannable scannable, CoreSubscriber sub) { - return new SpanSubscriptionProvider<>(beanFactory, sub, sub.currentContext(), - scannable.name()); + static String name(CoreSubscriber sub) { + return Scannable.from(sub).name(); } - private static Map CACHE = new ConcurrentHashMap<>(); - - static CoreSubscriber scopePassingSpanSubscription( - BeanFactory beanFactory, CoreSubscriber 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(); } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java index 488e26632..2e24dc278 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java @@ -54,8 +54,7 @@ final class ScopePassingSpanSubscriber implements SpanSubscription, Scanna this.subscriber = subscriber; this.currentTraceContext = currentTraceContext; this.parent = parent; - this.context = ctx != null && parent != null ? ctx.put(TraceContext.class, parent) - : ctx != null ? ctx : Context.empty(); + this.context = parent != null ? ctx.put(TraceContext.class, parent) : ctx; if (log.isTraceEnabled()) { log.trace("Parent span [" + parent + "], context [" + this.context + "]"); } @@ -81,7 +80,6 @@ final class ScopePassingSpanSubscriber implements SpanSubscription, Scanna try (Scope scope = this.currentTraceContext.maybeScope(this.parent)) { this.s.cancel(); } - } @Override @@ -120,4 +118,10 @@ final class ScopePassingSpanSubscriber implements SpanSubscription, Scanna } } + @Override + public String toString() { + return "ScopePassingSpanSubscriber{" + "subscriber=" + this.subscriber + + ", parent=" + this.parent + "}"; + } + } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java index 61ad068fc..b5d0014d9 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java @@ -109,7 +109,7 @@ final class SpanSubscriber extends AtomicBoolean implements SpanSubscription< // no additional cleaning is required cause we operate on scopes if (log.isTraceEnabled()) { log.trace("Request after cleaning. Current span [{}]", - this.currentTraceContext.get()); + this.span.context()); } } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java deleted file mode 100644 index e90b34836..000000000 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java +++ /dev/null @@ -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 type of returned subscription - * @author Marcin Grzejszczak - */ -final class SpanSubscriptionProvider implements Supplier> { - - private static final Log log = LogFactory.getLog(SpanSubscriptionProvider.class); - - final BeanFactory beanFactory; - - final Subscriber subscriber; - - final Context context; - - final String name; - - private volatile CurrentTraceContext currentTraceContext; - - SpanSubscriptionProvider(BeanFactory beanFactory, Subscriber 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 get() { - return newCoreSubscriber(currentTraceContext()); - } - - SpanSubscription newCoreSubscriber(CurrentTraceContext currentTraceContext) { - TraceContext root = this.context.hasKey(TraceContext.class) - ? this.context.get(TraceContext.class) : currentTraceContext.get(); - return new ScopePassingSpanSubscriber<>(this.subscriber, this.context, - currentTraceContext, root); - } - - private CurrentTraceContext currentTraceContext() { - if (this.currentTraceContext == null) { - try { - this.currentTraceContext = this.beanFactory - .getBean(CurrentTraceContext.class); - } - catch (Exception ex) { - if (log.isDebugEnabled()) { - log.debug( - "Exception occurred while trying to get the currentTraceContext bean. Will return a default instance", - ex); - } - return CurrentTraceContext.Default.create(); - } - } - return this.currentTraceContext; - } - -} diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java index c0c6c6ed4..1f913ec08 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java @@ -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 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)); } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java index c78114d84..ef596dddf 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java @@ -43,6 +43,7 @@ import org.springframework.boot.web.client.RestTemplateCustomizer; import org.springframework.cloud.commons.httpclient.HttpClientConfiguration; import org.springframework.cloud.gateway.filter.headers.HttpHeadersFilter; import org.springframework.cloud.sleuth.instrument.web.TraceWebServletAutoConfiguration; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.annotation.Order; @@ -133,8 +134,8 @@ public class TraceWebClientAutoConfiguration { @Bean static TraceWebClientBeanPostProcessor traceWebClientBeanPostProcessor( - BeanFactory beanFactory) { - return new TraceWebClientBeanPostProcessor(beanFactory); + ConfigurableApplicationContext springContext) { + return new TraceWebClientBeanPostProcessor(springContext); } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java index 8cafff570..558f384cf 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java @@ -38,9 +38,8 @@ import reactor.util.annotation.Nullable; import reactor.util.context.Context; import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.BeanPostProcessor; -import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.web.client.RestClientException; import org.springframework.web.reactive.function.client.ClientRequest; @@ -49,6 +48,8 @@ import org.springframework.web.reactive.function.client.ExchangeFilterFunction; import org.springframework.web.reactive.function.client.ExchangeFunction; import org.springframework.web.reactive.function.client.WebClient; +import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.scopePassingSpanOperator; + /** * {@link BeanPostProcessor} to wrap a {@link WebClient} instance into its trace * representation. @@ -58,10 +59,10 @@ import org.springframework.web.reactive.function.client.WebClient; */ final class TraceWebClientBeanPostProcessor implements BeanPostProcessor { - private final BeanFactory beanFactory; + private final ConfigurableApplicationContext springContext; - TraceWebClientBeanPostProcessor(BeanFactory beanFactory) { - this.beanFactory = beanFactory; + TraceWebClientBeanPostProcessor(ConfigurableApplicationContext springContext) { + this.springContext = springContext; } @Override @@ -92,7 +93,7 @@ final class TraceWebClientBeanPostProcessor implements BeanPostProcessor { return functions -> { boolean noneMatch = noneMatchTraceExchangeFunction(functions); if (noneMatch) { - functions.add(new TraceExchangeFilterFunction(this.beanFactory)); + functions.add(new TraceExchangeFilterFunction(this.springContext)); } }; } @@ -134,7 +135,7 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { private static final String CANCELLED_SUBSCRIPTION_ERROR = "CANCELLED"; - final BeanFactory beanFactory; + final ConfigurableApplicationContext springContext; final Function, ? extends Publisher> scopePassingTransformer; @@ -146,14 +147,14 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { TraceContext.Injector injector; - TraceExchangeFilterFunction(BeanFactory beanFactory) { - this.beanFactory = beanFactory; - this.scopePassingTransformer = ReactorSleuth - .scopePassingSpanOperator(beanFactory); + TraceExchangeFilterFunction(ConfigurableApplicationContext springContext) { + this.springContext = springContext; + this.scopePassingTransformer = scopePassingSpanOperator(springContext); } - public static ExchangeFilterFunction create(BeanFactory beanFactory) { - return new TraceExchangeFilterFunction(beanFactory); + public static ExchangeFilterFunction create( + ConfigurableApplicationContext springContext) { + return new TraceExchangeFilterFunction(springContext); } @Override @@ -177,7 +178,7 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { HttpClientHandler handler() { if (this.handler == null) { this.handler = HttpClientHandler - .create(this.beanFactory.getBean(HttpTracing.class)); + .create(this.springContext.getBean(HttpTracing.class)); } return this.handler; } @@ -191,14 +192,14 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { HttpTracing httpTracing() { if (this.httpTracing == null) { - this.httpTracing = this.beanFactory.getBean(HttpTracing.class); + this.httpTracing = this.springContext.getBean(HttpTracing.class); } return this.httpTracing; } TraceContext.Injector injector() { if (this.injector == null) { - this.injector = this.beanFactory.getBean(HttpTracing.class).tracing() + this.injector = this.springContext.getBean(HttpTracing.class).tracing() .propagation().injector(SETTER); } return this.injector; diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java index 1d8423f6c..10b5c1c91 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java @@ -29,8 +29,10 @@ import org.junit.After; import org.junit.Test; import zipkin2.Span; +import org.springframework.amqp.support.AmqpHeaders; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.kafka.support.KafkaHeaders; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -51,10 +53,11 @@ public class TracingChannelInterceptorTest { List spans = new ArrayList<>(); - ChannelInterceptor interceptor = TracingChannelInterceptor.create(Tracing.newBuilder() - .currentTraceContext(ThreadLocalCurrentTraceContext.newBuilder() - .addScopeDecorator(StrictScopeDecorator.create()).build()) - .spanReporter(this.spans::add).build()); + ChannelInterceptor interceptor = TracingChannelInterceptor + .create(Tracing.newBuilder() + .currentTraceContext(ThreadLocalCurrentTraceContext.newBuilder() + .addScopeDecorator(StrictScopeDecorator.create()).build()) + .spanReporter(this.spans::add).build()); QueueChannel channel = new QueueChannel(); @@ -83,10 +86,9 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", - "X-B3-SpanId", "X-B3-Sampled", "nativeHeaders"); - assertThat(this.spans).hasSize(1).flatExtracting(Span::kind) - .containsExactly(Span.Kind.PRODUCER); + assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", + "nativeHeaders"); + assertThat(this.spans).hasSize(1).flatExtracting(Span::kind).containsExactly(Span.Kind.PRODUCER); } @Test @@ -96,10 +98,9 @@ public class TracingChannelInterceptorTest { this.directChannel.send(MessageBuilder.withPayload("foo").build()); assertThat(this.message).isNotNull(); - assertThat(this.message.getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", - "X-B3-Sampled", "nativeHeaders"); - assertThat(this.spans).flatExtracting(Span::kind).contains(Span.Kind.CONSUMER, - Span.Kind.PRODUCER); + assertThat(this.message.getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", + "nativeHeaders"); + assertThat(this.spans).flatExtracting(Span::kind).contains(Span.Kind.CONSUMER, Span.Kind.PRODUCER); } @Test @@ -108,9 +109,8 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); - assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)) - .containsOnlyKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", - "spanTraceId", "spanId", "spanSampled"); + assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsOnlyKeys("X-B3-TraceId", + "X-B3-SpanId", "X-B3-Sampled", "spanTraceId", "spanId", "spanSampled"); } /** @@ -122,13 +122,11 @@ public class TracingChannelInterceptorTest { public void producerConsidersOldSpanIds() { this.channel.addInterceptor(producerSideOnly(this.interceptor)); - this.channel.send(MessageBuilder.withPayload("foo") - .setHeader("X-B3-TraceId", "000000000000000a") - .setHeader("X-B3-ParentSpanId", "000000000000000a") - .setHeader("X-B3-SpanId", "000000000000000b").build()); + this.channel.send(MessageBuilder.withPayload("foo").setHeader("X-B3-TraceId", "000000000000000a") + .setHeader("X-B3-ParentSpanId", "000000000000000a").setHeader("X-B3-SpanId", "000000000000000b") + .build()); - assertThat(this.channel.receive().getHeaders()).containsEntry("X-B3-ParentSpanId", - "000000000000000b"); + assertThat(this.channel.receive().getHeaders()).containsEntry("X-B3-ParentSpanId", "000000000000000b"); } @Test @@ -142,12 +140,10 @@ public class TracingChannelInterceptorTest { accessor.setNativeHeader("X-B3-ParentSpanId", "000000000000000a"); accessor.setNativeHeader("X-B3-SpanId", "000000000000000b"); - this.channel.send(MessageBuilder.withPayload("foo") - .copyHeaders(accessor.toMessageHeaders()).build()); + this.channel.send(MessageBuilder.withPayload("foo").copyHeaders(accessor.toMessageHeaders()).build()); - assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)) - .containsEntry("X-B3-ParentSpanId", - Collections.singletonList("000000000000000b")); + assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsEntry("X-B3-ParentSpanId", + Collections.singletonList("000000000000000b")); } /** @@ -160,10 +156,9 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", - "X-B3-SpanId", "X-B3-Sampled", "nativeHeaders"); - assertThat(this.spans).hasSize(1).flatExtracting(Span::kind) - .containsExactly(Span.Kind.CONSUMER); + assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", + "nativeHeaders"); + assertThat(this.spans).hasSize(1).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER); } @Test @@ -172,9 +167,8 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); - assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)) - .containsOnlyKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", - "spanTraceId", "spanId", "spanSampled"); + assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsOnlyKeys("X-B3-TraceId", + "X-B3-SpanId", "X-B3-Sampled", "spanTraceId", "spanId", "spanSampled"); } @Test @@ -186,10 +180,9 @@ public class TracingChannelInterceptorTest { channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", - "X-B3-SpanId", "X-B3-Sampled", "nativeHeaders"); - assertThat(this.spans).flatExtracting(Span::kind) - .containsExactly(Span.Kind.CONSUMER, null); + assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", + "nativeHeaders"); + assertThat(this.spans).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER, null); } /** @@ -206,8 +199,7 @@ public class TracingChannelInterceptorTest { channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", - "X-B3-SpanId", "X-B3-Sampled"); + assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled"); } @Test @@ -219,8 +211,8 @@ public class TracingChannelInterceptorTest { channel.send(MessageBuilder.withPayload("foo").build()); - assertThat((Map) messages.get(0).getHeaders().get(NATIVE_HEADERS)) - .doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled"); + assertThat((Map) messages.get(0).getHeaders().get(NATIVE_HEADERS)).doesNotContainKeys("X-B3-TraceId", + "X-B3-SpanId", "X-B3-Sampled"); } @Test @@ -230,8 +222,8 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); this.channel.receive(); - assertThat(this.spans).flatExtracting(Span::kind) - .containsExactlyInAnyOrder(Span.Kind.CONSUMER, Span.Kind.PRODUCER); + assertThat(this.spans).flatExtracting(Span::kind).containsExactlyInAnyOrder(Span.Kind.CONSUMER, + Span.Kind.PRODUCER); } @Test @@ -243,8 +235,7 @@ public class TracingChannelInterceptorTest { channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(this.spans).flatExtracting(Span::kind) - .containsExactly(Span.Kind.CONSUMER, null, Span.Kind.PRODUCER); + assertThat(this.spans).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER, null, Span.Kind.PRODUCER); } @Test @@ -255,54 +246,41 @@ public class TracingChannelInterceptorTest { Map errorChannelHeaders = new HashMap<>(); errorChannelHeaders.put(MessageHeaders.REPLY_CHANNEL, errorsReplyChannel); errorChannelHeaders.put(MessageHeaders.ERROR_CHANNEL, errorsReplyChannel); - this.channel - .send(new ErrorMessage( - new MessagingException(MessageBuilder.withPayload("hi") - .setHeader(TraceMessageHeaders.TRACE_ID_NAME, - "000000000000000a") - .setHeader(TraceMessageHeaders.SPAN_ID_NAME, - "000000000000000a") - .setReplyChannel(deadReplyChannel) - .setErrorChannel(deadReplyChannel).build()), - errorChannelHeaders)); + this.channel.send(new ErrorMessage( + new MessagingException(MessageBuilder.withPayload("hi") + .setHeader(TraceMessageHeaders.TRACE_ID_NAME, "000000000000000a") + .setHeader(TraceMessageHeaders.SPAN_ID_NAME, "000000000000000a") + .setReplyChannel(deadReplyChannel).setErrorChannel(deadReplyChannel).build()), + errorChannelHeaders)); this.message = this.channel.receive(); assertThat(this.message).isNotNull(); - String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, - String.class); + String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, String.class); assertThat(spanId).isNotNull(); - String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, - String.class); + String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, String.class); assertThat(traceId).isEqualTo("000000000000000a"); assertThat(spanId).isNotEqualTo("000000000000000a"); assertThat(this.spans).hasSize(2); - assertThat(this.message.getHeaders().getReplyChannel()) - .isSameAs(errorsReplyChannel); - assertThat(this.message.getHeaders().getErrorChannel()) - .isSameAs(errorsReplyChannel); + assertThat(this.message.getHeaders().getReplyChannel()).isSameAs(errorsReplyChannel); + assertThat(this.message.getHeaders().getErrorChannel()).isSameAs(errorsReplyChannel); } @Test public void errorMessageOriginalMessageRetained() { this.channel.addInterceptor(this.interceptor); - Message originalMessage = MessageBuilder.withPayload("Hello") - .setHeader("header", "value").build(); - Message failedMessage = MessageBuilder.fromMessage(originalMessage) - .removeHeader("header").build(); - this.channel.send(new ErrorMessage(new MessagingException(failedMessage), - originalMessage.getHeaders(), originalMessage)); + Message originalMessage = MessageBuilder.withPayload("Hello").setHeader("header", "value").build(); + Message failedMessage = MessageBuilder.fromMessage(originalMessage).removeHeader("header").build(); + this.channel.send( + new ErrorMessage(new MessagingException(failedMessage), originalMessage.getHeaders(), originalMessage)); this.message = this.channel.receive(); assertThat(this.message).isNotNull(); - assertThat(this.message).isInstanceOfSatisfying(ErrorMessage.class, - errorMessage -> { - assertThat(errorMessage.getOriginalMessage()) - .isSameAs(originalMessage); - assertThat(errorMessage.getHeaders().get("header")) - .isEqualTo("value"); - }); + assertThat(this.message).isInstanceOfSatisfying(ErrorMessage.class, errorMessage -> { + assertThat(errorMessage.getOriginalMessage()).isSameAs(originalMessage); + assertThat(errorMessage.getHeaders().get("header")).isEqualTo("value"); + }); } @Test @@ -311,22 +289,60 @@ public class TracingChannelInterceptorTest { Map errorChannelHeaders = new HashMap<>(); errorChannelHeaders.put(TraceMessageHeaders.TRACE_ID_NAME, "000000000000000a"); errorChannelHeaders.put(TraceMessageHeaders.SPAN_ID_NAME, "000000000000000a"); - this.channel.send(new ErrorMessage(new MessagingException("exception"), - errorChannelHeaders)); + this.channel.send(new ErrorMessage(new MessagingException("exception"), errorChannelHeaders)); this.message = this.channel.receive(); assertThat(this.message).isNotNull(); - String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, - String.class); + String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, String.class); assertThat(spanId).isNotNull(); - String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, - String.class); + String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, String.class); assertThat(traceId).isEqualTo("000000000000000a"); assertThat(spanId).isNotEqualTo("000000000000000a"); assertThat(this.spans).hasSize(2); } + @Test + public void should_store_kafka_as_remote_service_name_when_kafka_header_is_present() { + ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(); + channel.addInterceptor(this.interceptor); + List> messages = new ArrayList<>(); + channel.subscribe(messages::add); + + Map headers = new HashMap<>(); + headers.put(KafkaHeaders.MESSAGE_KEY, "hello"); + channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers))); + + assertThat(this.spans).flatExtracting(Span::remoteServiceName).contains("kafka"); + } + + @Test + public void should_store_rabbitmq_as_remote_service_name_when_rabbit_header_is_present() { + ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(); + channel.addInterceptor(this.interceptor); + List> messages = new ArrayList<>(); + channel.subscribe(messages::add); + + Map headers = new HashMap<>(); + headers.put(AmqpHeaders.RECEIVED_ROUTING_KEY, "hello"); + channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers))); + + assertThat(this.spans).flatExtracting(Span::remoteServiceName).contains("rabbitmq"); + } + + @Test + public void should_store_broker_as_remote_service_name_when_no_special_headers_were_found() { + ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(); + channel.addInterceptor(this.interceptor); + List> messages = new ArrayList<>(); + channel.subscribe(messages::add); + + Map headers = new HashMap<>(); + channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers))); + + assertThat(this.spans).flatExtracting(Span::remoteServiceName).containsOnly("broker", null); + } + ChannelInterceptor producerSideOnly(ChannelInterceptor delegate) { return new ChannelInterceptorAdapter() { @Override @@ -335,8 +351,7 @@ public class TracingChannelInterceptorTest { } @Override - public void afterSendCompletion(Message message, MessageChannel channel, - boolean sent, Exception ex) { + public void afterSendCompletion(Message message, MessageChannel channel, boolean sent, Exception ex) { delegate.afterSendCompletion(message, channel, sent, ex); } }; @@ -350,29 +365,24 @@ public class TracingChannelInterceptorTest { } @Override - public void afterReceiveCompletion(Message message, MessageChannel channel, - Exception ex) { + public void afterReceiveCompletion(Message message, MessageChannel channel, Exception ex) { delegate.afterReceiveCompletion(message, channel, ex); } }; } ExecutorChannelInterceptor executorSideOnly(ChannelInterceptor delegate) { - class ExecutorSideOnly extends ChannelInterceptorAdapter - implements ExecutorChannelInterceptor { + class ExecutorSideOnly extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor { @Override - public Message beforeHandle(Message message, MessageChannel channel, - MessageHandler handler) { - return ((ExecutorChannelInterceptor) delegate).beforeHandle(message, - channel, handler); + public Message beforeHandle(Message message, MessageChannel channel, MessageHandler handler) { + return ((ExecutorChannelInterceptor) delegate).beforeHandle(message, channel, handler); } @Override - public void afterMessageHandled(Message message, MessageChannel channel, - MessageHandler handler, Exception ex) { - ((ExecutorChannelInterceptor) delegate).afterMessageHandled(message, - channel, handler, ex); + public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, + Exception ex) { + ((ExecutorChannelInterceptor) delegate).afterMessageHandled(message, channel, handler, ex); } } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBeanTests.java similarity index 53% rename from spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java rename to spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBeanTests.java index 6838e4f3a..3f9694ba6 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/LazyBeanTests.java @@ -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 provider = new LazyBean<>(springContext, + CurrentTraceContext.class); - BDDAssertions.then(spanSubscription).isNotNull(); + then(provider.get()).isNull(); } } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java index 7cbf973f6..763574221 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java @@ -51,9 +51,7 @@ public final class TraceReactorAutoConfigurationAccessorConfiguration { if (log.isTraceEnabled()) { log.trace("Setting up hooks"); } - TraceReactorAutoConfiguration.TraceReactorConfiguration - .traceHookRegisteringBeanDefinitionRegistryPostProcessor(context) - .setupHooks(context); + HookRegisteringBeanDefinitionRegistryPostProcessor.setupHooks(context); } } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java index db74c2a13..4bf00805c 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java @@ -22,7 +22,7 @@ import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; -import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.web.reactive.function.client.WebClient; /** @@ -32,12 +32,12 @@ import org.springframework.web.reactive.function.client.WebClient; public class TraceWebClientBeanPostProcessorTest { @Mock - BeanFactory beanFactory; + ConfigurableApplicationContext springContext; @Test public void should_add_filter_only_once_to_web_client() { TraceWebClientBeanPostProcessor processor = new TraceWebClientBeanPostProcessor( - this.beanFactory); + this.springContext); WebClient client = WebClient.create(); client = (WebClient) processor.postProcessAfterInitialization(client, "foo"); @@ -53,7 +53,7 @@ public class TraceWebClientBeanPostProcessorTest { @Test public void should_add_filter_only_once_to_web_client_via_builder() { TraceWebClientBeanPostProcessor processor = new TraceWebClientBeanPostProcessor( - this.beanFactory); + this.springContext); WebClient.Builder builder = WebClient.builder(); builder = (WebClient.Builder) processor.postProcessAfterInitialization(builder, diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberSpringBootTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberSpringBootTests.java new file mode 100644 index 000000000..ef0a6bbd8 --- /dev/null +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberSpringBootTests.java @@ -0,0 +1,204 @@ +/* + * Copyright 2013-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.instrument.reactor; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicReference; + +import brave.propagation.CurrentTraceContext; +import brave.propagation.CurrentTraceContext.Scope; +import brave.propagation.TraceContext; +import brave.sampler.Sampler; +import org.awaitility.Awaitility; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.BDDAssertions.then; + +/** + * Like {@link ScopePassingSpanSubscriberTests}, except this tests wiring with spring boot + * config. + */ +@RunWith(SpringRunner.class) +@SpringBootTest(classes = ScopePassingSpanSubscriberSpringBootTests.Config.class, + webEnvironment = SpringBootTest.WebEnvironment.NONE) +public class ScopePassingSpanSubscriberSpringBootTests { + + @Autowired + CurrentTraceContext currentTraceContext; + + TraceContext context = TraceContext.newBuilder().traceId(1).spanId(1).sampled(true) + .build(); + + TraceContext context2 = TraceContext.newBuilder().traceId(1).spanId(2).sampled(true) + .build(); + + @Test + public void should_pass_tracing_info_when_using_reactor() { + final AtomicReference spanInOperation = new AtomicReference<>(); + Publisher traced = Flux.just(1, 2, 3); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Flux.from(traced).map(d -> d + 1).map(d -> d + 1).map((d) -> { + spanInOperation.set(this.currentTraceContext.get()); + return d + 1; + }).map(d -> d + 1).subscribe(d -> { + }); + } + + then(this.currentTraceContext.get()).isNull(); + then(spanInOperation.get()).isEqualTo(context); + } + + @Test + public void should_support_reactor_fusion_optimization() { + final AtomicReference spanInOperation = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Mono.just(1).flatMap(d -> Flux.just(d + 1).collectList().map(p -> p.get(0))) + .map(d -> d + 1).map((d) -> { + spanInOperation.set(this.currentTraceContext.get()); + return d + 1; + }).map(d -> d + 1).subscribe(d -> { + }); + } + + then(this.currentTraceContext.get()).isNull(); + then(spanInOperation.get()).isEqualTo(context); + } + + @Test + public void should_pass_tracing_info_when_using_reactor_async() { + final AtomicReference spanInOperation = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.1") + .map(d -> d + 1).map(d -> d + 1) + .publishOn(Schedulers.newSingle("secondThread")).log("reactor.2") + .map((d) -> { + spanInOperation.set(this.currentTraceContext.get()); + return d + 1; + }).map(d -> d + 1).blockLast(); + + Awaitility.await() + .untilAsserted(() -> then(spanInOperation.get()).isEqualTo(context)); + then(this.currentTraceContext.get()).isEqualTo(context); + } + + then(this.currentTraceContext.get()).isNull(); + + try (Scope ws = this.currentTraceContext.newScope(context2)) { + Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.") + .map(d -> d + 1).map(d -> d + 1).map((d) -> { + spanInOperation.set(this.currentTraceContext.get()); + return d + 1; + }).map(d -> d + 1).blockLast(); + + then(this.currentTraceContext.get()).isEqualTo(context2); + then(spanInOperation.get()).isEqualTo(context2); + } + + then(this.currentTraceContext.get()).isNull(); + } + + @Test + public void onlyConsidersContextDuringSubscribe() { + Mono fromMono = Mono.fromCallable(this.currentTraceContext::get); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + then(fromMono.map(context -> context).block()).isNotNull(); + } + } + + @Test + public void checkTraceIdDuringZipOperation() { + final AtomicReference spanInOperation = new AtomicReference<>(); + final AtomicReference spanInZipOperation = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Mono.fromCallable(this.currentTraceContext::get).map(span -> span) + .doOnNext(spanInOperation::set) + .zipWith(Mono.fromCallable(this.currentTraceContext::get) + .map(span -> span).doOnNext(spanInZipOperation::set)) + .block(); + } + + then(spanInZipOperation).hasValue(context); + then(spanInOperation).hasValue(context); + } + + // #646 + @Test + public void should_work_for_mono_just_with_flat_map() { + try (Scope ws = this.currentTraceContext.newScope(context)) { + Mono.just("value1") + .flatMap(request -> Mono.just("value2").then(Mono.just("foo"))) + .map(a -> "qwe").block(); + } + } + + // #1030 + @Test + public void checkTraceIdFromSubscriberContext() { + final AtomicReference spanInSubscriberContext = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Mono.subscriberContext().map(context -> this.currentTraceContext.get()) + .doOnNext(spanInSubscriberContext::set).block(); + } + + then(spanInSubscriberContext).hasValue(context); // ok here + } + + @Test + public void should_pass_tracing_info_into_inner_publishers() { + final AtomicReference spanInOperation = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Flux.range(0, 5) + .flatMap(it -> Mono.delay(Duration.ofMillis(1)) + .map(context -> this.currentTraceContext.get()) + .doOnNext(spanInOperation::set)) + .blockFirst(); + } + + then(spanInOperation.get()).isEqualTo(context); + } + + @EnableAutoConfiguration + @Configuration + static class Config { + + @Bean + Sampler sampler() { + return Sampler.ALWAYS_SAMPLE; + } + + } + +} diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java index 6c0bf89c6..b809a743e 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java @@ -16,32 +16,59 @@ package org.springframework.cloud.sleuth.instrument.reactor; +import java.util.Objects; +import java.util.function.Function; + import brave.propagation.CurrentTraceContext; import brave.propagation.CurrentTraceContext.Scope; import brave.propagation.TraceContext; +import org.assertj.core.presentation.StandardRepresentation; +import org.awaitility.Awaitility; +import org.junit.After; import org.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.BDDMockito; -import org.mockito.junit.MockitoJUnitRunner; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; import reactor.core.CoreSubscriber; import reactor.core.publisher.BaseSubscriber; +import reactor.core.publisher.Mono; import reactor.util.context.Context; -import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.BDDAssertions.then; +import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.scopePassingSpanOperator; /** * @author Marcin Grzejszczak */ -@RunWith(MockitoJUnitRunner.class) public class ScopePassingSpanSubscriberTests { - CurrentTraceContext currentTraceContext = CurrentTraceContext.Default.create(); + static { + // AssertJ will recognise QueueSubscription implements queue and try to invoke + // iterator. That's not allowed, and will cause an exception + // Fuseable$QueueSubscription.NOT_SUPPORTED_MESSAGE. + // This ensures AssertJ uses normal toString. + StandardRepresentation.registerFormatterForType(ScopePassingSpanSubscriber.class, + Objects::toString); + } + + final CurrentTraceContext currentTraceContext = CurrentTraceContext.Default.create(); TraceContext context = TraceContext.newBuilder().traceId(1).spanId(1).sampled(true) .build(); + TraceContext context2 = TraceContext.newBuilder().traceId(1).spanId(2).sampled(true) + .build(); + + AnnotationConfigApplicationContext springContext = new AnnotationConfigApplicationContext(); + + @After + public void close() { + springContext.close(); + } + @Test public void should_propagate_current_context() { ScopePassingSpanSubscriber subscriber = new ScopePassingSpanSubscriber<>(null, @@ -53,28 +80,96 @@ public class ScopePassingSpanSubscriberTests { @Test public void should_set_empty_context_when_context_is_null() { ScopePassingSpanSubscriber subscriber = new ScopePassingSpanSubscriber<>(null, - null, this.currentTraceContext, null); + Context.empty(), this.currentTraceContext, null); then(subscriber.currentContext().isEmpty()).isTrue(); } @Test public void should_put_current_span_to_context() { - try (Scope ws = this.currentTraceContext.newScope(context)) { - CoreSubscriber subscriber = ReactorSleuth.scopePassingSpanSubscription( - beanFactory(), new BaseSubscriber() { - }); + try (Scope ws = this.currentTraceContext.newScope(context2)) { + CoreSubscriber subscriber = new ScopePassingSpanSubscriber<>( + new BaseSubscriber() { + }, Context.empty(), currentTraceContext, context); then(subscriber.currentContext().get(TraceContext.class)).isEqualTo(context); } - } - private BeanFactory beanFactory() { - BeanFactory beanFactory = BDDMockito.mock(BeanFactory.class); - BDDMockito.given(beanFactory.getBean(CurrentTraceContext.class)) - .willReturn(this.currentTraceContext); - return beanFactory; + @Test + public void should_not_trace_scalar_flows() { + springContext.registerBean(CurrentTraceContext.class, () -> currentTraceContext); + springContext.refresh(); + + Function, ? extends Publisher> transformer = scopePassingSpanOperator( + this.springContext); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Subscriber assertNoSpanSubscriber = new CoreSubscriber() { + @Override + public void onSubscribe(Subscription s) { + s.request(Long.MAX_VALUE); + assertThat(s).isNotInstanceOf(ScopePassingSpanSubscriber.class); + } + + @Override + public void onNext(Object o) { + + } + + @Override + public void onError(Throwable t) { + + } + + @Override + public void onComplete() { + + } + }; + + Subscriber assertSpanSubscriber = new CoreSubscriber() { + @Override + public void onSubscribe(Subscription s) { + s.request(Long.MAX_VALUE); + assertThat(s).isInstanceOf(ScopePassingSpanSubscriber.class); + } + + @Override + public void onNext(Object o) { + + } + + @Override + public void onError(Throwable t) { + + } + + @Override + public void onComplete() { + + } + }; + transformer.apply(Mono.just(1).hide()).subscribe(assertSpanSubscriber); + + transformer.apply(Mono.just(1)).subscribe(assertNoSpanSubscriber); + + transformer.apply(Mono.error(new Exception()).hide()) + .subscribe(assertSpanSubscriber); + + transformer.apply(Mono.error(new Exception())) + .subscribe(assertNoSpanSubscriber); + + transformer.apply(Mono.empty().hide()) + .subscribe(assertSpanSubscriber); + + transformer.apply(Mono.empty()).subscribe(assertNoSpanSubscriber); + + } + + Awaitility.await().untilAsserted(() -> { + then(this.currentTraceContext.get()).isNull(); + }); } } diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java deleted file mode 100644 index 9a8278f6c..000000000 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java +++ /dev/null @@ -1,333 +0,0 @@ -/* - * Copyright 2013-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.sleuth.instrument.reactor; - -import java.time.Duration; -import java.util.concurrent.atomic.AtomicReference; -import java.util.function.Function; - -import brave.Span; -import brave.Tracer; -import brave.sampler.Sampler; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.awaitility.Awaitility; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; -import reactor.core.CoreSubscriber; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.core.scheduler.Schedulers; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.test.context.junit4.SpringRunner; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.BDDAssertions.then; - -@RunWith(SpringRunner.class) -@SpringBootTest(classes = SpanSubscriberTests.Config.class, - webEnvironment = SpringBootTest.WebEnvironment.NONE) -public class SpanSubscriberTests { - - private static final Log log = LogFactory.getLog(SpanSubscriberTests.class); - - @Autowired - Tracer tracer; - - @Autowired - ConfigurableApplicationContext factory; - - @Test - public void should_pass_tracing_info_when_using_reactor() { - Span span = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - Publisher traced = Flux.just(1, 2, 3); - log.info("Hello"); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - Flux.from(traced).map(d -> d + 1).map(d -> d + 1).map((d) -> { - spanInOperation.set(this.tracer.currentSpan()); - return d + 1; - }).map(d -> d + 1).subscribe(System.out::println); - } - finally { - span.finish(); - } - - then(this.tracer.currentSpan()).isNull(); - then(spanInOperation.get().context().spanId()).isEqualTo(span.context().spanId()); - } - - @Test - public void should_support_reactor_fusion_optimization() { - Span span = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - log.info("Hello"); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - Mono.just(1).flatMap(d -> Flux.just(d + 1).collectList().map(p -> p.get(0))) - .map(d -> d + 1).map((d) -> { - spanInOperation.set(this.tracer.currentSpan()); - return d + 1; - }).map(d -> d + 1).subscribe(System.out::println); - } - finally { - span.finish(); - } - - then(this.tracer.currentSpan()).isNull(); - then(spanInOperation.get().context().spanId()).isEqualTo(span.context().spanId()); - } - - @Test - public void should_not_trace_scalar_flows() { - Span span = this.tracer.nextSpan().name("foo").start(); - log.info("Hello"); - - // Disable global hooks for local hook testing - TraceReactorAutoConfigurationAccessorConfiguration.close(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - - Function, ? extends Publisher> transformer = ReactorSleuth - .scopePassingSpanOperator(this.factory); - - Subscriber assertNoSpanSubscriber = new CoreSubscriber() { - @Override - public void onSubscribe(Subscription s) { - s.request(Long.MAX_VALUE); - assertThat(s).isNotInstanceOf(ScopePassingSpanSubscriber.class); - } - - @Override - public void onNext(Object o) { - - } - - @Override - public void onError(Throwable t) { - - } - - @Override - public void onComplete() { - - } - }; - - Subscriber assertSpanSubscriber = new CoreSubscriber() { - @Override - public void onSubscribe(Subscription s) { - s.request(Long.MAX_VALUE); - assertThat(s).isInstanceOf(ScopePassingSpanSubscriber.class); - } - - @Override - public void onNext(Object o) { - - } - - @Override - public void onError(Throwable t) { - - } - - @Override - public void onComplete() { - - } - }; - transformer.apply(Mono.just(1).hide()).subscribe(assertSpanSubscriber); - - transformer.apply(Mono.just(1)).subscribe(assertNoSpanSubscriber); - - transformer.apply(Mono.error(new Exception()).hide()) - .subscribe(assertSpanSubscriber); - - transformer.apply(Mono.error(new Exception())) - .subscribe(assertNoSpanSubscriber); - - transformer.apply(Mono.empty().hide()) - .subscribe(assertSpanSubscriber); - - transformer.apply(Mono.empty()).subscribe(assertNoSpanSubscriber); - - } - finally { - span.finish(); - } - - Awaitility.await().untilAsserted(() -> { - then(this.tracer.currentSpan()).isNull(); - }); - - TraceReactorAutoConfigurationAccessorConfiguration.setup(this.factory); - } - - @Test - public void should_pass_tracing_info_when_using_reactor_async() { - Span span = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - log.info("Hello"); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.1") - .map(d -> d + 1).map(d -> d + 1) - .publishOn(Schedulers.newSingle("secondThread")).log("reactor.2") - .map((d) -> { - spanInOperation.set(this.tracer.currentSpan()); - return d + 1; - }).map(d -> d + 1).blockLast(); - - Awaitility.await().untilAsserted(() -> { - then(spanInOperation.get().context().traceId()) - .isEqualTo(span.context().traceId()); - }); - then(this.tracer.currentSpan()).isEqualTo(span); - } - finally { - span.finish(); - } - - then(this.tracer.currentSpan()).isNull(); - Span foo2 = this.tracer.nextSpan().name("foo").start(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(foo2)) { - Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.") - .map(d -> d + 1).map(d -> d + 1).map((d) -> { - spanInOperation.set(this.tracer.currentSpan()); - return d + 1; - }).map(d -> d + 1).blockLast(); - - then(this.tracer.currentSpan()).isEqualTo(foo2); - // parent cause there's an async span in the meantime - then(spanInOperation.get().context().traceId()) - .isEqualTo(foo2.context().traceId()); - } - finally { - foo2.finish(); - } - - then(this.tracer.currentSpan()).isNull(); - } - - @Test - public void checkSequenceOfOperations() { - Span parentSpan = this.tracer.nextSpan().name("foo").start(); - log.info("Hello"); - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(parentSpan)) { - final Long spanId = Mono.fromCallable(this.tracer::currentSpan) - .map(span -> span.context().spanId()).block(); - then(spanId).isNotNull(); - - final Long secondSpanId = Mono.fromCallable(this.tracer::currentSpan) - .map(span -> span.context().spanId()).block(); - then(secondSpanId).isEqualTo(spanId); // different trace ids here - } - } - - @Test - public void checkTraceIdDuringZipOperation() { - Span initSpan = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - final AtomicReference spanInZipOperation = new AtomicReference<>(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) { - Mono.fromCallable(this.tracer::currentSpan) - .map(span -> span.context().spanId()).doOnNext(spanInOperation::set) - .zipWith(Mono.fromCallable(this.tracer::currentSpan) - .map(span -> span.context().spanId()) - .doOnNext(spanInZipOperation::set)) - .block(); - } - - then(spanInZipOperation).hasValue(initSpan.context().spanId()); // ok here - then(spanInOperation).hasValue(initSpan.context().spanId()); // Expecting - // - // to have value: - // <1L> but did - // not. - } - - // #646 - @Test - public void should_work_for_mono_just_with_flat_map() { - Span initSpan = this.tracer.nextSpan().name("foo").start(); - log.info("Hello"); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) { - Mono.just("value1") - .flatMap(request -> Mono.just("value2").then(Mono.just("foo"))) - .map(a -> "qwe").block(); - } - } - - // #1030 - @Test - public void checkTraceIdFromSubscriberContext() { - Span initSpan = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInSubscriberContext = new AtomicReference<>(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) { - Mono.subscriberContext() - .map(context -> this.tracer.currentSpan().context().spanId()) - .doOnNext(spanInSubscriberContext::set).block(); - } - - then(spanInSubscriberContext).hasValue(initSpan.context().spanId()); // ok here - } - - @Test - public void should_pass_tracing_info_into_inner_publishers() { - Span span = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - Flux.range(0, 5) - .flatMap(it -> Mono.delay(Duration.ofMillis(1)) - .map(context -> this.tracer.currentSpan()) - .doOnNext(spanInOperation::set)) - .blockFirst(); - } - finally { - span.finish(); - } - - then(spanInOperation.get().context().spanId()).isEqualTo(span.context().spanId()); - } - - @EnableAutoConfiguration - @Configuration - static class Config { - - @Bean - Sampler sampler() { - return Sampler.ALWAYS_SAMPLE; - } - - } - -} diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java index 7cbf973f6..763574221 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfigurationAccessorConfiguration.java @@ -51,9 +51,7 @@ public final class TraceReactorAutoConfigurationAccessorConfiguration { if (log.isTraceEnabled()) { log.trace("Setting up hooks"); } - TraceReactorAutoConfiguration.TraceReactorConfiguration - .traceHookRegisteringBeanDefinitionRegistryPostProcessor(context) - .setupHooks(context); + HookRegisteringBeanDefinitionRegistryPostProcessor.setupHooks(context); } }