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

TracingChannelInterceptor: simple refactoring (#1948)

Browse Source 
* 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
This commit is contained in:
Artem Bilan
2021-05-25 09:42:53 -04:00
committed by GitHub
parent 30b17f7576
commit fa1e0b87b5
1 changed files with 105 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<MessageHeaderAccessor> injector;
final Propagator.Getter<MessageHeaderAccessor> 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<MessageHeaderAccessor> injector;
private final Propagator.Getter<MessageHeaderAccessor> extractor;
private final MessageSpanCustomizer messageSpanCustomizer;
private final Propagator propagator;
private final Function<String, String> remoteServiceNameMapper;
private ApplicationContext applicationContext;
public TracingChannelInterceptor(Tracer tracer, Propagator propagator,
Propagator.Setter<MessageHeaderAccessor> setter, Propagator.Getter<MessageHeaderAccessor> getter,
Function<String, String> 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<String, String> 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<Map.Entry<String, BinderType>> 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<String> 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<SpanAndScope> threadLocalSpan = new ThreadLocal<>();
final LinkedBlockingDeque<SpanAndScope> 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<SpanAndScope> threadLocalSpan = new ThreadLocal<>();
private final LinkedBlockingDeque<SpanAndScope> 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");
}
}
}
}