From fa1e0b87b58eb3ec44137563e15269274e198cb1 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 25 May 2021 09:42:53 -0400 Subject: [PATCH] TracingChannelInterceptor: simple refactoring (#1948) * TracingChannelInterceptor: simple refactoring * * More clean up in the `TracingChannelInterceptor` - Move Spring Cloud Stream classes logic into `static` methods to avoid eager load for those classes which are not present on CP - Remove `emptyMessage()` since the contract for `ChannelInterceptor` never accept a `null` - Remove thread state manipulation from the `postReceive()` since no one takes care about thread local afterwards - Remove `afterReceiveCompletion()` in favor of its `default` impl in the interface: we don't do thread local manipulation for `receive()` since `afterReceiveCompletion()` is called *before* the message is really returned to the target subscriber on the channel * * Restore span manipulation for `receive()`. Even if we don't need a thread local store, we still need to call `span.end()` and set an exception tag to it if any. So, just reuse an existing API which includes thread local. Even if we don't need it logically, there might be some other interceptors in between which would like to take a span from thread local do something even on that "void" `receive()` operation --- .../messaging/TracingChannelInterceptor.java | 279 +++++++----------- 1 file changed, 105 insertions(+), 174 deletions(-) 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"); } } + } }