Adds MessageSpanCustomizer to customize message spans

fixes gh-1772

docs/src/main/asciidoc/_configprops.adoc

@@ -6,6 +6,11 @@
 |spring.sleuth.async.enabled | true | Enable instrumenting async related components so that the tracing information is passed between threads.
 |spring.sleuth.async.ignored-beans |  | List of {@link java.util.concurrent.Executor} bean names that should be ignored and not wrapped in a trace representation.
 |spring.sleuth.baggage-keys |  | List of baggage key names that should be propagated out of process. These keys will be prefixed with `baggage` before the actual key. This property is set in order to be backward compatible with previous Sleuth versions. @see brave.propagation.ExtraFieldPropagation.FactoryBuilder#addPrefixedFields(String, java.util.Collection)
+|spring.sleuth.baggage.correlation-enabled | true | Adds a {@link CorrelationScopeDecorator} to put baggage values into the correlation context.
+|spring.sleuth.baggage.correlation-fields |  | A list of {@link BaggageField#name() fields} to add to correlation (MDC) context. @see CorrelationScopeConfig.SingleCorrelationField#create(BaggageField)
+|spring.sleuth.baggage.local-fields |  | Same as {@link #remoteFields} except that this field is not propagated to remote services. @see BaggagePropagationConfig.SingleBaggageField#local(BaggageField)
+|spring.sleuth.baggage.remote-fields |  | List of fields that are referenced the same in-process as it is on the wire. For example, the field "x-vcap-request-id" would be set as-is including the prefix. @see BaggagePropagationConfig.SingleBaggageField#remote(BaggageField) @see BaggagePropagationConfig.SingleBaggageField.Builder#addKeyName(String)
+|spring.sleuth.baggage.tag-fields |  | A list of {@link BaggageField#name() fields} to tag into the span. @see Tags#BAGGAGE_FIELD
 |spring.sleuth.circuitbreaker.enabled | true | Enable Spring Cloud CircuitBreaker instrumentation.
 |spring.sleuth.enabled | true | 
 |spring.sleuth.feign.enabled | true | Enable span information propagation when using Feign.

docs/src/main/asciidoc/spring-cloud-sleuth.adoc

@@ -946,6 +946,18 @@ include::{project-root}/spring-cloud-sleuth-core/src/test/java/org/springframewo
 
 For more, see https://github.com/openzipkin/brave/tree/master/instrumentation/messaging#sampling-policy
 
+==== 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::{project-root}/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java[tags=message_span_customizer,indent=2]
+----
+
+
 === RPC
 
 Sleuth automatically configures the `RpcTracing` bean which serves as a

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/DefaultMessageSpanCustomizer.java

@@ -0,0 +1,105 @@
+/*
+ * 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 brave.SpanCustomizer;
+
+import org.springframework.cloud.sleuth.util.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 SpanCustomizer customizeHandle(SpanCustomizer spanCustomizer,
+			Message<?> message, @Nullable MessageChannel messageChannel) {
+		spanCustomizer.name("handle");
+		addTags(spanCustomizer, messageChannel);
+		return spanCustomizer;
+	}
+
+	@Override
+	public SpanCustomizer customizeReceive(SpanCustomizer spanCustomizer,
+			Message<?> message, @Nullable MessageChannel messageChannel) {
+		spanCustomizer.name("receive");
+		addTags(spanCustomizer, messageChannel);
+		return spanCustomizer;
+	}
+
+	@Override
+	public SpanCustomizer customizeSend(SpanCustomizer spanCustomizer, Message<?> message,
+			@Nullable MessageChannel messageChannel) {
+		spanCustomizer.name("send");
+		addTags(spanCustomizer, messageChannel);
+		return spanCustomizer;
+	}
+
+	/**
+	 * 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
+	 */
+	private void addTags(SpanCustomizer result, MessageChannel channel) {
+		if (channel != null) {
+			result.tag("channel", messageChannelName(channel));
+		}
+	}
+
+}

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/MessageSpanCustomizer.java

@@ -0,0 +1,42 @@
+/*
+ * 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 brave.SpanCustomizer;
+
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * Allows customization of messaging spans.
+ *
+ * @author Marcin Grzejszczak
+ * @since 2.2.7
+ */
+public interface MessageSpanCustomizer {
+
+	SpanCustomizer customizeHandle(SpanCustomizer spanCustomizer, Message<?> message,
+			@Nullable MessageChannel messageChannel);
+
+	SpanCustomizer customizeReceive(SpanCustomizer spanCustomizer, Message<?> message,
+			@Nullable MessageChannel messageChannel);
+
+	SpanCustomizer customizeSend(SpanCustomizer spanCustomizer, Message<?> message,
+			@Nullable MessageChannel messageChannel);
+
+}

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceSpringIntegrationAutoConfiguration.java

@@ -22,6 +22,7 @@ import brave.propagation.Propagation;
 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.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.sleuth.autoconfig.TraceAutoConfiguration;
@@ -62,13 +63,21 @@ public class TraceSpringIntegrationAutoConfiguration {
 		return wrapper;
 	}
 
+	@Bean
+	@ConditionalOnMissingBean
+	MessageSpanCustomizer defaultMessageSpanCustomizer() {
+		return new DefaultMessageSpanCustomizer();
+	}
+
 	@Bean
 	TracingChannelInterceptor traceChannelInterceptor(Tracing tracing,
 			SleuthMessagingProperties properties,
 			Propagation.Setter<MessageHeaderAccessor, String> traceMessagePropagationSetter,
-			Propagation.Getter<MessageHeaderAccessor, String> traceMessagePropagationGetter) {
+			Propagation.Getter<MessageHeaderAccessor, String> traceMessagePropagationGetter,
+			MessageSpanCustomizer messageSpanCustomizer) {
 		return new TracingChannelInterceptor(tracing, properties,
-				traceMessagePropagationSetter, traceMessagePropagationGetter);
+				traceMessagePropagationSetter, traceMessagePropagationGetter,
+				messageSpanCustomizer);
 	}
 
 }

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java

@@ -113,6 +113,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 
 	final SleuthMessagingProperties properties;
 
+	final MessageSpanCustomizer messageSpanCustomizer;
+
 	final boolean integrationObjectSupportPresent;
 
 	private final boolean hasDirectChannelClass;
@@ -125,17 +127,20 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 	private ApplicationContext applicationContext;
 
 	@Autowired
-	TracingChannelInterceptor(Tracing tracing, SleuthMessagingProperties properties) {
+	TracingChannelInterceptor(Tracing tracing, SleuthMessagingProperties properties,
+			MessageSpanCustomizer messageSpanCustomizer) {
 		this(tracing, properties, MessageHeaderPropagation.INSTANCE,
-				MessageHeaderPropagation.INSTANCE);
+				MessageHeaderPropagation.INSTANCE, messageSpanCustomizer);
 	}
 
 	TracingChannelInterceptor(Tracing tracing, SleuthMessagingProperties properties,
 			Propagation.Setter<MessageHeaderAccessor, String> setter,
-			Propagation.Getter<MessageHeaderAccessor, String> getter) {
+			Propagation.Getter<MessageHeaderAccessor, String> getter,
+			MessageSpanCustomizer messageSpanCustomizer) {
 		this.tracing = tracing;
 		this.properties = properties;
 		this.tracer = tracing.tracer();
+		this.messageSpanCustomizer = messageSpanCustomizer;
 		this.threadLocalSpan = ThreadLocalSpan.create(this.tracer);
 		this.injector = tracing.propagation().injector(setter);
 		this.extractor = tracing.propagation().extractor(getter);
@@ -151,8 +156,9 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 	}
 
 	public static TracingChannelInterceptor create(Tracing tracing,
-			SleuthMessagingProperties properties) {
-		return new TracingChannelInterceptor(tracing, properties);
+			SleuthMessagingProperties properties,
+			MessageSpanCustomizer messageSpanCustomizer) {
+		return new TracingChannelInterceptor(tracing, properties, messageSpanCustomizer);
 	}
 
 	/**
@@ -164,7 +170,9 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 	 * span if one couldn't be extracted.
 	 * @param message message to use for span creation
 	 * @return span to be created
+	 * @deprecated scheduled for removal in 3.0.0
 	 */
+	@Deprecated
 	public Span nextSpan(Message<?> message) {
 		MessageHeaderAccessor headers = mutableHeaderAccessor(message);
 		TraceContextOrSamplingFlags extracted = this.extractor.extract(headers);
@@ -195,9 +203,9 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 				this.tracing.propagation().keys());
 		this.injector.inject(span.context(), headers);
 		if (!span.isNoop()) {
-			span.kind(Span.Kind.PRODUCER).name("send").start();
+			span.kind(Span.Kind.PRODUCER).start();
+			this.messageSpanCustomizer.customizeSend(span, message, channel);
 			span.remoteServiceName(toRemoteServiceName(headers));
-			addTags(message, span, channel);
 		}
 		if (log.isDebugEnabled()) {
 			log.debug("Created a new span in pre send" + span);
@@ -311,9 +319,9 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 				this.tracing.propagation().keys());
 		this.injector.inject(span.context(), headers);
 		if (!span.isNoop()) {
-			span.kind(Span.Kind.CONSUMER).name("receive").start();
+			span.kind(Span.Kind.CONSUMER).start();
+			this.messageSpanCustomizer.customizeReceive(span, message, channel);
 			span.remoteServiceName(toRemoteServiceName(headers));
-			addTags(message, span, channel);
 		}
 		if (log.isDebugEnabled()) {
 			log.debug("Created a new span in post receive " + span);
@@ -361,9 +369,9 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 			consumerSpan.finish();
 		}
 		// create and scope a span for the message processor
-		this.threadLocalSpan
-				.next(TraceContextOrSamplingFlags.create(consumerSpan.context()))
-				.name("handle").start();
+		Span span = this.threadLocalSpan
+				.next(TraceContextOrSamplingFlags.create(consumerSpan.context())).start();
+		this.messageSpanCustomizer.customizeHandle(span, message, channel);
 		// 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.
@@ -393,19 +401,16 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 		finishSpan(ex);
 	}
 
-	/**
-	 * When an upstream context was not present, lookup keys are unlikely added.
-	 * @param message a message to append tags to
-	 * @param result span to customize
-	 * @param channel channel to which a message was sent
-	 */
-	void addTags(Message<?> message, SpanCustomizer result, MessageChannel channel) {
+	@Deprecated
+	private void addTags(Message<?> message, SpanCustomizer result,
+			MessageChannel channel) {
 		// TODO topic etc
 		if (channel != null) {
 			result.tag("channel", messageChannelName(channel));
 		}
 	}
 
+	@Deprecated
 	private String channelName(MessageChannel channel) {
 		String name = null;
 		if (this.integrationObjectSupportPresent) {
@@ -422,6 +427,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 		return name;
 	}
 
+	@Deprecated
 	private String messageChannelName(MessageChannel channel) {
 		return SpanNameUtil.shorten(channelName(channel));
 	}

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/websocket/TraceWebSocketAutoConfiguration.java

@@ -21,9 +21,13 @@ import brave.Tracing;
 import org.springframework.beans.factory.annotation.Autowired;
 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.instrument.messaging.DefaultMessageSpanCustomizer;
+import org.springframework.cloud.sleuth.instrument.messaging.MessageSpanCustomizer;
 import org.springframework.cloud.sleuth.instrument.messaging.SleuthMessagingProperties;
 import org.springframework.cloud.sleuth.instrument.messaging.TracingChannelInterceptor;
+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;
@@ -56,6 +60,15 @@ public class TraceWebSocketAutoConfiguration
 	@Autowired
 	SleuthMessagingProperties properties;
 
+	@Autowired
+	MessageSpanCustomizer messageSpanCustomizer;
+
+	@Bean
+	@ConditionalOnMissingBean
+	MessageSpanCustomizer defaultMessageSpanCustomizer() {
+		return new DefaultMessageSpanCustomizer();
+	}
+
 	@Override
 	public void registerStompEndpoints(StompEndpointRegistry registry) {
 		// The user must register their own endpoints
@@ -63,20 +76,20 @@ public class TraceWebSocketAutoConfiguration
 
 	@Override
 	public void configureMessageBroker(MessageBrokerRegistry registry) {
-		registry.configureBrokerChannel().setInterceptors(
-				TracingChannelInterceptor.create(this.tracing, this.properties));
+		registry.configureBrokerChannel().setInterceptors(TracingChannelInterceptor
+				.create(this.tracing, this.properties, this.messageSpanCustomizer));
 	}
 
 	@Override
 	public void configureClientOutboundChannel(ChannelRegistration registration) {
-		registration.setInterceptors(
-				TracingChannelInterceptor.create(this.tracing, this.properties));
+		registration.setInterceptors(TracingChannelInterceptor.create(this.tracing,
+				this.properties, this.messageSpanCustomizer));
 	}
 
 	@Override
 	public void configureClientInboundChannel(ChannelRegistration registration) {
-		registration.setInterceptors(
-				TracingChannelInterceptor.create(this.tracing, this.properties));
+		registration.setInterceptors(TracingChannelInterceptor.create(this.tracing,
+				this.properties, this.messageSpanCustomizer));
 	}
 
 }

spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java

@@ -23,6 +23,7 @@ import java.util.List;
 import java.util.Map;
 
 import brave.Span;
+import brave.SpanCustomizer;
 import brave.Tracing;
 import brave.handler.MutableSpan;
 import brave.propagation.StrictCurrentTraceContext;
@@ -60,7 +61,7 @@ public class TracingChannelInterceptorTest {
 			.addSpanHandler(this.spans).build();
 
 	ChannelInterceptor interceptor = TracingChannelInterceptor.create(tracing,
-			new SleuthMessagingProperties());
+			new SleuthMessagingProperties(), new DefaultMessageSpanCustomizer());
 
 	QueueChannel channel = new QueueChannel();
 
@@ -114,6 +115,24 @@ public class TracingChannelInterceptorTest {
 				Span.Kind.PRODUCER);
 	}
 
+	@Test
+	public void allowsSpanCustomization() {
+		this.interceptor = TracingChannelInterceptor.create(tracing,
+				new SleuthMessagingProperties(), new MyMessageSpanCustomizer());
+
+		this.directChannel.addInterceptor(this.interceptor);
+		this.directChannel.subscribe(this.handler);
+		this.directChannel.send(MessageBuilder.withPayload("foo").build());
+
+		assertThat(
+				this.spans.spans().stream().filter(s -> "changedHandle".equals(s.name()))
+						.findFirst().map(s -> s.tag("handleKey"))).isPresent().get()
+								.isEqualTo("handleValue");
+		assertThat(this.spans.spans().stream().filter(s -> "changedSend".equals(s.name()))
+				.findFirst().map(s -> s.tag("sendKey"))).isPresent().get()
+						.isEqualTo("sendValue");
+	}
+
 	@Test
 	public void injectsProducerSpan_nativeHeaders() {
 		this.channel.addInterceptor(producerSideOnly(this.interceptor));
@@ -436,3 +455,27 @@ public class TracingChannelInterceptorTest {
 	}
 
 }
+
+// @formatter:off
+// tag::message_span_customizer[]
+class MyMessageSpanCustomizer extends DefaultMessageSpanCustomizer {
+	@Override
+	public SpanCustomizer customizeHandle(SpanCustomizer spanCustomizer,
+			Message<?> message, MessageChannel messageChannel) {
+		return super.customizeHandle(spanCustomizer, message, messageChannel)
+				.name("changedHandle")
+				.tag("handleKey", "handleValue")
+				.tag("channelName", channelName(messageChannel));
+	}
+
+	@Override
+	public SpanCustomizer customizeSend(SpanCustomizer spanCustomizer,
+			Message<?> message, MessageChannel messageChannel) {
+		return super.customizeSend(spanCustomizer, message, messageChannel)
+				.name("changedSend")
+				.tag("sendKey", "sendValue")
+				.tag("channelName", channelName(messageChannel));
+	}
+}
+// end::message_span_customizer[]
+// @formatter:on