diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java index 9640fdb2a..25ed6e3b9 100644 --- a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java +++ b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java @@ -16,31 +16,24 @@ package org.springframework.cloud.sleuth.instrument.messaging; -import java.util.Iterator; -import java.util.Map; import java.util.NoSuchElementException; +import java.util.Set; import java.util.concurrent.LinkedBlockingDeque; import java.util.function.Function; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; import org.springframework.cloud.sleuth.Span; import org.springframework.cloud.sleuth.Tracer; import org.springframework.cloud.sleuth.propagation.Propagator; -import org.springframework.cloud.stream.binder.BinderType; -import org.springframework.cloud.stream.binder.BinderTypeRegistry; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; -import org.springframework.integration.channel.DirectChannel; +import org.springframework.core.log.LogAccessor; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; -import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.ExecutorChannelInterceptor; import org.springframework.messaging.support.GenericMessage; @@ -55,18 +48,17 @@ import org.springframework.util.StringUtils; * a handler later calls {@link MessageHandler#handleMessage(Message)}. * * @author Marcin Grzejszczak + * @author Artem Bilan * @since 3.0.0 */ -public final class TracingChannelInterceptor extends ChannelInterceptorAdapter - implements ExecutorChannelInterceptor, ApplicationContextAware { +public final class TracingChannelInterceptor implements ExecutorChannelInterceptor, ApplicationContextAware { /** * Name of the class in Spring Cloud Stream that is a direct channel. */ - public static final String STREAM_DIRECT_CHANNEL = "org.springframework." - + "cloud.stream.messaging.DirectWithAttributesChannel"; + public static final String STREAM_DIRECT_CHANNEL = "org.springframework.cloud.stream.messaging.DirectWithAttributesChannel"; - private static final Log log = LogFactory.getLog(TracingChannelInterceptor.class); + private static final LogAccessor log = new LogAccessor(TracingChannelInterceptor.class); /** * Using the literal "broker" until we come up with a better solution. @@ -88,44 +80,47 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter */ private static final String REMOTE_SERVICE_NAME = "broker"; - final Tracer tracer; + private static final boolean hasDirectChannelClass = ClassUtils + .isPresent("org.springframework.integration.channel.DirectChannel", null); - final Propagator.Setter injector; - - final Propagator.Getter extractor; - - final MessageSpanCustomizer messageSpanCustomizer; - - private final boolean hasDirectChannelClass; - - private final boolean hasBinderTypeRegistry; + private static final boolean hasBinderTypeRegistry = ClassUtils + .isPresent("org.springframework.cloud.stream.binder.BinderTypeRegistry", null); // special case of a Stream - private final Class directWithAttributesChannelClass; - - private ApplicationContext applicationContext; - - private final Propagator propagator; + private static final Class directWithAttributesChannelClass = ClassUtils.isPresent(STREAM_DIRECT_CHANNEL, null) + ? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null; private final ThreadLocalSpan threadLocalSpan = new ThreadLocalSpan(); + private final Tracer tracer; + + private final Propagator.Setter injector; + + private final Propagator.Getter extractor; + + private final MessageSpanCustomizer messageSpanCustomizer; + + private final Propagator propagator; + private final Function remoteServiceNameMapper; + private ApplicationContext applicationContext; + public TracingChannelInterceptor(Tracer tracer, Propagator propagator, Propagator.Setter setter, Propagator.Getter getter, Function remoteServiceNameMapper, MessageSpanCustomizer messageSpanCustomizer) { + this.tracer = tracer; this.propagator = propagator; this.injector = setter; this.extractor = getter; this.remoteServiceNameMapper = remoteServiceNameMapper; this.messageSpanCustomizer = messageSpanCustomizer; - this.hasDirectChannelClass = ClassUtils.isPresent("org.springframework.integration.channel.DirectChannel", - null); - this.hasBinderTypeRegistry = ClassUtils.isPresent("org.springframework.cloud.stream.binder.BinderTypeRegistry", - null); - this.directWithAttributesChannelClass = ClassUtils.isPresent(STREAM_DIRECT_CHANNEL, null) - ? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; } /** @@ -133,28 +128,19 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter */ @Override public Message preSend(Message message, MessageChannel channel) { - if (emptyMessage(message)) { - return message; - } Message retrievedMessage = getMessage(message); - if (log.isDebugEnabled()) { - log.debug("Received a message in pre-send " + retrievedMessage); - } + log.debug(() -> "Received a message in pre-send " + retrievedMessage); MessageHeaderAccessor headers = mutableHeaderAccessor(retrievedMessage); Span.Builder spanBuilder = this.propagator.extract(headers, this.extractor); MessageHeaderPropagatorSetter.removeAnyTraceHeaders(headers, this.propagator.fields()); spanBuilder = spanBuilder.kind(Span.Kind.PRODUCER); spanBuilder = this.messageSpanCustomizer.customizeSend(spanBuilder, message, channel) - .remoteServiceName(toRemoteServiceName(headers)); + .remoteServiceName(toRemoteServiceName(headers, remoteServiceNameMapper, applicationContext)); Span span = spanBuilder.start(); - if (log.isDebugEnabled()) { - log.debug("Extracted result from headers " + span); - } + log.debug(() -> "Extracted result from headers " + span); setSpanInScope(span); this.propagator.inject(span.context(), headers, this.injector); - if (log.isDebugEnabled()) { - log.debug("Created a new span in pre send " + span); - } + log.debug(() -> "Created a new span in pre send " + span); Message outputMessage = outputMessage(message, retrievedMessage, headers); if (isDirectChannel(channel)) { beforeHandle(outputMessage, channel, null); @@ -165,24 +151,25 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter private void setSpanInScope(Span span) { Tracer.SpanInScope spanInScope = this.tracer.withSpan(span); this.threadLocalSpan.set(new SpanAndScope(span, spanInScope)); - if (log.isDebugEnabled()) { - log.debug("Put span in scope " + span); - } + log.debug(() -> "Put span in scope " + span); } - private String toRemoteServiceName(MessageHeaderAccessor headers) { + private static String toRemoteServiceName(MessageHeaderAccessor headers, + Function remoteServiceNameMapper, ApplicationContext applicationContext) { + for (String key : headers.getMessageHeaders().keySet()) { - String remoteServiceName = this.remoteServiceNameMapper.apply(key); + String remoteServiceName = remoteServiceNameMapper.apply(key); if (StringUtils.hasText(remoteServiceName)) { return remoteServiceName; } } - if (this.hasBinderTypeRegistry && this.applicationContext != null) { - BinderTypeRegistry typeRegistry = this.applicationContext.getBean(BinderTypeRegistry.class); - Iterator> iterator = typeRegistry.getAll().entrySet().iterator(); - if (iterator.hasNext()) { - String binderName = iterator.next().getKey(); - String remoteServiceName = this.remoteServiceNameMapper.apply(binderName); + + if (hasBinderTypeRegistry && applicationContext != null) { + org.springframework.cloud.stream.binder.BinderTypeRegistry typeRegistry = applicationContext + .getBean(org.springframework.cloud.stream.binder.BinderTypeRegistry.class); + Set binderNames = typeRegistry.getAll().keySet(); + for (String binderName : binderNames) { + String remoteServiceName = remoteServiceNameMapper.apply(binderName); if (StringUtils.hasText(remoteServiceName)) { return remoteServiceName; } @@ -201,43 +188,29 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter return new ErrorMessage(errorMessage.getPayload(), isWebSockets(headers) ? headers.getMessageHeaders() : new MessageHeaders(headers.getMessageHeaders()), errorMessage.getOriginalMessage()); } - headers.copyHeaders(new MessageHeaders(additionalHeaders.getMessageHeaders())); + headers.copyHeaders(additionalHeaders.getMessageHeaders()); return new GenericMessage<>(retrievedMessage.getPayload(), isWebSockets(headers) ? headers.getMessageHeaders() : new MessageHeaders(headers.getMessageHeaders())); } - private boolean isWebSockets(MessageHeaderAccessor headerAccessor) { + private static boolean isWebSockets(MessageHeaderAccessor headerAccessor) { return headerAccessor.getMessageHeaders().containsKey("stompCommand") || headerAccessor.getMessageHeaders().containsKey("simpMessageType"); } - private boolean isDirectChannel(MessageChannel channel) { + private static boolean isDirectChannel(MessageChannel channel) { Class targetClass = AopUtils.getTargetClass(channel); - boolean directChannel = this.hasDirectChannelClass && DirectChannel.class.isAssignableFrom(targetClass); - if (!directChannel) { - return false; - } - if (this.directWithAttributesChannelClass == null) { - return true; - } - return !isStreamSpecialDirectChannel(targetClass); - } - - private boolean isStreamSpecialDirectChannel(Class targetClass) { - return this.directWithAttributesChannelClass.isAssignableFrom(targetClass); + return (directWithAttributesChannelClass == null + || !directWithAttributesChannelClass.isAssignableFrom(targetClass)) && hasDirectChannelClass + && org.springframework.integration.channel.DirectChannel.class.isAssignableFrom(targetClass); } @Override public void afterSendCompletion(Message message, MessageChannel channel, boolean sent, Exception ex) { - if (emptyMessage(message)) { - return; - } if (isDirectChannel(channel)) { 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); } @@ -247,26 +220,15 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter */ @Override public Message postReceive(Message message, MessageChannel channel) { - if (emptyMessage(message)) { - return message; - } MessageHeaderAccessor headers = mutableHeaderAccessor(message); - if (log.isDebugEnabled()) { - log.debug("Received a message in post-receive " + message); - } + log.debug(() -> "Received a message in post-receive " + message); Span result = this.propagator.extract(headers, this.extractor).start(); - if (log.isDebugEnabled()) { - log.debug("Extracted result from headers " + result); - } + log.debug(() -> "Extracted result from headers " + result); Span span = consumerSpanReceive(message, channel, headers, result); setSpanInScope(span); - if (log.isDebugEnabled()) { - log.debug("Created a new span that will be injected in the headers " + span); - } + log.debug(() -> "Created a new span that will be injected in the headers " + span); this.propagator.inject(span.context(), headers, this.injector); - if (log.isDebugEnabled()) { - log.debug("Created a new span in post receive " + span); - } + log.debug(() -> "Created a new span in post receive " + span); headers.setImmutable(); if (message instanceof ErrorMessage) { ErrorMessage errorMessage = (ErrorMessage) message; @@ -282,18 +244,13 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter MessageHeaderPropagatorSetter.removeAnyTraceHeaders(headers, this.propagator.fields()); builder = builder.kind(Span.Kind.CONSUMER); builder = this.messageSpanCustomizer.customizeReceive(builder, message, channel); - builder = builder.remoteServiceName(toRemoteServiceName(headers)); + builder = builder.remoteServiceName(toRemoteServiceName(headers, remoteServiceNameMapper, applicationContext)); return builder.start(); } @Override 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); } @@ -303,13 +260,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter */ @Override public Message beforeHandle(Message message, MessageChannel channel, MessageHandler handler) { - if (emptyMessage(message)) { - return message; - } MessageHeaderAccessor headers = mutableHeaderAccessor(message); - if (log.isDebugEnabled()) { - log.debug("Received a message in before handle " + message); - } + log.debug(() -> "Received a message in before handle " + message); Span consumerSpan = consumerSpan(message, channel, headers); // create and scope a span for the message processor Span handle = this.tracer.nextSpan(consumerSpan); @@ -348,12 +300,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter @Override 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); } @@ -365,9 +312,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter Span span = spanAndScope.span; Tracer.SpanInScope scope = spanAndScope.scope; if (span.isNoop()) { - if (log.isDebugEnabled()) { - log.debug("Span " + span + " is noop - will stope the scope"); - } + log.debug(() -> "Span " + span + " is noop - will stop the scope"); scope.close(); return; } @@ -378,23 +323,19 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter } span.tag("error", message); } - if (log.isDebugEnabled()) { - log.debug("Will finish the and its corresponding scope " + span); - } + log.debug(() -> "Will finish the and its corresponding scope " + span); span.end(); scope.close(); } private SpanAndScope getSpanFromThreadLocal() { SpanAndScope span = this.threadLocalSpan.get(); - if (log.isDebugEnabled()) { - log.debug("Took span [" + span + "] from thread local"); - } + log.debug(() -> "Took span [" + span + "] from thread local"); this.threadLocalSpan.remove(); return span; } - private MessageHeaderAccessor mutableHeaderAccessor(Message message) { + private static MessageHeaderAccessor mutableHeaderAccessor(Message message) { MessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, MessageHeaderAccessor.class); if (accessor != null && accessor.isMutable()) { return accessor; @@ -404,7 +345,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter return headers; } - private Message getMessage(Message message) { + private static Message getMessage(Message message) { Object payload = message.getPayload(); if (payload instanceof MessagingException) { MessagingException e = (MessagingException) payload; @@ -414,67 +355,57 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter return message; } - private boolean emptyMessage(Message message) { - return message == null; - } + private static class SpanAndScope { - @Override - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - this.applicationContext = applicationContext; - } + final Span span; -} + final Tracer.SpanInScope scope; -class SpanAndScope { - - final Span span; - - final Tracer.SpanInScope scope; - - SpanAndScope(Span span, Tracer.SpanInScope scope) { - this.span = span; - this.scope = scope; - } - -} - -class ThreadLocalSpan { - - private static final Log log = LogFactory.getLog(ThreadLocalSpan.class); - - final ThreadLocal threadLocalSpan = new ThreadLocal<>(); - - final LinkedBlockingDeque spans = new LinkedBlockingDeque<>(); - - void set(SpanAndScope spanAndScope) { - SpanAndScope scope = this.threadLocalSpan.get(); - if (scope != null) { - this.spans.addFirst(scope); + SpanAndScope(Span span, Tracer.SpanInScope scope) { + this.span = span; + this.scope = scope; } - this.threadLocalSpan.set(spanAndScope); + } - SpanAndScope get() { - return this.threadLocalSpan.get(); - } + private static class ThreadLocalSpan { - void remove() { - this.threadLocalSpan.remove(); - if (this.spans.isEmpty()) { - return; + private static final LogAccessor log = new LogAccessor(ThreadLocalSpan.class); + + private final ThreadLocal threadLocalSpan = new ThreadLocal<>(); + + private final LinkedBlockingDeque spans = new LinkedBlockingDeque<>(); + + ThreadLocalSpan() { } - try { - SpanAndScope span = this.spans.removeFirst(); - if (log.isDebugEnabled()) { - log.debug("Took span [" + span + "] from thread local"); + + void set(SpanAndScope spanAndScope) { + SpanAndScope scope = this.threadLocalSpan.get(); + if (scope != null) { + this.spans.addFirst(scope); } - this.threadLocalSpan.set(span); + this.threadLocalSpan.set(spanAndScope); } - catch (NoSuchElementException ex) { - if (log.isTraceEnabled()) { - log.trace("Failed to remove a span from the queue", ex); + + SpanAndScope get() { + return this.threadLocalSpan.get(); + } + + void remove() { + this.threadLocalSpan.remove(); + if (this.spans.isEmpty()) { + return; + } + try { + SpanAndScope span = this.spans.removeFirst(); + log.debug(() -> "Took span [" + span + "] from thread local"); + this.threadLocalSpan.set(span); + } + catch (NoSuchElementException ex) { + log.trace(ex, () -> "Failed to remove a span from the queue"); } } + } }