Merged 2.2.x

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<MessageHeaderAccessor>` - for writing headers to the message
 * `Propagator.Getter<MessageHeaderAccessor>` - 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
 

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<MessageHeaderAccessor> traceMessagePropagationSetter,
 			Propagator.Getter<MessageHeaderAccessor> traceMessagePropagationGetter,
-			SleuthMessagingProperties properties) {
+			SleuthMessagingProperties properties, MessageSpanCustomizer messageSpanCustomizer) {
 		return new TracingChannelInterceptor(tracer, propagator, traceMessagePropagationSetter,
-				traceMessagePropagationGetter, remoteServiceNameMapper(properties));
+				traceMessagePropagationGetter, remoteServiceNameMapper(properties), messageSpanCustomizer);
 	}
 
 	static Function<String, String> remoteServiceNameMapper(SleuthMessagingProperties properties) {

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));
 	}
 
 }

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));
+		}
+	}
+
+}

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);
+
+}

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<MessageHeaderAccessor> 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<MessageHeaderAccessor> setter, Propagator.Getter<MessageHeaderAccessor> getter,
-			Function<String, String> remoteServiceNameMapper) {
+			Function<String, String> 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) {

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

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();