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

Sleuth no longer corrupts Error Message Channel Headers (#765)

Browse Source 
fixes #761
This commit is contained in:
Marcin Grzejszczak
2017-11-01 14:21:41 +01:00
committed by GitHub
parent 3c2c43adff
commit dadb46731e
3 changed files with 55 additions and 1 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/messaging/TraceChannelInterceptor.java
View File

@@ -16,6 +16,9 @@
package org.springframework.cloud.sleuth.instrument.messaging;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.cloud.sleuth.Log;
@@ -27,6 +30,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.support.MessageHeaderAccessor;
@@ -124,10 +128,24 @@ public class TraceChannelInterceptor extends AbstractTraceChannelInterceptor {
}
getSpanInjector().inject(span, new MessagingTextMap(messageBuilder));
MessageHeaderAccessor headers = MessageHeaderAccessor.getMutableAccessor(message);
if (message instanceof ErrorMessage) {
headers.copyHeaders(sleuthHeaders(messageBuilder.build().getHeaders()));
return new ErrorMessage((Throwable) message.getPayload(), headers.getMessageHeaders());
}
headers.copyHeaders(messageBuilder.build().getHeaders());
return new GenericMessage<>(message.getPayload(), headers.getMessageHeaders());
}
private Map<String, ?> sleuthHeaders(Map<String, ?> headers) {
Map<String, Object> headersToCopy = new HashMap<>();
for (Map.Entry<String, ?> entry : headers.entrySet()) {
if (TraceMessageHeaders.ALL_HEADERS.contains(entry.getKey())) {
headersToCopy.put(entry.getKey(), entry.getValue());
}
}
return headersToCopy;
}
private Message getMessage(Message<?> message) {
Object payload = message.getPayload();
if (payload instanceof MessagingException) {

5
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessageHeaders.java
View File

@@ -16,6 +16,9 @@
package org.springframework.cloud.sleuth.instrument.messaging;
import java.util.Arrays;
import java.util.List;
/**
* Contains trace related messaging headers. The deprecated headers contained `-` which
* for example in the JMS specs is invalid. That's why the public constants in this class
@@ -33,6 +36,8 @@ public class TraceMessageHeaders {
public static final String TRACE_ID_NAME = "spanTraceId";
public static final String SPAN_NAME_NAME = "spanName";
public static final String SPAN_FLAGS_NAME = "spanFlags";
static final List<String> ALL_HEADERS = Arrays.asList(SPAN_ID_NAME, SAMPLED_NAME,
PROCESS_ID_NAME, PARENT_ID_NAME, TRACE_ID_NAME, SPAN_NAME_NAME, SPAN_FLAGS_NAME);
static final String MESSAGE_SENT_FROM_CLIENT = "messageSent";
static final String HEADER_DELIMITER = "_";

33
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TraceChannelInterceptorTests.java
View File

@@ -40,15 +40,17 @@ import org.springframework.cloud.sleuth.util.ExceptionUtils;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.test.annotation.DirtiesContext;
@@ -342,6 +344,35 @@ public class TraceChannelInterceptorTests implements MessageHandler {
then(this.accumulator.getSpans()).hasSize(1);
}
@Test
public void errorMessageHeadersRetained() {
QueueChannel deadReplyChannel = new QueueChannel();
QueueChannel errorsReplyChannel = new QueueChannel();
Map<String, Object> errorChannelHeaders = new HashMap<>();
errorChannelHeaders.put(MessageHeaders.REPLY_CHANNEL, errorsReplyChannel);
errorChannelHeaders.put(MessageHeaders.ERROR_CHANNEL, errorsReplyChannel);
this.tracedChannel.send(new ErrorMessage(
new MessagingException(MessageBuilder.withPayload("hi")
.setHeader(TraceMessageHeaders.TRACE_ID_NAME, Span.idToHex(10L))
.setHeader(TraceMessageHeaders.SPAN_ID_NAME, Span.idToHex(20L))
.setReplyChannel(deadReplyChannel)
.setErrorChannel(deadReplyChannel)
.build() ),
errorChannelHeaders));
then(this.message).isNotNull();
String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, String.class);
then(spanId).isNotNull();
long traceId = Span
.hexToId(this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, String.class));
then(traceId).isEqualTo(10L);
then(spanId).isNotEqualTo(20L);
then(this.accumulator.getSpans()).hasSize(1);
then(this.message.getHeaders().getReplyChannel()).isSameAs(errorsReplyChannel);
then(this.message.getHeaders().getErrorChannel()).isSameAs(errorsReplyChannel);
}
@Configuration
@EnableAutoConfiguration
static class App {