diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/hystrix/SleuthHystrixConcurrencyStrategy.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/hystrix/SleuthHystrixConcurrencyStrategy.java index 89f555a51..b8753ef8f 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/hystrix/SleuthHystrixConcurrencyStrategy.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/hystrix/SleuthHystrixConcurrencyStrategy.java @@ -1,20 +1,26 @@ package org.springframework.cloud.sleuth.instrument.hystrix; -import java.util.concurrent.Callable; - -import org.springframework.cloud.sleuth.TraceManager; -import org.springframework.cloud.sleuth.instrument.TraceCallable; - import com.netflix.hystrix.strategy.HystrixPlugins; import com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategy; +import lombok.extern.slf4j.Slf4j; +import org.springframework.cloud.sleuth.TraceManager; +import org.springframework.cloud.sleuth.instrument.TraceCallable; +import java.util.concurrent.Callable; + +@Slf4j public class SleuthHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy { private final TraceManager traceManager; public SleuthHystrixConcurrencyStrategy(TraceManager traceManager) { this.traceManager = traceManager; - HystrixPlugins.getInstance().registerConcurrencyStrategy(this); + try { + HystrixPlugins.getInstance().registerConcurrencyStrategy(this); + } catch (Exception e) { + HystrixConcurrencyStrategy concurrencyStrategy = HystrixPlugins.getInstance().getConcurrencyStrategy(); + log.warn("Failed to register Sleuth Hystrix Concurrency Strategy. Will use the current one which is [" + concurrencyStrategy + "]", e); + } } @Override diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/AbstractTraceChannelInterceptor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/AbstractTraceChannelInterceptor.java new file mode 100644 index 000000000..16c576774 --- /dev/null +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/AbstractTraceChannelInterceptor.java @@ -0,0 +1,79 @@ +package org.springframework.cloud.sleuth.instrument.integration; + +import org.springframework.cloud.sleuth.MilliSpan; +import org.springframework.cloud.sleuth.Span; +import org.springframework.cloud.sleuth.Trace; +import org.springframework.cloud.sleuth.TraceManager; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.context.IntegrationObjectSupport; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.ChannelInterceptorAdapter; +import org.springframework.util.StringUtils; + +/** + * Abstraction over classes related to channel intercepting + * + * @author Marcin Grzejszczak + */ +abstract class AbstractTraceChannelInterceptor extends ChannelInterceptorAdapter { + + protected final TraceManager traceManager; + + protected AbstractTraceChannelInterceptor(TraceManager traceManager) { + this.traceManager = traceManager; + } + + /** + * Returns a span given the message and a channel. Returns null when there was no + * trace id passed initially. + */ + Span buildSpan(Message message) { + String spanId = getHeader(message, Trace.SPAN_ID_NAME); + String traceId = getHeader(message, Trace.TRACE_ID_NAME); + if (StringUtils.hasText(traceId)) { + MilliSpan.MilliSpanBuilder span = MilliSpan.builder().traceId(traceId).spanId(spanId); + String parentId = getHeader(message, Trace.PARENT_ID_NAME); + if (message.getHeaders().containsKey(Trace.NOT_SAMPLED_NAME)) { + span.exportable(false); + } + String processId = getHeader(message, Trace.PROCESS_ID_NAME); + String spanName = getHeader(message, Trace.SPAN_NAME_NAME); + if (spanName != null) { + span.name(spanName); + } + if (processId != null) { + span.processId(processId); + } + if (parentId != null) { + span.parent(parentId); + } + span.remote(true); + return span.build(); + } + return null; + } + + String getHeader(Message message, String name) { + return (String) message.getHeaders().get(name); + } + + String getChannelName(MessageChannel channel) { + String name = null; + if (channel instanceof IntegrationObjectSupport) { + name = ((IntegrationObjectSupport) channel).getComponentName(); + } + if (name == null && channel instanceof AbstractMessageChannel) { + name = ((AbstractMessageChannel) channel).getFullChannelName(); + } + if (name == null) { + name = channel.toString(); + } + return name; + } + + String getMessageChannelName(MessageChannel channel) { + return "message/" + getChannelName(channel); + } + +} diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/SpanMessageHeaders.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/SpanMessageHeaders.java index 1495c75d7..33c0a9e6f 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/SpanMessageHeaders.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/SpanMessageHeaders.java @@ -16,15 +16,15 @@ package org.springframework.cloud.sleuth.instrument.integration; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - import org.springframework.cloud.sleuth.Span; import org.springframework.cloud.sleuth.Trace; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + /** * Utility for manipulating message headers related to span data. * @@ -68,8 +68,10 @@ public class SpanMessageHeaders { span.addAnnotation(key, value); } } + addPayloadAnnotations(message.getPayload(), span); + } - Object payload = message.getPayload(); + static void addPayloadAnnotations(Object payload, Span span) { if (payload != null) { span.addAnnotation("/messaging/payload/type", payload.getClass().getCanonicalName()); diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceChannelInterceptor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceChannelInterceptor.java index a021f2b2e..5abd786dd 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceChannelInterceptor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceChannelInterceptor.java @@ -16,30 +16,23 @@ package org.springframework.cloud.sleuth.instrument.integration; -import org.springframework.cloud.sleuth.MilliSpan; -import org.springframework.cloud.sleuth.MilliSpan.MilliSpanBuilder; +import org.springframework.cloud.sleuth.Span; import org.springframework.cloud.sleuth.Trace; import org.springframework.cloud.sleuth.TraceManager; import org.springframework.cloud.sleuth.sampler.IsTracingSampler; -import org.springframework.integration.channel.AbstractMessageChannel; -import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.support.ChannelInterceptorAdapter; -import org.springframework.util.StringUtils; /** * @author Dave Syer * */ -public class TraceChannelInterceptor extends ChannelInterceptorAdapter { +public class TraceChannelInterceptor extends AbstractTraceChannelInterceptor { private ThreadLocal traceHolder = new ThreadLocal<>(); - private final TraceManager traceManager; - public TraceChannelInterceptor(TraceManager traceManager) { - this.traceManager = traceManager; + super(traceManager); } @Override @@ -56,59 +49,20 @@ public class TraceChannelInterceptor extends ChannelInterceptorAdapter { return SpanMessageHeaders.addSpanHeaders(message, this.traceManager.getCurrentSpan()); } - String spanId = getHeader(message, Trace.SPAN_ID_NAME); - String traceId = getHeader(message, Trace.TRACE_ID_NAME); - String name = "message/" + getChannelName(channel); - Trace trace; - if (StringUtils.hasText(traceId)) { - - MilliSpanBuilder span = MilliSpan.builder().traceId(traceId).spanId(spanId); - String parentId = getHeader(message, Trace.PARENT_ID_NAME); - if (message.getHeaders().containsKey(Trace.NOT_SAMPLED_NAME)) { - span.exportable(false); - } - String processId = getHeader(message, Trace.PROCESS_ID_NAME); - String spanName = getHeader(message, Trace.SPAN_NAME_NAME); - if (spanName != null) { - span.name(spanName); - } - if (processId != null) { - span.processId(processId); - } - if (parentId != null) { - span.parent(parentId); - } - span.remote(true); - - trace = this.traceManager.startSpan(name, span.build()); - } - else { - if (message.getHeaders().containsKey(Trace.NOT_SAMPLED_NAME)) { - trace = this.traceManager.startSpan(name, IsTracingSampler.INSTANCE, null); - } else { - trace = this.traceManager.startSpan(name); - } - } + String name = getMessageChannelName(channel); + Trace trace = startSpan(buildSpan(message), name, message); this.traceHolder.set(trace); return SpanMessageHeaders.addSpanHeaders(message, trace.getSpan()); } - private String getChannelName(MessageChannel channel) { - String name = null; - if (channel instanceof IntegrationObjectSupport) { - name = ((IntegrationObjectSupport) channel).getComponentName(); + private Trace startSpan(Span span, String name, Message message) { + if (span != null) { + return traceManager.startSpan(name, span); } - if (name == null && channel instanceof AbstractMessageChannel) { - name = ((AbstractMessageChannel) channel).getFullChannelName(); + if (message.getHeaders().containsKey(Trace.NOT_SAMPLED_NAME)) { + return traceManager.startSpan(name, IsTracingSampler.INSTANCE, null); } - if (name == null) { - name = channel.toString(); - } - return name; - } - - private String getHeader(Message message, String name) { - return (String) message.getHeaders().get(name); + return this.traceManager.startSpan(name); } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceStompMessageChannelInterceptor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceStompMessageChannelInterceptor.java index 154cf93fe..52d59c05b 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceStompMessageChannelInterceptor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceStompMessageChannelInterceptor.java @@ -15,71 +15,45 @@ */ package org.springframework.cloud.sleuth.instrument.integration; -import static org.springframework.util.StringUtils.hasText; - -import org.springframework.cloud.sleuth.MilliSpan; -import org.springframework.cloud.sleuth.MilliSpan.MilliSpanBuilder; +import org.springframework.cloud.sleuth.Span; import org.springframework.cloud.sleuth.Trace; import org.springframework.cloud.sleuth.TraceManager; -import org.springframework.integration.channel.AbstractMessageChannel; -import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.ChannelInterceptor; -import org.springframework.messaging.support.ChannelInterceptorAdapter; /** * Interceptor for Stomp Messages sent over websocket * * @author Gaurav Rai Mazra + * @author Marcin Grzejszczak * */ -public class TraceStompMessageChannelInterceptor extends ChannelInterceptorAdapter implements ChannelInterceptor { +public class TraceStompMessageChannelInterceptor extends AbstractTraceChannelInterceptor implements ChannelInterceptor { private ThreadLocal traceScopeHolder = new ThreadLocal(); - private final TraceManager traceManager; public TraceStompMessageChannelInterceptor(final TraceManager traceManager) { - this.traceManager = traceManager; + super(traceManager); } @Override public Message preSend(Message message, MessageChannel channel) { - final TraceManager traceManager = this.traceManager; - if (traceManager.isTracing() || message.getHeaders().containsKey(Trace.NOT_SAMPLED_NAME)) { return StompMessageBuilder.fromMessage(message).setHeadersFromSpan(traceManager.getCurrentSpan()).build(); } - - String spanId = getHeader(message, Trace.SPAN_ID_NAME); - String traceId = getHeader(message, Trace.TRACE_ID_NAME); - String name = "message/" + getChannelName(channel); - Trace trace; - if (hasText(spanId) && hasText(traceId)) { - - MilliSpanBuilder span = MilliSpan.builder().traceId(traceId).spanId(spanId); - String parentId = getHeader(message, Trace.PARENT_ID_NAME); - String processId = getHeader(message, Trace.PROCESS_ID_NAME); - String spanName = getHeader(message, Trace.SPAN_NAME_NAME); - if (spanName != null) { - span.name(spanName); - } - if (processId != null) { - span.processId(processId); - } - if (parentId != null) { - span.parent(parentId); - } - span.remote(true); - - // TODO: traceManager description? - trace = traceManager.startSpan(name, span.build()); - } else { - trace = traceManager.startSpan(name); - } + String name = getMessageChannelName(channel); + Trace trace = startSpan(buildSpan(message), name); this.traceScopeHolder.set(trace); return StompMessageBuilder.fromMessage(message).setHeadersFromSpan(trace.getSpan()).build(); } + private Trace startSpan(Span span, String name) { + if (span != null) { + return traceManager.startSpan(name, span); + } + return traceManager.startSpan(name); + } + @Override public void postSend(Message message, MessageChannel channel, boolean sent) { final ThreadLocal traceScopeHolder = this.traceScopeHolder; @@ -87,22 +61,4 @@ public class TraceStompMessageChannelInterceptor extends ChannelInterceptorAdapt this.traceManager.close(traceInScope); traceScopeHolder.remove(); } - - private String getChannelName(MessageChannel channel) { - String name = null; - if (channel instanceof IntegrationObjectSupport) { - name = ((IntegrationObjectSupport) channel).getComponentName(); - } - if (name == null && channel instanceof AbstractMessageChannel) { - name = ((AbstractMessageChannel) channel).getFullChannelName(); - } - if (name == null) { - name = channel.toString(); - } - return name; - } - - private String getHeader(Message message, String name) { - return (String) message.getHeaders().get(name); - } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceStompMessageContextPropagationChannelInterceptor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceStompMessageContextPropagationChannelInterceptor.java index d13753884..387c2981b 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceStompMessageContextPropagationChannelInterceptor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceStompMessageContextPropagationChannelInterceptor.java @@ -34,6 +34,7 @@ import java.util.Map; /** * * @author Gaurav Rai Mazra + * @author Marcin Grzejszczak * */ public class TraceStompMessageContextPropagationChannelInterceptor extends ChannelInterceptorAdapter @@ -63,9 +64,7 @@ public class TraceStompMessageContextPropagationChannelInterceptor extends Chann public final Message postReceive(Message message, MessageChannel channel) { if (message instanceof MessageWithSpan) { MessageWithSpan messageWithSpan = (MessageWithSpan) message; - Message messageToHandle = messageWithSpan.message; populatePropagatedContext(messageWithSpan.span); - return message; } return message; @@ -114,15 +113,7 @@ public class TraceStompMessageContextPropagationChannelInterceptor extends Chann span.addAnnotation(key, value); } } - Object payload = message.getPayload(); - if (payload != null) { - span.addAnnotation("/messaging/payload/type", payload.getClass().getCanonicalName()); - if (payload instanceof String) { - span.addAnnotation("/messaging/payload/size", String.valueOf(((String) payload).length())); - } else if (payload instanceof byte[]) { - span.addAnnotation("/messaging/payload/size", String.valueOf(((byte[]) payload).length)); - } - } + SpanMessageHeaders.addPayloadAnnotations(message.getPayload(), span); } @Override