diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java index 07d3efd1c..bef74c433 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2019 the original author or authors. + * Copyright 2013-2020 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. @@ -59,8 +59,7 @@ import org.springframework.util.ClassUtils; * * @author Marcin Grzejszczak */ -public final class TracingChannelInterceptor extends ChannelInterceptorAdapter - implements ExecutorChannelInterceptor { +public final class TracingChannelInterceptor extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor { /** * Name of the class in Spring Cloud Stream that is a direct channel. @@ -109,25 +108,22 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter @Autowired TracingChannelInterceptor(Tracing tracing) { - this(tracing, MessageHeaderPropagation.INSTANCE, - MessageHeaderPropagation.INSTANCE); + this(tracing, MessageHeaderPropagation.INSTANCE, MessageHeaderPropagation.INSTANCE); } - TracingChannelInterceptor(Tracing tracing, - Propagation.Setter setter, + TracingChannelInterceptor(Tracing tracing, Propagation.Setter setter, Propagation.Getter getter) { this.tracing = tracing; this.tracer = tracing.tracer(); this.threadLocalSpan = ThreadLocalSpan.create(this.tracer); this.injector = tracing.propagation().injector(setter); this.extractor = tracing.propagation().extractor(getter); - this.integrationObjectSupportPresent = ClassUtils.isPresent( - "org.springframework.integration.context.IntegrationObjectSupport", null); - this.hasDirectChannelClass = ClassUtils - .isPresent("org.springframework.integration.channel.DirectChannel", null); - this.directWithAttributesChannelClass = ClassUtils - .isPresent(STREAM_DIRECT_CHANNEL, null) - ? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null; + this.integrationObjectSupportPresent = ClassUtils + .isPresent("org.springframework.integration.context.IntegrationObjectSupport", null); + this.hasDirectChannelClass = ClassUtils.isPresent("org.springframework.integration.channel.DirectChannel", + null); + this.directWithAttributesChannelClass = ClassUtils.isPresent(STREAM_DIRECT_CHANNEL, null) + ? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null; } public static TracingChannelInterceptor create(Tracing tracing) { @@ -170,12 +166,11 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter MessageHeaderAccessor headers = mutableHeaderAccessor(retrievedMessage); TraceContextOrSamplingFlags extracted = this.extractor.extract(headers); Span span = this.threadLocalSpan.next(extracted); - MessageHeaderPropagation.removeAnyTraceHeaders(headers, - this.tracing.propagation().keys()); + MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys()); this.injector.inject(span.context(), headers); if (!span.isNoop()) { span.kind(Span.Kind.PRODUCER).name("send").start(); - span.remoteServiceName(REMOTE_SERVICE_NAME); + span.remoteServiceName(toRemoteServiceName(headers)); addTags(message, span, channel); } if (log.isDebugEnabled()) { @@ -188,24 +183,31 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter return outputMessage; } - private Message outputMessage(Message originalMessage, - Message retrievedMessage, MessageHeaderAccessor additionalHeaders) { - MessageHeaderAccessor headers = MessageHeaderAccessor - .getMutableAccessor(originalMessage); + private String toRemoteServiceName(MessageHeaderAccessor headers) { + for (String key : headers.getMessageHeaders().keySet()) { + if (key.startsWith("kafka_")) { + return "kafka"; + } + else if (key.startsWith("amqp_")) { + return "rabbitmq"; + } + } + return REMOTE_SERVICE_NAME; + } + + private Message outputMessage(Message originalMessage, Message retrievedMessage, + MessageHeaderAccessor additionalHeaders) { + MessageHeaderAccessor headers = MessageHeaderAccessor.getMutableAccessor(originalMessage); if (originalMessage instanceof ErrorMessage) { ErrorMessage errorMessage = (ErrorMessage) originalMessage; - headers.copyHeaders(MessageHeaderPropagation.propagationHeaders( - additionalHeaders.getMessageHeaders(), + headers.copyHeaders(MessageHeaderPropagation.propagationHeaders(additionalHeaders.getMessageHeaders(), this.tracing.propagation().keys())); - return new ErrorMessage(errorMessage.getPayload(), - isWebSockets(headers) ? headers.getMessageHeaders() - : new MessageHeaders(headers.getMessageHeaders()), - errorMessage.getOriginalMessage()); + return new ErrorMessage(errorMessage.getPayload(), isWebSockets(headers) ? headers.getMessageHeaders() + : new MessageHeaders(headers.getMessageHeaders()), errorMessage.getOriginalMessage()); } headers.copyHeaders(additionalHeaders.getMessageHeaders()); return new GenericMessage<>(retrievedMessage.getPayload(), - isWebSockets(headers) ? headers.getMessageHeaders() - : new MessageHeaders(headers.getMessageHeaders())); + isWebSockets(headers) ? headers.getMessageHeaders() : new MessageHeaders(headers.getMessageHeaders())); } private boolean isWebSockets(MessageHeaderAccessor headerAccessor) { @@ -215,8 +217,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter private boolean isDirectChannel(MessageChannel channel) { Class targetClass = AopUtils.getTargetClass(channel); - boolean directChannel = this.hasDirectChannelClass - && DirectChannel.class.isAssignableFrom(targetClass); + boolean directChannel = this.hasDirectChannelClass && DirectChannel.class.isAssignableFrom(targetClass); if (!directChannel) { return false; } @@ -231,8 +232,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter } @Override - public void afterSendCompletion(Message message, MessageChannel channel, - boolean sent, Exception ex) { + public void afterSendCompletion(Message message, MessageChannel channel, boolean sent, Exception ex) { if (emptyMessage(message)) { return; } @@ -240,8 +240,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter 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); } @@ -258,12 +257,11 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter MessageHeaderAccessor headers = mutableHeaderAccessor(message); TraceContextOrSamplingFlags extracted = this.extractor.extract(headers); Span span = this.threadLocalSpan.next(extracted); - MessageHeaderPropagation.removeAnyTraceHeaders(headers, - this.tracing.propagation().keys()); + MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys()); this.injector.inject(span.context(), headers); if (!span.isNoop()) { span.kind(Span.Kind.CONSUMER).name("receive").start(); - span.remoteServiceName(REMOTE_SERVICE_NAME); + span.remoteServiceName(toRemoteServiceName(headers)); addTags(message, span, channel); } if (log.isDebugEnabled()) { @@ -272,21 +270,19 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter headers.setImmutable(); if (message instanceof ErrorMessage) { ErrorMessage errorMessage = (ErrorMessage) message; - return new ErrorMessage(errorMessage.getPayload(), - headers.getMessageHeaders(), errorMessage.getOriginalMessage()); + return new ErrorMessage(errorMessage.getPayload(), headers.getMessageHeaders(), + errorMessage.getOriginalMessage()); } return new GenericMessage<>(message.getPayload(), headers.getMessageHeaders()); } @Override - public void afterReceiveCompletion(Message message, MessageChannel channel, - Exception ex) { + 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); } @@ -296,8 +292,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter * context. It then creates a span for the handler, placing it in scope. */ @Override - public Message beforeHandle(Message message, MessageChannel channel, - MessageHandler handler) { + public Message beforeHandle(Message message, MessageChannel channel, MessageHandler handler) { if (emptyMessage(message)) { return message; } @@ -312,34 +307,28 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter consumerSpan.finish(); } // create and scope a span for the message processor - this.threadLocalSpan - .next(TraceContextOrSamplingFlags.create(consumerSpan.context())) - .name("handle").start(); + this.threadLocalSpan.next(TraceContextOrSamplingFlags.create(consumerSpan.context())).name("handle").start(); // remove any trace headers, but don't re-inject as we are synchronously // processing the // message and can rely on scoping to access this span later. - MessageHeaderPropagation.removeAnyTraceHeaders(headers, - this.tracing.propagation().keys()); + MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys()); if (log.isDebugEnabled()) { log.debug("Created a new span in before handle" + consumerSpan); } if (message instanceof ErrorMessage) { - return new ErrorMessage((Throwable) message.getPayload(), - headers.getMessageHeaders()); + return new ErrorMessage((Throwable) message.getPayload(), headers.getMessageHeaders()); } headers.setImmutable(); return new GenericMessage<>(message.getPayload(), headers.getMessageHeaders()); } @Override - public void afterMessageHandled(Message message, MessageChannel channel, - MessageHandler handler, Exception ex) { + 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); } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java index 1d8423f6c..10b5c1c91 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java @@ -29,8 +29,10 @@ import org.junit.After; import org.junit.Test; import zipkin2.Span; +import org.springframework.amqp.support.AmqpHeaders; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.kafka.support.KafkaHeaders; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -51,10 +53,11 @@ public class TracingChannelInterceptorTest { List spans = new ArrayList<>(); - ChannelInterceptor interceptor = TracingChannelInterceptor.create(Tracing.newBuilder() - .currentTraceContext(ThreadLocalCurrentTraceContext.newBuilder() - .addScopeDecorator(StrictScopeDecorator.create()).build()) - .spanReporter(this.spans::add).build()); + ChannelInterceptor interceptor = TracingChannelInterceptor + .create(Tracing.newBuilder() + .currentTraceContext(ThreadLocalCurrentTraceContext.newBuilder() + .addScopeDecorator(StrictScopeDecorator.create()).build()) + .spanReporter(this.spans::add).build()); QueueChannel channel = new QueueChannel(); @@ -83,10 +86,9 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", - "X-B3-SpanId", "X-B3-Sampled", "nativeHeaders"); - assertThat(this.spans).hasSize(1).flatExtracting(Span::kind) - .containsExactly(Span.Kind.PRODUCER); + assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", + "nativeHeaders"); + assertThat(this.spans).hasSize(1).flatExtracting(Span::kind).containsExactly(Span.Kind.PRODUCER); } @Test @@ -96,10 +98,9 @@ public class TracingChannelInterceptorTest { this.directChannel.send(MessageBuilder.withPayload("foo").build()); assertThat(this.message).isNotNull(); - assertThat(this.message.getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", - "X-B3-Sampled", "nativeHeaders"); - assertThat(this.spans).flatExtracting(Span::kind).contains(Span.Kind.CONSUMER, - Span.Kind.PRODUCER); + assertThat(this.message.getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", + "nativeHeaders"); + assertThat(this.spans).flatExtracting(Span::kind).contains(Span.Kind.CONSUMER, Span.Kind.PRODUCER); } @Test @@ -108,9 +109,8 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); - assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)) - .containsOnlyKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", - "spanTraceId", "spanId", "spanSampled"); + assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsOnlyKeys("X-B3-TraceId", + "X-B3-SpanId", "X-B3-Sampled", "spanTraceId", "spanId", "spanSampled"); } /** @@ -122,13 +122,11 @@ public class TracingChannelInterceptorTest { public void producerConsidersOldSpanIds() { this.channel.addInterceptor(producerSideOnly(this.interceptor)); - this.channel.send(MessageBuilder.withPayload("foo") - .setHeader("X-B3-TraceId", "000000000000000a") - .setHeader("X-B3-ParentSpanId", "000000000000000a") - .setHeader("X-B3-SpanId", "000000000000000b").build()); + this.channel.send(MessageBuilder.withPayload("foo").setHeader("X-B3-TraceId", "000000000000000a") + .setHeader("X-B3-ParentSpanId", "000000000000000a").setHeader("X-B3-SpanId", "000000000000000b") + .build()); - assertThat(this.channel.receive().getHeaders()).containsEntry("X-B3-ParentSpanId", - "000000000000000b"); + assertThat(this.channel.receive().getHeaders()).containsEntry("X-B3-ParentSpanId", "000000000000000b"); } @Test @@ -142,12 +140,10 @@ public class TracingChannelInterceptorTest { accessor.setNativeHeader("X-B3-ParentSpanId", "000000000000000a"); accessor.setNativeHeader("X-B3-SpanId", "000000000000000b"); - this.channel.send(MessageBuilder.withPayload("foo") - .copyHeaders(accessor.toMessageHeaders()).build()); + this.channel.send(MessageBuilder.withPayload("foo").copyHeaders(accessor.toMessageHeaders()).build()); - assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)) - .containsEntry("X-B3-ParentSpanId", - Collections.singletonList("000000000000000b")); + assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsEntry("X-B3-ParentSpanId", + Collections.singletonList("000000000000000b")); } /** @@ -160,10 +156,9 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", - "X-B3-SpanId", "X-B3-Sampled", "nativeHeaders"); - assertThat(this.spans).hasSize(1).flatExtracting(Span::kind) - .containsExactly(Span.Kind.CONSUMER); + assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", + "nativeHeaders"); + assertThat(this.spans).hasSize(1).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER); } @Test @@ -172,9 +167,8 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); - assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)) - .containsOnlyKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", - "spanTraceId", "spanId", "spanSampled"); + assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsOnlyKeys("X-B3-TraceId", + "X-B3-SpanId", "X-B3-Sampled", "spanTraceId", "spanId", "spanSampled"); } @Test @@ -186,10 +180,9 @@ public class TracingChannelInterceptorTest { channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", - "X-B3-SpanId", "X-B3-Sampled", "nativeHeaders"); - assertThat(this.spans).flatExtracting(Span::kind) - .containsExactly(Span.Kind.CONSUMER, null); + assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled", + "nativeHeaders"); + assertThat(this.spans).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER, null); } /** @@ -206,8 +199,7 @@ public class TracingChannelInterceptorTest { channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", - "X-B3-SpanId", "X-B3-Sampled"); + assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled"); } @Test @@ -219,8 +211,8 @@ public class TracingChannelInterceptorTest { channel.send(MessageBuilder.withPayload("foo").build()); - assertThat((Map) messages.get(0).getHeaders().get(NATIVE_HEADERS)) - .doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled"); + assertThat((Map) messages.get(0).getHeaders().get(NATIVE_HEADERS)).doesNotContainKeys("X-B3-TraceId", + "X-B3-SpanId", "X-B3-Sampled"); } @Test @@ -230,8 +222,8 @@ public class TracingChannelInterceptorTest { this.channel.send(MessageBuilder.withPayload("foo").build()); this.channel.receive(); - assertThat(this.spans).flatExtracting(Span::kind) - .containsExactlyInAnyOrder(Span.Kind.CONSUMER, Span.Kind.PRODUCER); + assertThat(this.spans).flatExtracting(Span::kind).containsExactlyInAnyOrder(Span.Kind.CONSUMER, + Span.Kind.PRODUCER); } @Test @@ -243,8 +235,7 @@ public class TracingChannelInterceptorTest { channel.send(MessageBuilder.withPayload("foo").build()); - assertThat(this.spans).flatExtracting(Span::kind) - .containsExactly(Span.Kind.CONSUMER, null, Span.Kind.PRODUCER); + assertThat(this.spans).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER, null, Span.Kind.PRODUCER); } @Test @@ -255,54 +246,41 @@ public class TracingChannelInterceptorTest { Map errorChannelHeaders = new HashMap<>(); errorChannelHeaders.put(MessageHeaders.REPLY_CHANNEL, errorsReplyChannel); errorChannelHeaders.put(MessageHeaders.ERROR_CHANNEL, errorsReplyChannel); - this.channel - .send(new ErrorMessage( - new MessagingException(MessageBuilder.withPayload("hi") - .setHeader(TraceMessageHeaders.TRACE_ID_NAME, - "000000000000000a") - .setHeader(TraceMessageHeaders.SPAN_ID_NAME, - "000000000000000a") - .setReplyChannel(deadReplyChannel) - .setErrorChannel(deadReplyChannel).build()), - errorChannelHeaders)); + this.channel.send(new ErrorMessage( + new MessagingException(MessageBuilder.withPayload("hi") + .setHeader(TraceMessageHeaders.TRACE_ID_NAME, "000000000000000a") + .setHeader(TraceMessageHeaders.SPAN_ID_NAME, "000000000000000a") + .setReplyChannel(deadReplyChannel).setErrorChannel(deadReplyChannel).build()), + errorChannelHeaders)); this.message = this.channel.receive(); assertThat(this.message).isNotNull(); - String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, - String.class); + String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, String.class); assertThat(spanId).isNotNull(); - String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, - String.class); + String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, String.class); assertThat(traceId).isEqualTo("000000000000000a"); assertThat(spanId).isNotEqualTo("000000000000000a"); assertThat(this.spans).hasSize(2); - assertThat(this.message.getHeaders().getReplyChannel()) - .isSameAs(errorsReplyChannel); - assertThat(this.message.getHeaders().getErrorChannel()) - .isSameAs(errorsReplyChannel); + assertThat(this.message.getHeaders().getReplyChannel()).isSameAs(errorsReplyChannel); + assertThat(this.message.getHeaders().getErrorChannel()).isSameAs(errorsReplyChannel); } @Test public void errorMessageOriginalMessageRetained() { this.channel.addInterceptor(this.interceptor); - Message originalMessage = MessageBuilder.withPayload("Hello") - .setHeader("header", "value").build(); - Message failedMessage = MessageBuilder.fromMessage(originalMessage) - .removeHeader("header").build(); - this.channel.send(new ErrorMessage(new MessagingException(failedMessage), - originalMessage.getHeaders(), originalMessage)); + Message originalMessage = MessageBuilder.withPayload("Hello").setHeader("header", "value").build(); + Message failedMessage = MessageBuilder.fromMessage(originalMessage).removeHeader("header").build(); + this.channel.send( + new ErrorMessage(new MessagingException(failedMessage), originalMessage.getHeaders(), originalMessage)); this.message = this.channel.receive(); assertThat(this.message).isNotNull(); - assertThat(this.message).isInstanceOfSatisfying(ErrorMessage.class, - errorMessage -> { - assertThat(errorMessage.getOriginalMessage()) - .isSameAs(originalMessage); - assertThat(errorMessage.getHeaders().get("header")) - .isEqualTo("value"); - }); + assertThat(this.message).isInstanceOfSatisfying(ErrorMessage.class, errorMessage -> { + assertThat(errorMessage.getOriginalMessage()).isSameAs(originalMessage); + assertThat(errorMessage.getHeaders().get("header")).isEqualTo("value"); + }); } @Test @@ -311,22 +289,60 @@ public class TracingChannelInterceptorTest { Map errorChannelHeaders = new HashMap<>(); errorChannelHeaders.put(TraceMessageHeaders.TRACE_ID_NAME, "000000000000000a"); errorChannelHeaders.put(TraceMessageHeaders.SPAN_ID_NAME, "000000000000000a"); - this.channel.send(new ErrorMessage(new MessagingException("exception"), - errorChannelHeaders)); + this.channel.send(new ErrorMessage(new MessagingException("exception"), errorChannelHeaders)); this.message = this.channel.receive(); assertThat(this.message).isNotNull(); - String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, - String.class); + String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, String.class); assertThat(spanId).isNotNull(); - String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, - String.class); + String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, String.class); assertThat(traceId).isEqualTo("000000000000000a"); assertThat(spanId).isNotEqualTo("000000000000000a"); assertThat(this.spans).hasSize(2); } + @Test + public void should_store_kafka_as_remote_service_name_when_kafka_header_is_present() { + ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(); + channel.addInterceptor(this.interceptor); + List> messages = new ArrayList<>(); + channel.subscribe(messages::add); + + Map headers = new HashMap<>(); + headers.put(KafkaHeaders.MESSAGE_KEY, "hello"); + channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers))); + + assertThat(this.spans).flatExtracting(Span::remoteServiceName).contains("kafka"); + } + + @Test + public void should_store_rabbitmq_as_remote_service_name_when_rabbit_header_is_present() { + ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(); + channel.addInterceptor(this.interceptor); + List> messages = new ArrayList<>(); + channel.subscribe(messages::add); + + Map headers = new HashMap<>(); + headers.put(AmqpHeaders.RECEIVED_ROUTING_KEY, "hello"); + channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers))); + + assertThat(this.spans).flatExtracting(Span::remoteServiceName).contains("rabbitmq"); + } + + @Test + public void should_store_broker_as_remote_service_name_when_no_special_headers_were_found() { + ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(); + channel.addInterceptor(this.interceptor); + List> messages = new ArrayList<>(); + channel.subscribe(messages::add); + + Map headers = new HashMap<>(); + channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers))); + + assertThat(this.spans).flatExtracting(Span::remoteServiceName).containsOnly("broker", null); + } + ChannelInterceptor producerSideOnly(ChannelInterceptor delegate) { return new ChannelInterceptorAdapter() { @Override @@ -335,8 +351,7 @@ public class TracingChannelInterceptorTest { } @Override - public void afterSendCompletion(Message message, MessageChannel channel, - boolean sent, Exception ex) { + public void afterSendCompletion(Message message, MessageChannel channel, boolean sent, Exception ex) { delegate.afterSendCompletion(message, channel, sent, ex); } }; @@ -350,29 +365,24 @@ public class TracingChannelInterceptorTest { } @Override - public void afterReceiveCompletion(Message message, MessageChannel channel, - Exception ex) { + public void afterReceiveCompletion(Message message, MessageChannel channel, Exception ex) { delegate.afterReceiveCompletion(message, channel, ex); } }; } ExecutorChannelInterceptor executorSideOnly(ChannelInterceptor delegate) { - class ExecutorSideOnly extends ChannelInterceptorAdapter - implements ExecutorChannelInterceptor { + class ExecutorSideOnly extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor { @Override - public Message beforeHandle(Message message, MessageChannel channel, - MessageHandler handler) { - return ((ExecutorChannelInterceptor) delegate).beforeHandle(message, - channel, handler); + public Message beforeHandle(Message message, MessageChannel channel, MessageHandler handler) { + return ((ExecutorChannelInterceptor) delegate).beforeHandle(message, channel, handler); } @Override - public void afterMessageHandled(Message message, MessageChannel channel, - MessageHandler handler, Exception ex) { - ((ExecutorChannelInterceptor) delegate).afterMessageHandled(message, - channel, handler, ex); + public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, + Exception ex) { + ((ExecutorChannelInterceptor) delegate).afterMessageHandled(message, channel, handler, ex); } }