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

Fix headers in messaging (#335)

Browse Source 
* Fix headers in messaging

with this change the span related headers are changed to be compatible with messaging specs. The change is backwards compatible so we accept the old ones as input, and as output we're producing both new and old headers.

fixes #332
This commit is contained in:
Marcin Grzejszczak
2016-07-14 11:57:09 +02:00
parent 795b7c8e81
commit 3ac3d41bad
8 changed files with 206 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
circle.yml
View File

@@ -8,10 +8,11 @@ dependencies:
- ./mvnw -s .settings.xml --fail-never dependency:go-offline || true
test:
override:
- ./mvnw -s .settings.xml clean install -nsu --batch-mode -Dmaven.test.redirectTestOutputToFile=true -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn
- ./mvnw -s .settings.xml clean org.jacoco:jacoco-maven-plugin:prepare-agent install -P sonar -nsu --batch-mode -Dmaven.test.redirectTestOutputToFile=true -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn
post:
- mkdir -p $CIRCLE_TEST_REPORTS/junit/
- find . -type f -regex ".*/target/.*-reports/.*" -exec cp {} $CIRCLE_TEST_REPORTS/junit/ \;
- bash <(curl -s https://codecov.io/bash)
notify:
webhooks:

8
docs/src/main/asciidoc/spring-cloud-sleuth.adoc
View File

@@ -449,6 +449,9 @@ We're injecting a `RestTemplate` interceptor that ensures that all the tracing i
call is made a new Span is created. It gets closed upon receiving the response. In order to block the synchronous `RestTemplate` features
just set `spring.sleuth.web.client.enabled` to `false`.
IMPORTANT: You have to register `RestTemplate` as a bean so that the interceptors will get injected.
If you create a `RestTemplate` instance with a `new` keyword then the instrumentation WILL NOT work.
==== Asynchronous Rest Template
Custom instrumentation is set to create and close Spans upon sending and receiving requests. You can customize the `ClientHttpRequestFactory`
@@ -512,6 +515,11 @@ include::../../../../spring-cloud-sleuth-core/src/test/java/org/springframework/
Spring Cloud Sleuth integrates with http://projects.spring.io/spring-integration/[Spring Integration]. It creates spans for publish and
subscribe events. To disable Spring Integration instrumentation, set `spring.sleuth.integration.enabled` to false.
Spring Cloud Sleuth up till version 1.0.4 is sending invalid tracing headers when using messaging. Those headers are actually
the same as the ones sent in HTTP (they contain a `-`) in its name. For the sake of
backwards compatibility in 1.0.4 we've started sending both valid and invalid headers. Please upgrade to 1.0.4 because
in Spring Cloud Sleuth 1.1 we will remove the support for the deprecated headers.
=== Zuul
We're registering Zuul filters to propagate the tracing information (the request header is enriched with tracing data).

2
pom.xml
View File

@@ -195,7 +195,7 @@
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<version>2.1.0</version>
<version>3.5.1</version>
<scope>test</scope>
</dependency>
</dependencies>

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

@@ -16,8 +16,11 @@
package org.springframework.cloud.sleuth.instrument.messaging;
import java.lang.invoke.MethodHandles;
import java.util.Random;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.Span.SpanBuilder;
import org.springframework.cloud.sleuth.SpanExtractor;
@@ -29,7 +32,9 @@ import org.springframework.messaging.Message;
* @author Marcin Grzejszczak
* @since 1.0.0
*/
public class MessagingSpanExtractor implements SpanExtractor<Message<?>> {
class MessagingSpanExtractor implements SpanExtractor<Message<?>> {
private static final Log log = LogFactory.getLog(MethodHandles.lookup().lookupClass());
private final Random random;
@@ -39,28 +44,55 @@ public class MessagingSpanExtractor implements SpanExtractor<Message<?>> {
@Override
public Span joinTrace(Message<?> carrier) {
if (!hasHeader(carrier, Span.TRACE_ID_NAME)
|| !hasHeader(carrier, Span.SPAN_ID_NAME)) {
if ((!hasHeader(carrier, Span.TRACE_ID_NAME)
|| !hasHeader(carrier, Span.SPAN_ID_NAME))
&& (!hasHeader(carrier, TraceMessageHeaders.SPAN_ID_NAME)
|| !hasHeader(carrier, TraceMessageHeaders.TRACE_ID_NAME))) {
return null;
// TODO: Consider throwing IllegalArgumentException;
}
if (hasHeader(carrier, Span.TRACE_ID_NAME)
|| hasHeader(carrier, Span.SPAN_ID_NAME)) {
log.warn("Deprecated trace headers detected. Please upgrade Sleuth to 1.1 "
+ "or start sending headers present in the TraceMessageHeaders class");
return extractSpanFromOldHeaders(carrier, Span.builder());
}
return extractSpanFromNewHeaders(carrier, Span.builder());
}
// Backwards compatibility
private Span extractSpanFromOldHeaders(Message<?> carrier, SpanBuilder spanBuilder) {
return extractSpanFromHeaders(carrier, spanBuilder, Span.TRACE_ID_NAME, Span.SPAN_ID_NAME,
Span.SAMPLED_NAME, Span.PROCESS_ID_NAME, Span.SPAN_NAME_NAME, Span.PARENT_ID_NAME);
}
private Span extractSpanFromNewHeaders(Message<?> carrier, SpanBuilder spanBuilder) {
return extractSpanFromHeaders(carrier, spanBuilder, TraceMessageHeaders.TRACE_ID_NAME,
TraceMessageHeaders.SPAN_ID_NAME, TraceMessageHeaders.SAMPLED_NAME,
TraceMessageHeaders.PROCESS_ID_NAME, TraceMessageHeaders.SPAN_NAME_NAME,
TraceMessageHeaders.PARENT_ID_NAME);
}
private Span extractSpanFromHeaders(Message<?> carrier, SpanBuilder spanBuilder,
String traceIdHeader, String spanIdHeader, String spanSampledHeader,
String spanProcessIdHeader, String spanNameHeader, String spanParentIdHeader) {
long traceId = Span
.hexToId(getHeader(carrier, Span.TRACE_ID_NAME));
long spanId = hasHeader(carrier, Span.SPAN_ID_NAME)
? Span.hexToId(getHeader(carrier, Span.SPAN_ID_NAME))
.hexToId(getHeader(carrier, traceIdHeader));
long spanId = hasHeader(carrier, spanIdHeader)
? Span.hexToId(getHeader(carrier, spanIdHeader))
: this.random.nextLong();
SpanBuilder spanBuilder = Span.builder().traceId(traceId).spanId(spanId);
spanBuilder = spanBuilder.traceId(traceId).spanId(spanId);
spanBuilder.exportable(
Span.SPAN_SAMPLED.equals(getHeader(carrier, Span.SAMPLED_NAME)));
String processId = getHeader(carrier, Span.PROCESS_ID_NAME);
String spanName = getHeader(carrier, Span.SPAN_NAME_NAME);
Span.SPAN_SAMPLED.equals(getHeader(carrier, spanSampledHeader)));
String processId = getHeader(carrier, spanProcessIdHeader);
String spanName = getHeader(carrier, spanNameHeader);
if (spanName != null) {
spanBuilder.name(spanName);
}
if (processId != null) {
spanBuilder.processId(processId);
}
setParentIdIfApplicable(carrier, spanBuilder);
setParentIdIfApplicable(carrier, spanBuilder, spanParentIdHeader);
spanBuilder.remote(true);
return spanBuilder.build();
}
@@ -77,8 +109,9 @@ public class MessagingSpanExtractor implements SpanExtractor<Message<?>> {
return message.getHeaders().containsKey(name);
}
private void setParentIdIfApplicable(Message<?> carrier, SpanBuilder spanBuilder) {
String parentId = getHeader(carrier, Span.PARENT_ID_NAME);
private void setParentIdIfApplicable(Message<?> carrier, SpanBuilder spanBuilder,
String spanParentIdHeader) {
String parentId = getHeader(carrier, spanParentIdHeader);
if (parentId != null) {
spanBuilder.parent(Span.hexToId(parentId));
}

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

@@ -36,8 +36,6 @@ import org.springframework.util.StringUtils;
*/
class MessagingSpanInjector implements SpanInjector<MessageBuilder<?>> {
public static final String SPAN_HEADER = "X-Current-Span";
private final TraceKeys traceKeys;
public MessagingSpanInjector(TraceKeys traceKeys) {
@@ -50,31 +48,19 @@ class MessagingSpanInjector implements SpanInjector<MessageBuilder<?>> {
MessageHeaderAccessor accessor = MessageHeaderAccessor
.getMutableAccessor(initialMessage);
if (span == null) {
if (!Span.SPAN_SAMPLED
.equals(initialMessage.getHeaders().get(Span.SAMPLED_NAME))) {
if (!isSampled(initialMessage, Span.SAMPLED_NAME) ||
!isSampled(initialMessage, TraceMessageHeaders.SAMPLED_NAME)) {
// Backwards compatibility
accessor.setHeader(Span.SAMPLED_NAME, Span.SPAN_NOT_SAMPLED);
accessor.setHeader(TraceMessageHeaders.SAMPLED_NAME, Span.SPAN_NOT_SAMPLED);
carrier.setHeaders(accessor);
return;
}
return;
}
Map<String, String> headers = new HashMap<>();
addHeader(headers, Span.TRACE_ID_NAME, Span.idToHex(span.getTraceId()));
addHeader(headers, Span.SPAN_ID_NAME, Span.idToHex(span.getSpanId()));
if (span.isExportable()) {
addAnnotations(this.traceKeys, initialMessage, span);
Long parentId = getFirst(span.getParents());
if (parentId != null) {
addHeader(headers, Span.PARENT_ID_NAME, Span.idToHex(parentId));
}
addHeader(headers, Span.SPAN_NAME_NAME, span.getName());
addHeader(headers, Span.PROCESS_ID_NAME, span.getProcessId());
addHeader(headers, Span.SAMPLED_NAME, Span.SPAN_SAMPLED);
}
else {
addHeader(headers, Span.SAMPLED_NAME, Span.SPAN_NOT_SAMPLED);
}
accessor.setHeader(SPAN_HEADER, span);
addOldHeaders(span, initialMessage, accessor, headers);
addNewHeaders(span, initialMessage, accessor, headers);
accessor.copyHeaders(headers);
if (accessor instanceof NativeMessageHeaderAccessor) {
NativeMessageHeaderAccessor nativeAccessor = (NativeMessageHeaderAccessor) accessor;
@@ -85,6 +71,48 @@ class MessagingSpanInjector implements SpanInjector<MessageBuilder<?>> {
carrier.setHeaders(accessor);
}
private boolean isSampled(Message<?> initialMessage, String sampledHeaderName) {
return Span.SPAN_SAMPLED
.equals(initialMessage.getHeaders().get(sampledHeaderName));
}
// Backwards compatibility
private void addOldHeaders(Span span, Message<?> initialMessage,
MessageHeaderAccessor accessor, Map<String, String> headers) {
addHeaders(span, initialMessage, accessor, headers, Span.TRACE_ID_NAME,
Span.SPAN_ID_NAME, Span.PARENT_ID_NAME, Span.SPAN_NAME_NAME, Span.PROCESS_ID_NAME,
Span.SAMPLED_NAME, TraceMessageHeaders.OLD_SPAN_HEADER);
}
private void addNewHeaders(Span span, Message<?> initialMessage,
MessageHeaderAccessor accessor, Map<String, String> headers) {
addHeaders(span, initialMessage, accessor, headers, TraceMessageHeaders.TRACE_ID_NAME,
TraceMessageHeaders.SPAN_ID_NAME, TraceMessageHeaders.PARENT_ID_NAME, TraceMessageHeaders.SPAN_NAME_NAME,
TraceMessageHeaders.PROCESS_ID_NAME, TraceMessageHeaders.SAMPLED_NAME, TraceMessageHeaders.SPAN_HEADER);
}
private void addHeaders(Span span, Message<?> initialMessage,
MessageHeaderAccessor accessor, Map<String, String> headers, String traceIdHeader,
String spanIdHeader, String parentIdHeader, String spanNameHeader, String processIdHeader,
String spanSampledHeader, String spanHeader) {
addHeader(headers, traceIdHeader, Span.idToHex(span.getTraceId()));
addHeader(headers, spanIdHeader, Span.idToHex(span.getSpanId()));
if (span.isExportable()) {
addAnnotations(this.traceKeys, initialMessage, span);
Long parentId = getFirst(span.getParents());
if (parentId != null) {
addHeader(headers, parentIdHeader, Span.idToHex(parentId));
}
addHeader(headers, spanNameHeader, span.getName());
addHeader(headers, processIdHeader, span.getProcessId());
addHeader(headers, spanSampledHeader, Span.SPAN_SAMPLED);
}
else {
addHeader(headers, spanSampledHeader, Span.SPAN_NOT_SAMPLED);
}
accessor.setHeader(spanHeader, span);
}
private void addAnnotations(TraceKeys traceKeys, Message<?> message, Span span) {
for (String name : traceKeys.getMessage().getHeaders()) {
if (message.getHeaders().containsKey(name)) {

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

@@ -37,9 +37,6 @@ import org.springframework.messaging.support.MessageBuilder;
*/
public class TraceChannelInterceptor extends AbstractTraceChannelInterceptor {
private static final String SPAN_HEADER = "X-Current-Span";
private static final String MESSAGE_SENT_FROM_CLIENT = "X-Message-Sent";
public TraceChannelInterceptor(Tracer tracer, TraceKeys traceKeys,
SpanExtractor<Message<?>> spanExtractor,
SpanInjector<MessageBuilder<?>> spanInjector) {
@@ -76,11 +73,15 @@ public class TraceChannelInterceptor extends AbstractTraceChannelInterceptor {
String name = getMessageChannelName(channel);
Span span = startSpan(parentSpan, name, message);
MessageBuilder<?> messageBuilder = MessageBuilder.fromMessage(message);
if (message.getHeaders().containsKey(MESSAGE_SENT_FROM_CLIENT)) {
// Backwards compatibility
if (message.getHeaders().containsKey(TraceMessageHeaders.OLD_MESSAGE_SENT_FROM_CLIENT) ||
message.getHeaders().containsKey(TraceMessageHeaders.MESSAGE_SENT_FROM_CLIENT)) {
span.logEvent(Span.SERVER_RECV);
} else {
span.logEvent(Span.CLIENT_SEND);
messageBuilder.setHeader(MESSAGE_SENT_FROM_CLIENT, true);
// Backwards compatibility
messageBuilder.setHeader(TraceMessageHeaders.OLD_MESSAGE_SENT_FROM_CLIENT, true);
messageBuilder.setHeader(TraceMessageHeaders.MESSAGE_SENT_FROM_CLIENT, true);
}
getSpanInjector().inject(span, messageBuilder);
return messageBuilder.build();
@@ -90,7 +91,9 @@ public class TraceChannelInterceptor extends AbstractTraceChannelInterceptor {
if (span != null) {
return getTracer().createSpan(name, span);
}
if (Span.SPAN_NOT_SAMPLED.equals(message.getHeaders().get(Span.SAMPLED_NAME))) {
// Backwards compatibility
if (Span.SPAN_NOT_SAMPLED.equals(message.getHeaders().get(Span.SAMPLED_NAME)) ||
Span.SPAN_NOT_SAMPLED.equals(message.getHeaders().get(TraceMessageHeaders.SAMPLED_NAME))) {
return getTracer().createSpan(name, NeverSampler.INSTANCE);
}
return getTracer().createSpan(name);
@@ -121,7 +124,11 @@ public class TraceChannelInterceptor extends AbstractTraceChannelInterceptor {
if (message == null) {
return null;
}
Object object = message.getHeaders().get(SPAN_HEADER);
Object object = message.getHeaders().get(TraceMessageHeaders.OLD_SPAN_HEADER);
if (object instanceof Span) {
return (Span) object;
}
object = message.getHeaders().get(TraceMessageHeaders.SPAN_HEADER);
if (object instanceof Span) {
return (Span) object;
}

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

@@ -0,0 +1,40 @@
/*
* Copyright 2013-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.sleuth.instrument.messaging;
/**
* Contains trace related messaging headers
*
* @author Marcin Grzejszczak
* @since 1.0.4
*/
public class TraceMessageHeaders {
public static final String SPAN_ID_NAME = "spanId";
public static final String SAMPLED_NAME = "spanSampled";
public static final String PROCESS_ID_NAME = "spanProcessId";
public static final String PARENT_ID_NAME = "spanParentSpanId";
public static final String TRACE_ID_NAME = "spanTraceId";
public static final String SPAN_NAME_NAME = "spanName";
static final String OLD_SPAN_HEADER = "X-Current-Span";
static final String OLD_MESSAGE_SENT_FROM_CLIENT = "X-Message-Sent";
static final String SPAN_HEADER = "currentSpan";
static final String MESSAGE_SENT_FROM_CLIENT = "messageSent";
private TraceMessageHeaders() {}
}

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

@@ -88,12 +88,14 @@ public class TraceChannelInterceptorTests implements MessageHandler {
@Before
public void init() {
this.channel.subscribe(this);
this.accumulator.getSpans().clear();
}
@After
public void close() {
TestSpanContextHolder.removeCurrentSpan();
this.channel.unsubscribe(this);
this.accumulator.getSpans().clear();
}
@Test
@@ -124,6 +126,51 @@ public class TraceChannelInterceptorTests implements MessageHandler {
then(this.accumulator.getSpans()).hasSize(1);
}
// #332
@Test
public void shouldSendNewAndOldHeadersWhenNewHeadersWerePassed() {
this.channel.send(MessageBuilder.withPayload("hi")
.setHeader(TraceMessageHeaders.TRACE_ID_NAME, Span.idToHex(10L))
.setHeader(TraceMessageHeaders.SPAN_ID_NAME, Span.idToHex(20L)).build());
then(this.message).isNotNull();
String newSpanId = thenNewSpanIdEqualsOldSpanId();
thenNewTraceIdEqualsOldTraceId();
then(newSpanId).isNotEqualTo(20L);
then(this.accumulator.getSpans()).hasSize(1);
}
private String thenNewSpanIdEqualsOldSpanId() {
String newSpanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, String.class);
then(newSpanId).isNotNull();
String oldSpanId = this.message.getHeaders().get(Span.SPAN_ID_NAME, String.class);
then(oldSpanId).isEqualTo(newSpanId);
return newSpanId;
}
// #332
@Test
public void shouldSendNewAndOldHeadersWhenOldHeadersWerePassed() {
this.channel.send(MessageBuilder.withPayload("hi")
.setHeader(Span.TRACE_ID_NAME, Span.idToHex(10L))
.setHeader(Span.SPAN_ID_NAME, Span.idToHex(20L)).build());
then(this.message).isNotNull();
String newSpanId = thenNewSpanIdEqualsOldSpanId();
thenNewTraceIdEqualsOldTraceId();
then(newSpanId).isNotEqualTo(20L);
then(this.accumulator.getSpans()).hasSize(1);
}
private void thenNewTraceIdEqualsOldTraceId() {
long traceId = Span
.hexToId(this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, String.class));
then(traceId).isEqualTo(10L);
long oldTraceId = Span
.hexToId(this.message.getHeaders().get(Span.TRACE_ID_NAME, String.class));
then(oldTraceId).isEqualTo(traceId);
}
@Test
public void spanCreation() {
this.channel.send(MessageBuilder.withPayload("hi").build());