diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java index 08806b453..80655caa9 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java @@ -57,9 +57,11 @@ public class SleuthMessagingProperties { /** * An array of patterns against which channel names will be matched. * @see org.springframework.integration.config.GlobalChannelInterceptor#patterns() - * Defaults to any channel name not matching the Hystrix Stream and functional Stream channel names. + * Defaults to any channel name not matching the Hystrix Stream and functional + * Stream channel names. */ - private String[] patterns = new String[] { "!hystrixStreamOutput*", "*", "!channel*"}; + private String[] patterns = new String[] { "!hystrixStreamOutput*", "*", + "!channel*" }; /** * Enable Spring Integration sleuth instrumentation. 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 bef74c433..6a0330028 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 @@ -59,7 +59,8 @@ 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. @@ -108,22 +109,25 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i @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) { @@ -166,7 +170,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i 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(); @@ -195,19 +200,24 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i return REMOTE_SERVICE_NAME; } - private Message outputMessage(Message originalMessage, Message retrievedMessage, - MessageHeaderAccessor additionalHeaders) { - MessageHeaderAccessor headers = MessageHeaderAccessor.getMutableAccessor(originalMessage); + 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) { @@ -217,7 +227,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i 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; } @@ -232,7 +243,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i } @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,7 +252,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i 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); } @@ -257,7 +270,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i 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(); @@ -270,19 +284,21 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i 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); } @@ -292,7 +308,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i * 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; } @@ -307,28 +324,34 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i 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 10b5c1c91..88f334ccc 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 @@ -53,11 +53,10 @@ 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(); @@ -86,9 +85,10 @@ 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 @@ -98,9 +98,10 @@ 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 @@ -109,8 +110,9 @@ 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,11 +124,13 @@ 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 @@ -140,10 +144,12 @@ 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")); } /** @@ -156,9 +162,10 @@ 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 @@ -167,8 +174,9 @@ 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 @@ -180,9 +188,10 @@ 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); } /** @@ -199,7 +208,8 @@ 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 @@ -211,8 +221,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 @@ -222,8 +232,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 @@ -235,7 +245,8 @@ 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 @@ -246,41 +257,54 @@ 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 @@ -289,14 +313,17 @@ 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); @@ -327,7 +354,8 @@ public class TracingChannelInterceptorTest { headers.put(AmqpHeaders.RECEIVED_ROUTING_KEY, "hello"); channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers))); - assertThat(this.spans).flatExtracting(Span::remoteServiceName).contains("rabbitmq"); + assertThat(this.spans).flatExtracting(Span::remoteServiceName) + .contains("rabbitmq"); } @Test @@ -340,7 +368,8 @@ public class TracingChannelInterceptorTest { Map headers = new HashMap<>(); channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers))); - assertThat(this.spans).flatExtracting(Span::remoteServiceName).containsOnly("broker", null); + assertThat(this.spans).flatExtracting(Span::remoteServiceName) + .containsOnly("broker", null); } ChannelInterceptor producerSideOnly(ChannelInterceptor delegate) { @@ -351,7 +380,8 @@ 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); } }; @@ -365,24 +395,29 @@ 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); } } diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/FlowsScopePassingSpanSubscriberTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/FlowsScopePassingSpanSubscriberTests.java new file mode 100644 index 000000000..a90a2c0ac --- /dev/null +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/FlowsScopePassingSpanSubscriberTests.java @@ -0,0 +1,153 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.reactor; + +import java.util.Objects; +import java.util.function.Function; + +import brave.propagation.CurrentTraceContext; +import brave.propagation.CurrentTraceContext.Scope; +import brave.propagation.TraceContext; +import org.assertj.core.presentation.StandardRepresentation; +import org.awaitility.Awaitility; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import reactor.core.CoreSubscriber; +import reactor.core.publisher.Hooks; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +import org.springframework.context.annotation.AnnotationConfigApplicationContext; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.BDDAssertions.then; +import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.scopePassingSpanOperator; +import static org.springframework.cloud.sleuth.instrument.reactor.TraceReactorAutoConfiguration.TraceReactorConfiguration.SLEUTH_TRACE_REACTOR_KEY; + +/** + * @author Marcin Grzejszczak + */ +public class FlowsScopePassingSpanSubscriberTests { + + static { + // AssertJ will recognise QueueSubscription implements queue and try to invoke + // iterator. That's not allowed, and will cause an exception + // Fuseable$QueueSubscription.NOT_SUPPORTED_MESSAGE. + // This ensures AssertJ uses normal toString. + StandardRepresentation.registerFormatterForType(ScopePassingSpanSubscriber.class, + Objects::toString); + } + + final CurrentTraceContext currentTraceContext = CurrentTraceContext.Default.create(); + + TraceContext context = TraceContext.newBuilder().traceId(1).spanId(1).sampled(true) + .build(); + + AnnotationConfigApplicationContext springContext = new AnnotationConfigApplicationContext(); + + @Before + public void setup() { + Hooks.resetOnEachOperator(SLEUTH_TRACE_REACTOR_KEY); + Hooks.resetOnLastOperator(SLEUTH_TRACE_REACTOR_KEY); + Schedulers.resetOnScheduleHooks(); + } + + @After + public void close() { + springContext.close(); + } + + @Test + public void should_not_trace_scalar_flows() { + springContext.registerBean(CurrentTraceContext.class, () -> currentTraceContext); + springContext.refresh(); + + Function, ? extends Publisher> transformer = scopePassingSpanOperator( + this.springContext); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Subscriber assertNoSpanSubscriber = new CoreSubscriber() { + @Override + public void onSubscribe(Subscription s) { + s.request(Long.MAX_VALUE); + assertThat(s).isNotInstanceOf(ScopePassingSpanSubscriber.class); + } + + @Override + public void onNext(Object o) { + + } + + @Override + public void onError(Throwable t) { + + } + + @Override + public void onComplete() { + + } + }; + + Subscriber assertSpanSubscriber = new CoreSubscriber() { + @Override + public void onSubscribe(Subscription s) { + s.request(Long.MAX_VALUE); + assertThat(s).isInstanceOf(ScopePassingSpanSubscriber.class); + } + + @Override + public void onNext(Object o) { + + } + + @Override + public void onError(Throwable t) { + + } + + @Override + public void onComplete() { + + } + }; + transformer.apply(Mono.just(1).hide()).subscribe(assertSpanSubscriber); + + transformer.apply(Mono.just(1)).subscribe(assertNoSpanSubscriber); + + transformer.apply(Mono.error(new Exception()).hide()) + .subscribe(assertSpanSubscriber); + + transformer.apply(Mono.error(new Exception())) + .subscribe(assertNoSpanSubscriber); + + transformer.apply(Mono.empty().hide()) + .subscribe(assertSpanSubscriber); + + transformer.apply(Mono.empty()).subscribe(assertNoSpanSubscriber); + + } + + Awaitility.await() + .untilAsserted(() -> then(this.currentTraceContext.get()).isNull()); + } + +} diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java index ccbb95a83..7e6223757 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java @@ -17,28 +17,20 @@ package org.springframework.cloud.sleuth.instrument.reactor; import java.util.Objects; -import java.util.function.Function; import brave.propagation.CurrentTraceContext; import brave.propagation.CurrentTraceContext.Scope; import brave.propagation.TraceContext; import org.assertj.core.presentation.StandardRepresentation; -import org.awaitility.Awaitility; import org.junit.After; import org.junit.Test; -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; import reactor.core.CoreSubscriber; import reactor.core.publisher.BaseSubscriber; -import reactor.core.publisher.Mono; import reactor.util.context.Context; import org.springframework.context.annotation.AnnotationConfigApplicationContext; -import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.BDDAssertions.then; -import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.scopePassingSpanOperator; /** * @author Marcin Grzejszczak @@ -50,14 +42,17 @@ public class ScopePassingSpanSubscriberTests { // iterator. That's not allowed, and will cause an exception // Fuseable$QueueSubscription.NOT_SUPPORTED_MESSAGE. // This ensures AssertJ uses normal toString. - StandardRepresentation.registerFormatterForType(ScopePassingSpanSubscriber.class, Objects::toString); + StandardRepresentation.registerFormatterForType(ScopePassingSpanSubscriber.class, + Objects::toString); } final CurrentTraceContext currentTraceContext = CurrentTraceContext.Default.create(); - TraceContext context = TraceContext.newBuilder().traceId(1).spanId(1).sampled(true).build(); + TraceContext context = TraceContext.newBuilder().traceId(1).spanId(1).sampled(true) + .build(); - TraceContext context2 = TraceContext.newBuilder().traceId(1).spanId(2).sampled(true).build(); + TraceContext context2 = TraceContext.newBuilder().traceId(1).spanId(2).sampled(true) + .build(); AnnotationConfigApplicationContext springContext = new AnnotationConfigApplicationContext(); @@ -68,16 +63,16 @@ public class ScopePassingSpanSubscriberTests { @Test public void should_propagate_current_context() { - ScopePassingSpanSubscriber subscriber = new ScopePassingSpanSubscriber<>(null, Context.of("foo", "bar"), - this.currentTraceContext, null); + ScopePassingSpanSubscriber subscriber = new ScopePassingSpanSubscriber<>(null, + Context.of("foo", "bar"), this.currentTraceContext, null); then((String) subscriber.currentContext().get("foo")).isEqualTo("bar"); } @Test public void should_set_empty_context_when_context_is_null() { - ScopePassingSpanSubscriber subscriber = new ScopePassingSpanSubscriber<>(null, Context.empty(), - this.currentTraceContext, null); + ScopePassingSpanSubscriber subscriber = new ScopePassingSpanSubscriber<>(null, + Context.empty(), this.currentTraceContext, null); then(subscriber.currentContext().isEmpty()).isTrue(); } @@ -85,84 +80,12 @@ public class ScopePassingSpanSubscriberTests { @Test public void should_put_current_span_to_context() { try (Scope ws = this.currentTraceContext.newScope(context2)) { - CoreSubscriber subscriber = new ScopePassingSpanSubscriber<>(new BaseSubscriber() { - }, Context.empty(), currentTraceContext, context); + CoreSubscriber subscriber = new ScopePassingSpanSubscriber<>( + new BaseSubscriber() { + }, Context.empty(), currentTraceContext, context); then(subscriber.currentContext().get(TraceContext.class)).isEqualTo(context); } } - @Test - public void should_not_trace_scalar_flows() { - springContext.registerBean(CurrentTraceContext.class, () -> currentTraceContext); - springContext.refresh(); - - Awaitility.await().untilAsserted(() -> { - Function, ? extends Publisher> transformer = scopePassingSpanOperator( - this.springContext); - - try (Scope ws = this.currentTraceContext.newScope(context)) { - Subscriber assertNoSpanSubscriber = new CoreSubscriber() { - @Override - public void onSubscribe(Subscription s) { - s.request(Long.MAX_VALUE); - assertThat(s).isNotInstanceOf(ScopePassingSpanSubscriber.class); - } - - @Override - public void onNext(Object o) { - - } - - @Override - public void onError(Throwable t) { - - } - - @Override - public void onComplete() { - - } - }; - - Subscriber assertSpanSubscriber = new CoreSubscriber() { - @Override - public void onSubscribe(Subscription s) { - s.request(Long.MAX_VALUE); - assertThat(s).isInstanceOf(ScopePassingSpanSubscriber.class); - } - - @Override - public void onNext(Object o) { - - } - - @Override - public void onError(Throwable t) { - - } - - @Override - public void onComplete() { - - } - }; - transformer.apply(Mono.just(1).hide()).subscribe(assertSpanSubscriber); - - transformer.apply(Mono.just(1)).subscribe(assertNoSpanSubscriber); - - transformer.apply(Mono.error(new Exception()).hide()).subscribe(assertSpanSubscriber); - - transformer.apply(Mono.error(new Exception())).subscribe(assertNoSpanSubscriber); - - transformer.apply(Mono.empty().hide()).subscribe(assertSpanSubscriber); - - transformer.apply(Mono.empty()).subscribe(assertNoSpanSubscriber); - - } - - then(this.currentTraceContext.get()).isNull(); - }); - } - }