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

Refactored Channel Interceptor code to remove duplications

Browse Source
This commit is contained in:
Marcin Grzejszczak
2016-01-04 16:37:03 +01:00
parent 7ed53116fd
commit 5a1f32c92c
6 changed files with 124 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/hystrix/SleuthHystrixConcurrencyStrategy.java
View File

@@ -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

79
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/AbstractTraceChannelInterceptor.java Normal file
View File

@@ -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);
}
}

12
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/SpanMessageHeaders.java
View File

@@ -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());

68
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceChannelInterceptor.java
View File

@@ -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<Trace> 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);
}
}

70
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceStompMessageChannelInterceptor.java
View File

@@ -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<Trace> traceScopeHolder = new ThreadLocal<Trace>();
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<Trace> 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);
}
}

13
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/integration/TraceStompMessageContextPropagationChannelInterceptor.java
View File

@@ -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