diff --git a/docs/src/main/asciidoc/integrations.adoc b/docs/src/main/asciidoc/integrations.adoc index c83059b2c..8eaa29b86 100644 --- a/docs/src/main/asciidoc/integrations.adoc +++ b/docs/src/main/asciidoc/integrations.adoc @@ -272,6 +272,20 @@ If you want to customize the way tracing context is read from and written to mes * `Propagator.Setter` - for writing headers to the message * `Propagator.Getter` - for reading headers from the message +[[sleuth-messaging-spring-integration-customization]] +==== Spring Integration Customization + +==== Customizing messaging spans + +In order to change the default span names and tags, just register a bean of type `MessageSpanCustomizer`. You can also +override the existing `DefaultMessageSpanCustomizer` to extend the existing behaviour. + +[source,java] +---- +@Component +include::{common_tests_path}/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java[tags=message_span_customizer,indent=2] +---- + [[sleuth-messaging-spring-cloud-function-integration]] === Spring Cloud Function and Spring Cloud Stream diff --git a/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceSpringIntegrationAutoConfiguration.java b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceSpringIntegrationAutoConfiguration.java index 88f40a177..e0a8a4e61 100644 --- a/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceSpringIntegrationAutoConfiguration.java +++ b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceSpringIntegrationAutoConfiguration.java @@ -22,6 +22,7 @@ import org.springframework.boot.autoconfigure.AutoConfigureAfter; import org.springframework.boot.autoconfigure.condition.AnyNestedCondition; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; @@ -29,6 +30,8 @@ import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.sleuth.Tracer; import org.springframework.cloud.sleuth.autoconfig.brave.BraveAutoConfiguration; import org.springframework.cloud.sleuth.autoconfig.otel.OtelAutoConfiguration; +import org.springframework.cloud.sleuth.instrument.messaging.DefaultMessageSpanCustomizer; +import org.springframework.cloud.sleuth.instrument.messaging.MessageSpanCustomizer; import org.springframework.cloud.sleuth.instrument.messaging.TracingChannelInterceptor; import org.springframework.cloud.sleuth.propagation.Propagator; import org.springframework.context.annotation.Bean; @@ -62,6 +65,12 @@ import org.springframework.util.StringUtils; @Conditional(TracingChannelInterceptorCondition.class) public class TraceSpringIntegrationAutoConfiguration { + @Bean + @ConditionalOnMissingBean + MessageSpanCustomizer defaultMessageSpanCustomizer() { + return new DefaultMessageSpanCustomizer(); + } + @Bean GlobalChannelInterceptorWrapper traceGlobalChannelInterceptorWrapper(TracingChannelInterceptor interceptor, SleuthIntegrationMessagingProperties properties) { @@ -74,9 +83,9 @@ public class TraceSpringIntegrationAutoConfiguration { TracingChannelInterceptor traceChannelInterceptor(Tracer tracer, Propagator propagator, Propagator.Setter traceMessagePropagationSetter, Propagator.Getter traceMessagePropagationGetter, - SleuthMessagingProperties properties) { + SleuthMessagingProperties properties, MessageSpanCustomizer messageSpanCustomizer) { return new TracingChannelInterceptor(tracer, propagator, traceMessagePropagationSetter, - traceMessagePropagationGetter, remoteServiceNameMapper(properties)); + traceMessagePropagationGetter, remoteServiceNameMapper(properties), messageSpanCustomizer); } static Function remoteServiceNameMapper(SleuthMessagingProperties properties) { diff --git a/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceWebSocketAutoConfiguration.java b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceWebSocketAutoConfiguration.java index ddd8e3025..d1b775d0d 100644 --- a/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceWebSocketAutoConfiguration.java +++ b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceWebSocketAutoConfiguration.java @@ -20,12 +20,16 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.AutoConfigureAfter; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.cloud.sleuth.Tracer; import org.springframework.cloud.sleuth.autoconfig.brave.BraveAutoConfiguration; import org.springframework.cloud.sleuth.autoconfig.otel.OtelAutoConfiguration; +import org.springframework.cloud.sleuth.instrument.messaging.DefaultMessageSpanCustomizer; +import org.springframework.cloud.sleuth.instrument.messaging.MessageSpanCustomizer; import org.springframework.cloud.sleuth.instrument.messaging.TracingChannelInterceptor; import org.springframework.cloud.sleuth.propagation.Propagator; +import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.messaging.simp.config.ChannelRegistration; import org.springframework.messaging.simp.config.MessageBrokerRegistry; @@ -64,6 +68,15 @@ class TraceWebSocketAutoConfiguration extends AbstractWebSocketMessageBrokerConf @Autowired SleuthMessagingProperties sleuthMessagingProperties; + @Autowired + MessageSpanCustomizer messageSpanCustomizer; + + @Bean + @ConditionalOnMissingBean + MessageSpanCustomizer defaultMessageSpanCustomizer() { + return new DefaultMessageSpanCustomizer(); + } + @Override public void registerStompEndpoints(StompEndpointRegistry registry) { // The user must register their own endpoints @@ -71,23 +84,26 @@ class TraceWebSocketAutoConfiguration extends AbstractWebSocketMessageBrokerConf @Override public void configureMessageBroker(MessageBrokerRegistry registry) { - registry.configureBrokerChannel().setInterceptors(new TracingChannelInterceptor(this.tracer, this.propagator, - this.setter, this.getter, - TraceSpringIntegrationAutoConfiguration.remoteServiceNameMapper(this.sleuthMessagingProperties))); + registry.configureBrokerChannel() + .setInterceptors(new TracingChannelInterceptor(this.tracer, this.propagator, this.setter, this.getter, + TraceSpringIntegrationAutoConfiguration.remoteServiceNameMapper(this.sleuthMessagingProperties), + this.messageSpanCustomizer)); } @Override public void configureClientOutboundChannel(ChannelRegistration registration) { - registration.setInterceptors(new TracingChannelInterceptor(this.tracer, this.propagator, this.setter, - this.getter, - TraceSpringIntegrationAutoConfiguration.remoteServiceNameMapper(this.sleuthMessagingProperties))); + registration + .setInterceptors(new TracingChannelInterceptor(this.tracer, this.propagator, this.setter, this.getter, + TraceSpringIntegrationAutoConfiguration.remoteServiceNameMapper(this.sleuthMessagingProperties), + this.messageSpanCustomizer)); } @Override public void configureClientInboundChannel(ChannelRegistration registration) { - registration.setInterceptors(new TracingChannelInterceptor(this.tracer, this.propagator, this.setter, - this.getter, - TraceSpringIntegrationAutoConfiguration.remoteServiceNameMapper(this.sleuthMessagingProperties))); + registration + .setInterceptors(new TracingChannelInterceptor(this.tracer, this.propagator, this.setter, this.getter, + TraceSpringIntegrationAutoConfiguration.remoteServiceNameMapper(this.sleuthMessagingProperties), + this.messageSpanCustomizer)); } } diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/DefaultMessageSpanCustomizer.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/DefaultMessageSpanCustomizer.java new file mode 100644 index 000000000..b8a178abd --- /dev/null +++ b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/DefaultMessageSpanCustomizer.java @@ -0,0 +1,113 @@ +/* + * 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.messaging; + +import org.springframework.cloud.sleuth.Span; +import org.springframework.cloud.sleuth.SpanCustomizer; +import org.springframework.cloud.sleuth.internal.SpanNameUtil; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.context.IntegrationObjectSupport; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.ClassUtils; + +/** + * Provides default customization of messaging spans. + * + * @author Marcin Grzejszczak + * @since 2.2.7 + */ +public class DefaultMessageSpanCustomizer implements MessageSpanCustomizer { + + private final boolean integrationObjectSupportPresent; + + public DefaultMessageSpanCustomizer() { + this.integrationObjectSupportPresent = ClassUtils + .isPresent("org.springframework.integration.context.IntegrationObjectSupport", null); + } + + /** + * Retrieves the channel name from the {@link MessageChannel}. + * @param channel - message channel from which message got received or was sent to + * @return channel name + * @since 2.2.7 + */ + protected String channelName(MessageChannel channel) { + String name = null; + if (this.integrationObjectSupportPresent) { + if (channel instanceof IntegrationObjectSupport) { + name = ((IntegrationObjectSupport) channel).getComponentName(); + } + if (name == null && channel instanceof AbstractMessageChannel) { + name = ((AbstractMessageChannel) channel).getFullChannelName(); + } + } + if (name == null) { + return channel.toString(); + } + return name; + } + + private String messageChannelName(MessageChannel channel) { + return SpanNameUtil.shorten(channelName(channel)); + } + + @Override + public Span customizeHandle(Span spanCustomizer, Message message, @Nullable MessageChannel messageChannel) { + spanCustomizer.name("handle"); + addTags(spanCustomizer, messageChannel); + return spanCustomizer; + } + + @Override + public Span.Builder customizeHandle(Span.Builder spanCustomizer, Message message, + @Nullable MessageChannel messageChannel) { + spanCustomizer.name("handle"); + addTags(spanCustomizer, messageChannel); + return spanCustomizer; + } + + @Override + public Span.Builder customizeReceive(Span.Builder spanCustomizer, Message message, + @Nullable MessageChannel messageChannel) { + spanCustomizer.name("receive"); + addTags(spanCustomizer, messageChannel); + return spanCustomizer; + } + + @Override + public Span.Builder customizeSend(Span.Builder spanCustomizer, Message message, + @Nullable MessageChannel messageChannel) { + spanCustomizer.name("send"); + addTags(spanCustomizer, messageChannel); + return spanCustomizer; + } + + private void addTags(SpanCustomizer result, MessageChannel channel) { + if (channel != null) { + result.tag("channel", messageChannelName(channel)); + } + } + + private void addTags(Span.Builder result, MessageChannel channel) { + if (channel != null) { + result.tag("channel", messageChannelName(channel)); + } + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/MessageSpanCustomizer.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/MessageSpanCustomizer.java new file mode 100644 index 000000000..db761f716 --- /dev/null +++ b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/MessageSpanCustomizer.java @@ -0,0 +1,77 @@ +/* + * 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.messaging; + +import org.springframework.cloud.sleuth.Span; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.support.ExecutorChannelInterceptor; + +/** + * Allows customization of messaging spans. + * + * @author Marcin Grzejszczak + * @since 3.0.0 + */ +public interface MessageSpanCustomizer { + + /** + * Customizes the span created when + * {@link ExecutorChannelInterceptor#beforeHandle(Message, MessageChannel, MessageHandler)} + * gets called. + * @param spanCustomizer current span to customize + * @param message received or sent message + * @param messageChannel channel from / to which the message was sent + * @return customized span + */ + Span customizeHandle(Span spanCustomizer, Message message, @Nullable MessageChannel messageChannel); + + /** + * Customizes the span created when + * {@link ExecutorChannelInterceptor#beforeHandle(Message, MessageChannel, MessageHandler)} + * gets called for a span retrieved from the incoming headers. + * @param builder current span to customize + * @param message received or sent message + * @param messageChannel channel from / to which the message was sent + * @return customized span + */ + Span.Builder customizeHandle(Span.Builder builder, Message message, @Nullable MessageChannel messageChannel); + + /** + * Customizes the span created when + * {@link ExecutorChannelInterceptor#postReceive(Message, MessageChannel)} gets + * called. + * @param builder current span to customize + * @param message received or sent message + * @param messageChannel channel from / to which the message was sent + * @return customized span + */ + Span.Builder customizeReceive(Span.Builder builder, Message message, @Nullable MessageChannel messageChannel); + + /** + * Customizes the span created when + * {@link ExecutorChannelInterceptor#preSend(Message, MessageChannel)} gets called. + * @param builder current span to customize + * @param message received or sent message + * @param messageChannel channel from / to which the message was sent + * @return customized span + */ + Span.Builder customizeSend(Span.Builder builder, Message message, @Nullable MessageChannel messageChannel); + +} diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java index 5b879a0fd..030732574 100644 --- a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java +++ b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java @@ -28,15 +28,12 @@ import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; import org.springframework.cloud.sleuth.Span; import org.springframework.cloud.sleuth.Tracer; -import org.springframework.cloud.sleuth.internal.SpanNameUtil; import org.springframework.cloud.sleuth.propagation.Propagator; import org.springframework.cloud.stream.binder.BinderType; import org.springframework.cloud.stream.binder.BinderTypeRegistry; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; -import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -96,7 +93,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter final Propagator.Getter extractor; - final boolean integrationObjectSupportPresent; + final MessageSpanCustomizer messageSpanCustomizer; private final boolean hasDirectChannelClass; @@ -115,14 +112,13 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter public TracingChannelInterceptor(Tracer tracer, Propagator propagator, Propagator.Setter setter, Propagator.Getter getter, - Function remoteServiceNameMapper) { + Function remoteServiceNameMapper, MessageSpanCustomizer messageSpanCustomizer) { this.tracer = tracer; this.propagator = propagator; this.injector = setter; this.extractor = getter; this.remoteServiceNameMapper = remoteServiceNameMapper; - this.integrationObjectSupportPresent = ClassUtils - .isPresent("org.springframework.integration.context.IntegrationObjectSupport", null); + this.messageSpanCustomizer = messageSpanCustomizer; this.hasDirectChannelClass = ClassUtils.isPresent("org.springframework.integration.channel.DirectChannel", null); this.hasBinderTypeRegistry = ClassUtils.isPresent("org.springframework.cloud.stream.binder.BinderTypeRegistry", @@ -131,26 +127,6 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter ? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null; } - /** - * Use this to create a span for processing the given message. Note: the result has no - * name and is not started. This creates a child from identifiers extracted from the - * message headers, or a new span if one couldn't be extracted. - * @param message message to use for span creation - * @return span to be created - */ - public Span nextSpan(Message message) { - MessageHeaderAccessor headers = mutableHeaderAccessor(message); - Span result = this.propagator.extract(headers, this.extractor).start(); - headers.setImmutable(); - if (!result.isNoop()) { - addTags(result, null); - } - if (log.isDebugEnabled()) { - log.debug("Created a new span " + result); - } - return result; - } - /** * Starts and propagates {@link Span.Kind#PRODUCER} span for each message sent. */ @@ -166,8 +142,9 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter MessageHeaderAccessor headers = mutableHeaderAccessor(retrievedMessage); Span.Builder spanBuilder = this.propagator.extract(headers, this.extractor); MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.propagator.fields()); - spanBuilder.kind(Span.Kind.PRODUCER).name("send").remoteServiceName(toRemoteServiceName(headers)); - addTags(spanBuilder, channel); + spanBuilder = spanBuilder.kind(Span.Kind.PRODUCER); + spanBuilder = this.messageSpanCustomizer.customizeSend(spanBuilder, message, channel) + .remoteServiceName(toRemoteServiceName(headers)); Span span = spanBuilder.start(); if (log.isDebugEnabled()) { log.debug("Extracted result from headers " + span); @@ -302,8 +279,9 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter Span result) { Span.Builder builder = this.tracer.spanBuilder().setParent(result.context()); MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.propagator.fields()); - builder.kind(Span.Kind.CONSUMER).name("receive").remoteServiceName(toRemoteServiceName(headers)); - addTags(builder, channel); + builder = builder.kind(Span.Kind.CONSUMER); + builder = this.messageSpanCustomizer.customizeReceive(builder, message, channel); + builder = builder.remoteServiceName(toRemoteServiceName(headers)); return builder.start(); } @@ -333,7 +311,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter } Span consumerSpan = consumerSpan(message, channel, headers); // create and scope a span for the message processor - Span handle = this.tracer.nextSpan(consumerSpan).name("handle").start(); + Span handle = this.tracer.nextSpan(consumerSpan); + handle = this.messageSpanCustomizer.customizeHandle(handle, message, channel).start(); if (log.isDebugEnabled()) { log.debug("Created consumer span " + handle); } @@ -360,7 +339,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter // Start and finish a consumer span as we will immediately process it. consumerSpanBuilder.kind(Span.Kind.CONSUMER).start(); consumerSpanBuilder.remoteServiceName(REMOTE_SERVICE_NAME); - addTags(consumerSpanBuilder, channel); + consumerSpanBuilder = this.messageSpanCustomizer.customizeHandle(consumerSpanBuilder, message, channel); Span consumerSpan = consumerSpanBuilder.start(); consumerSpan.end(); return consumerSpan; @@ -377,45 +356,6 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter finishSpan(ex); } - /** - * When an upstream context was not present, lookup keys are unlikely added. - * @param result span to customize - * @param channel channel to which a message was sent - */ - void addTags(Span.Builder result, MessageChannel channel) { - // TODO topic etc - if (channel != null) { - result.tag("channel", messageChannelName(channel)); - } - } - - void addTags(Span result, MessageChannel channel) { - // TODO topic etc - if (channel != null) { - result.tag("channel", messageChannelName(channel)); - } - } - - private String channelName(MessageChannel channel) { - String name = null; - if (this.integrationObjectSupportPresent) { - 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 messageChannelName(MessageChannel channel) { - return SpanNameUtil.shorten(channelName(channel)); - } - void finishSpan(Exception error) { SpanAndScope spanAndScope = getSpanFromThreadLocal(); if (spanAndScope == null) { diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java index c74f549e4..4ef0bebde 100644 --- a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java @@ -75,7 +75,7 @@ public abstract class TracingChannelInterceptorTest implements TestTracingAwareS protected ChannelInterceptor interceptor = new TracingChannelInterceptor(tracerTest().tracing().tracer(), tracerTest().tracing().propagator(), MessageHeaderPropagation.INSTANCE, MessageHeaderPropagation.INSTANCE, - remoteServiceNameMapper(new SleuthMessagingProperties())); + remoteServiceNameMapper(new SleuthMessagingProperties()), new DefaultMessageSpanCustomizer()); protected TestSpanHandler spans = tracerTest().handler(); @@ -126,6 +126,23 @@ public abstract class TracingChannelInterceptorTest implements TestTracingAwareS assertThat(this.spans).extracting(FinishedSpan::getKind).contains(Span.Kind.CONSUMER, Span.Kind.PRODUCER); } + @Test + public void allowsSpanCustomization() { + this.interceptor = new TracingChannelInterceptor(tracerTest().tracing().tracer(), + tracerTest().tracing().propagator(), MessageHeaderPropagation.INSTANCE, + MessageHeaderPropagation.INSTANCE, remoteServiceNameMapper(new SleuthMessagingProperties()), + new MyMessageSpanCustomizer()); + + this.directChannel.addInterceptor(this.interceptor); + this.directChannel.subscribe(this.handler); + this.directChannel.send(MessageBuilder.withPayload("foo").build()); + + assertThat(this.spans.reportedSpans().stream().filter(s -> "changedHandle".equals(s.getName())).findFirst() + .map(s -> s.getTags().get("handleKey"))).isPresent().get().isEqualTo("handleValue"); + assertThat(this.spans.reportedSpans().stream().filter(s -> "changedSend".equals(s.getName())).findFirst() + .map(s -> s.getTags().get("sendKey"))).isPresent().get().isEqualTo("sendValue"); + } + @Test public void injectsProducerSpan_nativeHeaders() { this.channel.addInterceptor(producerSideOnly(this.interceptor)); @@ -405,3 +422,27 @@ class B3Context { } } + +// @formatter:off +// tag::message_span_customizer[] +class MyMessageSpanCustomizer extends DefaultMessageSpanCustomizer { + @Override + public Span customizeHandle(Span spanCustomizer, + Message message, MessageChannel messageChannel) { + return super.customizeHandle(spanCustomizer, message, messageChannel) + .name("changedHandle") + .tag("handleKey", "handleValue") + .tag("channelName", channelName(messageChannel)); + } + + @Override + public Span.Builder customizeSend(Span.Builder builder, + Message message, MessageChannel messageChannel) { + return super.customizeSend(builder, message, messageChannel) + .name("changedSend") + .tag("sendKey", "sendValue") + .tag("channelName", channelName(messageChannel)); + } +} +// end::message_span_customizer[] +// @formatter:on diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/otel/OtelTestTracing.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/otel/OtelTestTracing.java index 9ed46c129..2f909a591 100644 --- a/tests/common/src/main/java/org/springframework/cloud/sleuth/otel/OtelTestTracing.java +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/otel/OtelTestTracing.java @@ -20,12 +20,17 @@ import java.io.Closeable; import java.util.regex.Pattern; import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.metrics.MeterProvider; +import io.opentelemetry.api.trace.TracerProvider; import io.opentelemetry.context.propagation.ContextPropagators; import io.opentelemetry.context.propagation.DefaultContextPropagators; import io.opentelemetry.extension.trace.propagation.B3Propagator; import io.opentelemetry.sdk.trace.TracerSdkProvider; import io.opentelemetry.sdk.trace.config.TraceConfig; import io.opentelemetry.sdk.trace.samplers.Sampler; +import io.opentelemetry.sdk.trace.spi.TracerProviderFactorySdk; +import io.opentelemetry.spi.metrics.MeterProviderFactory; +import io.opentelemetry.spi.trace.TracerProviderFactory; import org.springframework.cloud.sleuth.CurrentTraceContext; import org.springframework.cloud.sleuth.SamplerFunction; @@ -36,6 +41,7 @@ import org.springframework.cloud.sleuth.http.HttpRequestParser; import org.springframework.cloud.sleuth.http.HttpServerHandler; import org.springframework.cloud.sleuth.otel.bridge.ArrayListSpanProcessor; import org.springframework.cloud.sleuth.otel.bridge.OtelAccessor; +import org.springframework.cloud.sleuth.otel.bridge.OtelOpenTelemetry; import org.springframework.cloud.sleuth.propagation.Propagator; import org.springframework.cloud.sleuth.test.TestSpanHandler; import org.springframework.cloud.sleuth.test.TestTracingAssertions; @@ -48,10 +54,12 @@ public class OtelTestTracing implements TracerAware, TestTracingAware, TestTraci ArrayListSpanProcessor spanProcessor = new ArrayListSpanProcessor(); - ContextPropagators defaultContextPropagators = OpenTelemetry.getGlobalPropagators(); - ContextPropagators contextPropagators = contextPropagators(); + OpenTelemetry openTelemetry = otel(); + + ContextPropagators defaultContextPropagators = OpenTelemetry.getGlobalPropagators(); + Sampler sampler = Sampler.alwaysOn(); HttpRequestParser clientRequestParser; @@ -60,14 +68,9 @@ public class OtelTestTracing implements TracerAware, TestTracingAware, TestTraci CurrentTraceContext currentTraceContext = OtelAccessor.currentTraceContext(publisher()); - // OtelBaggageManager otelBaggageManager = new - // OtelBaggageManager(this.currentTraceContext, - // new SleuthBaggageProperties(), publisher()); - io.opentelemetry.api.trace.Tracer otelTracer() { TracerSdkProvider provider = TracerSdkProvider.builder().build(); provider.addSpanProcessor(this.spanProcessor); - OpenTelemetry.setGlobalPropagators(this.contextPropagators); provider.updateActiveTraceConfig(TraceConfig.getDefault().toBuilder().setSampler(this.sampler).build()); return provider.get("org.springframework.cloud.sleuth"); } @@ -77,6 +80,32 @@ public class OtelTestTracing implements TracerAware, TestTracingAware, TestTraci .addTextMapPropagator(B3Propagator.builder().injectMultipleHeaders().build()).build(); } + OpenTelemetry otel() { + TracerProviderFactory providerFactory = otelTracerProviderFactory(); + MeterProviderFactory meterProviderFactory = otelMeterProviderFactory(); + OtelOpenTelemetry otelOpenTelemetry = new OtelOpenTelemetry(providerFactory, meterProviderFactory, + otelTracerProvider(providerFactory), otelMeterProvider(meterProviderFactory), this.contextPropagators); + OpenTelemetry.set(otelOpenTelemetry); + OpenTelemetry.setGlobalPropagators(contextPropagators); + return otelOpenTelemetry; + } + + TracerProviderFactory otelTracerProviderFactory() { + return new TracerProviderFactorySdk(); + } + + TracerProvider otelTracerProvider(TracerProviderFactory tracerProviderFactory) { + return tracerProviderFactory.create(); + } + + MeterProviderFactory otelMeterProviderFactory() { + return OpenTelemetry::getGlobalMeterProvider; + } + + MeterProvider otelMeterProvider(MeterProviderFactory meterProviderFactory) { + return meterProviderFactory.create(); + } + private void reset() { this.contextPropagators = contextPropagators(); this.tracer = otelTracer();