Add some infrastructure for Observation (#3879)

* Add some infrastructure for Observation

* Populate an `ObservationRegistry` bean from the `IntegrationManagementConfigurer`
into all the `IntegrationManagement` components
* Introduce `MessageReceiverContext` and `MessageSenderContext` for easier usage
in the target code
* Implement `Observation` handling in the `AbstractMessageHandler`
* Modify `ObservationPropagationChannelInterceptorTests` for new `MessageSenderContext`
* Use `BridgeHandler` to ensure that `Observation` is propagated and handled properly
* Verify that tags from the `AbstractMessageHandler` are preset on the consumer span

* * Add a `DocumentedObservation` infrastructure

* * Add `Timer` verification to the propagation test

* * Update to the latest Observation API

* * Add custom observation convention support for the `AbstractMessageHandler`
* Use more meaningful prefix for Spring Integration tags

* * Move singleton instance for `DefaultMessageReceiverObservationConvention`
into `DefaultMessageReceiverObservationConvention` per se as an `INSTANCE` constant
* Use `MeterRegistryAssert` in the `ObservationPropagationChannelInterceptorTests`
to verify meters emitted

* * And an integration test with Zipkin based on the `SampleTestRunner`

build.gradle

@@ -528,7 +528,12 @@ project('spring-integration-core') {
         testImplementation "org.aspectj:aspectjweaver:$aspectjVersion"
         testImplementation "org.hamcrest:hamcrest-core:$hamcrestVersion"
         testImplementation 'io.micrometer:micrometer-observation-test'
-        testImplementation 'io.micrometer:micrometer-tracing-test'
+        testImplementation ('io.micrometer:micrometer-tracing-integration-test') {
+            exclude group: 'io.opentelemetry'
+            exclude group: 'com.wavefront'
+            exclude group: 'io.micrometer', module: 'micrometer-tracing-bridge-otel'
+
+        }
     }
 
     dokkaHtmlPartial {

spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-2022 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.
@@ -31,6 +31,8 @@ import org.springframework.core.type.AnnotationMetadata;
 import org.springframework.integration.support.management.metrics.MetricsCaptor;
 import org.springframework.util.Assert;
 
+import io.micrometer.observation.ObservationRegistry;
+
 /**
  * {@code @Configuration} class that registers a {@link IntegrationManagementConfigurer} bean.
  *
@@ -64,12 +66,16 @@ public class IntegrationManagementConfiguration implements ImportAware, Environm
 
 	@Bean(name = IntegrationManagementConfigurer.MANAGEMENT_CONFIGURER_NAME)
 	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
-	public IntegrationManagementConfigurer managementConfigurer(ObjectProvider<MetricsCaptor> metricsCaptorProvider) {
+	public IntegrationManagementConfigurer managementConfigurer(
+			ObjectProvider<MetricsCaptor> metricsCaptorProvider,
+			ObjectProvider<ObservationRegistry> observationRegistryProvider) {
+
 		IntegrationManagementConfigurer configurer = new IntegrationManagementConfigurer();
 		configurer.setDefaultLoggingEnabled(
 				Boolean.parseBoolean(this.environment.resolvePlaceholders(
 						(String) this.attributes.get("defaultLoggingEnabled"))));
 		configurer.setMetricsCaptorProvider(metricsCaptorProvider);
+		configurer.setObservationRegistry(observationRegistryProvider);
 		return configurer;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-2022 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.
@@ -38,10 +38,12 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
 
+import io.micrometer.observation.ObservationRegistry;
+
 
 /**
  * Configures beans that implement {@link IntegrationManagement}.
- * Configures counts, stats, logging for all (or selected) components.
+ * Configures logging, {@link MetricsCaptor} and {@link ObservationRegistry} for all (or selected) components.
  *
  * @author Gary Russell
  * @author Artem Bilan
@@ -74,6 +76,10 @@ public class IntegrationManagementConfigurer
 
 	private ObjectProvider<MetricsCaptor> metricsCaptorProvider;
 
+	private ObservationRegistry observationRegistry;
+
+	private ObjectProvider<ObservationRegistry> observationRegistryProvider;
+
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
 		this.applicationContext = applicationContext;
@@ -95,8 +101,8 @@ public class IntegrationManagementConfigurer
 	 * {@link org.apache.commons.logging.Log#isDebugEnabled()} can be quite expensive
 	 * and account for an inordinate amount of CPU time.
 	 * <p>
-	 * Set this to false to disable logging by default in all framework components that implement
-	 * {@link IntegrationManagement} (channels, message handlers etc). This turns off logging such as
+	 * Set this to 'false' to disable logging by default in all framework components that implement
+	 * {@link IntegrationManagement} (channels, message handlers etc.) This turns off logging such as
 	 * "PreSend on channel", "Received message" etc.
 	 * <p>
 	 * After the context is initialized, individual components can have their setting changed by invoking
@@ -115,14 +121,21 @@ public class IntegrationManagementConfigurer
 		this.metricsCaptorProvider = metricsCaptorProvider;
 	}
 
-	@Nullable
-	MetricsCaptor obtainMetricsCaptor() {
-		if (this.metricsCaptor == null && this.metricsCaptorProvider != null) {
-			this.metricsCaptor = this.metricsCaptorProvider.getIfUnique();
-		}
-		return this.metricsCaptor;
+	/**
+	 * Set an {@link ObservationRegistry} to populate to the {@link IntegrationManagement} components
+	 * in the application context.
+	 * @param observationRegistry the {@link ObservationRegistry} to use.
+	 * @since 6.0
+	 */
+	public void setObservationRegistry(@Nullable ObservationRegistry observationRegistry) {
+		this.observationRegistry = observationRegistry;
 	}
 
+	void setObservationRegistry(ObjectProvider<ObservationRegistry> observationRegistryProvider) {
+		this.observationRegistryProvider = observationRegistryProvider;
+	}
+
+
 	@Override
 	public void afterSingletonsInstantiated() {
 		Assert.state(this.applicationContext != null, "'applicationContext' must not be null");
@@ -133,15 +146,29 @@ public class IntegrationManagementConfigurer
 			registerComponentGauges();
 		}
 
-		for (IntegrationManagement integrationManagement :
-				this.applicationContext.getBeansOfType(IntegrationManagement.class).values()) {
+		setupObservationRegistry();
 
-			enhanceIntegrationManagement(integrationManagement);
-		}
+		this.applicationContext.getBeansOfType(IntegrationManagement.class).values()
+				.forEach(this::enhanceIntegrationManagement);
 
 		this.singletonsInstantiated = true;
 	}
 
+	@Nullable
+	private MetricsCaptor obtainMetricsCaptor() {
+		if (this.metricsCaptor == null && this.metricsCaptorProvider != null) {
+			this.metricsCaptor = this.metricsCaptorProvider.getIfUnique();
+		}
+		return this.metricsCaptor;
+	}
+
+	@Nullable
+	private void setupObservationRegistry() {
+		if (this.observationRegistry == null && this.observationRegistryProvider != null) {
+			this.observationRegistry = this.observationRegistryProvider.getIfUnique();
+		}
+	}
+
 	private void registerComponentGauges() {
 		this.gauges.add(
 				this.metricsCaptor.gaugeBuilder("spring.integration.channels", this,
@@ -169,17 +196,21 @@ public class IntegrationManagementConfigurer
 		if (this.metricsCaptor != null) {
 			integrationManagement.registerMetricsCaptor(this.metricsCaptor);
 		}
+		if (this.observationRegistry != null) {
+			integrationManagement.registerObservationRegistry(this.observationRegistry);
+		}
 	}
 
 	@Override
 	public Object postProcessAfterInitialization(Object bean, String name) throws BeansException {
-		if (this.singletonsInstantiated && bean instanceof IntegrationManagement) {
-			enhanceIntegrationManagement((IntegrationManagement) bean);
+		if (this.singletonsInstantiated && bean instanceof IntegrationManagement integrationManagement) {
+			enhanceIntegrationManagement(integrationManagement);
 		}
 		return bean;
 	}
 
-	@Override public void onApplicationEvent(ContextClosedEvent event) {
+	@Override
+	public void onApplicationEvent(ContextClosedEvent event) {
 		if (event.getApplicationContext().equals(this.applicationContext)) {
 			this.gauges.forEach(MeterFacade::remove);
 			this.gauges.clear();

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java

@@ -21,11 +21,17 @@ import org.reactivestreams.Subscription;
 import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.support.management.metrics.MetricsCaptor;
 import org.springframework.integration.support.management.metrics.SampleFacade;
+import org.springframework.integration.support.management.observation.DefaultMessageReceiverObservationConvention;
+import org.springframework.integration.support.management.observation.IntegrationObservation;
+import org.springframework.integration.support.management.observation.MessageReceiverContext;
+import org.springframework.integration.support.management.observation.MessageReceiverObservationConvention;
 import org.springframework.integration.support.utils.IntegrationUtils;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
 
+import io.micrometer.observation.ObservationRegistry;
 import reactor.core.CoreSubscriber;
 
 /**
@@ -37,21 +43,49 @@ import reactor.core.CoreSubscriber;
 public abstract class AbstractMessageHandler extends MessageHandlerSupport
 		implements MessageHandler, CoreSubscriber<Message<?>> {
 
+	@Nullable
+	private MessageReceiverObservationConvention observationConvention;
+
+	/**
+	 * Set a custom {@link MessageReceiverObservationConvention} for {@link IntegrationObservation#HANDLER}.
+	 * Ignored if an {@link ObservationRegistry} is not configured for this component.
+	 * @param observationConvention the {@link MessageReceiverObservationConvention} to use.
+	 * @since 6.0
+	 */
+	public void setObservationConvention(@Nullable MessageReceiverObservationConvention observationConvention) {
+		this.observationConvention = observationConvention;
+	}
+
 	@Override // NOSONAR
 	public void handleMessage(Message<?> message) {
 		Assert.notNull(message, "Message must not be null");
-		if (isLoggingEnabled() && this.logger.isDebugEnabled()) {
-			this.logger.debug(this + " received message: " + message);
+		if (isLoggingEnabled()) {
+			this.logger.debug(() -> this + " received message: " + message);
 		}
-		MetricsCaptor metricsCaptor = getMetricsCaptor();
-		if (metricsCaptor != null) {
-			handleWithMetrics(message, metricsCaptor);
+		ObservationRegistry observationRegistry = getObservationRegistry();
+		if (observationRegistry != null) {
+			handleWithObservation(message, observationRegistry);
 		}
 		else {
-			doHandleMessage(message);
+			MetricsCaptor metricsCaptor = getMetricsCaptor();
+			if (metricsCaptor != null) {
+				handleWithMetrics(message, metricsCaptor);
+			}
+			else {
+				doHandleMessage(message);
+			}
 		}
 	}
 
+	private void handleWithObservation(Message<?> message, ObservationRegistry observationRegistry) {
+		IntegrationObservation.HANDLER.observation(
+						this.observationConvention,
+						DefaultMessageReceiverObservationConvention.INSTANCE,
+						new MessageReceiverContext(message, getComponentName()),
+						observationRegistry)
+				.observe(() -> doHandleMessage(message));
+	}
+
 	private void handleWithMetrics(Message<?> message, MetricsCaptor metricsCaptor) {
 		SampleFacade sample = metricsCaptor.start();
 		try {

spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-2022 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.
@@ -31,6 +31,8 @@ import org.springframework.integration.support.management.metrics.MeterFacade;
 import org.springframework.integration.support.management.metrics.MetricsCaptor;
 import org.springframework.integration.support.management.metrics.TimerFacade;
 
+import io.micrometer.observation.ObservationRegistry;
+
 /**
  * Base class for Message handling components that provides basic validation and error
  * handling capabilities. Asserts that the incoming Message is not null and that it does
@@ -61,6 +63,8 @@ public abstract class MessageHandlerSupport extends IntegrationObjectSupport
 
 	private MetricsCaptor metricsCaptor;
 
+	private ObservationRegistry observationRegistry;
+
 	private int order = Ordered.LOWEST_PRECEDENCE;
 
 	private String managedName;
@@ -89,6 +93,15 @@ public abstract class MessageHandlerSupport extends IntegrationObjectSupport
 		return this.metricsCaptor;
 	}
 
+	@Override
+	public void registerObservationRegistry(ObservationRegistry observationRegistry) {
+		this.observationRegistry = observationRegistry;
+	}
+
+	protected ObservationRegistry getObservationRegistry() {
+		return this.observationRegistry;
+	}
+
 	@Override
 	public void setOrder(int order) {
 		this.order = order;

spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagement.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2021 the original author or authors.
+ * Copyright 2015-2022 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.
@@ -22,10 +22,14 @@ import org.springframework.integration.support.management.metrics.MetricsCaptor;
 import org.springframework.jmx.export.annotation.ManagedAttribute;
 import org.springframework.lang.Nullable;
 
+import io.micrometer.observation.ObservationRegistry;
+
 /**
  * Base interface for Integration managed components.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.2
  *
  */
@@ -39,7 +43,7 @@ public interface IntegrationManagement extends NamedComponent, DisposableBean {
 
 	/**
 	 * Enable logging or not.
-	 * @param enabled dalse to disable.
+	 * @param enabled false to disable.
 	 */
 	@ManagedAttribute(description = "Use to disable debug logging during normal message flow")
 	default void setLoggingEnabled(boolean enabled) {
@@ -80,13 +84,28 @@ public interface IntegrationManagement extends NamedComponent, DisposableBean {
 
 	/**
 	 * Inject a {@link MetricsCaptor}.
+	 * Ignored if {@link ObservationRegistry} is provided.
 	 * @param captor the captor.
 	 * @since 5.0.4
+	 * @see #registerObservationRegistry(ObservationRegistry)
 	 */
 	default void registerMetricsCaptor(MetricsCaptor captor) {
 		// no op
 	}
 
+	/**
+	 * Inject an {@link ObservationRegistry}.
+	 * If provided, the {@link MetricsCaptor} is ignored.
+	 * The meters capturing has to be configured as an {@link io.micrometer.observation.ObservationHandler}
+	 * on the provided {@link ObservationRegistry}.
+	 * @param observationRegistry the {@link ObservationRegistry} to expose observations from the component.
+	 * @since 6.0
+	 * @see #registerMetricsCaptor(MetricsCaptor)
+	 */
+	default void registerObservationRegistry(ObservationRegistry observationRegistry) {
+		// no op
+	}
+
 	@Override
 	default void destroy() {
 		// no op

spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/DefaultMessageReceiverObservationConvention.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2022 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.integration.support.management.observation;
+
+import io.micrometer.common.KeyValues;
+
+/**
+ * A default {@link MessageReceiverObservationConvention} implementation.
+ * Provides low cardinalities as a {@link IntegrationObservation.HandlerTags} values.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class DefaultMessageReceiverObservationConvention implements MessageReceiverObservationConvention {
+
+	/**
+	 * A shared singleton instance for {@link DefaultMessageReceiverObservationConvention}.
+	 */
+	public static final DefaultMessageReceiverObservationConvention INSTANCE =
+			new DefaultMessageReceiverObservationConvention();
+
+	@Override
+	public KeyValues getLowCardinalityKeyValues(MessageReceiverContext context) {
+		return KeyValues.of(
+				IntegrationObservation.HandlerTags.COMPONENT_NAME.withValue(context.getHandlerName()),
+				IntegrationObservation.HandlerTags.COMPONENT_TYPE.withValue("handler"));
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/IntegrationObservation.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2022 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.integration.support.management.observation;
+
+import io.micrometer.common.docs.KeyName;
+import io.micrometer.observation.docs.DocumentedObservation;
+
+/**
+ * The {@link DocumentedObservation} implementation for Spring Integration infrastructure.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public enum IntegrationObservation implements DocumentedObservation {
+
+	/**
+	 * Observation for message handlers.
+	 */
+	HANDLER {
+		@Override
+		public String getName() {
+			return "spring.integration.handler";
+		}
+
+		@Override
+		public String getPrefix() {
+			return "spring.integration.";
+		}
+
+		@Override
+		public Class<DefaultMessageReceiverObservationConvention> getDefaultConvention() {
+			return DefaultMessageReceiverObservationConvention.class;
+		}
+
+		@Override
+		public KeyName[] getLowCardinalityKeyNames() {
+			return HandlerTags.values();
+		}
+
+	};
+
+	/**
+	 * Key names for message handler observations.
+	 */
+	public enum HandlerTags implements KeyName {
+
+		/**
+		 * Name of the message handler component.
+		 */
+		COMPONENT_NAME {
+			@Override
+			public String asString() {
+				return "spring.integration.name";
+			}
+
+		},
+
+		/**
+		 * Type of the component - 'handler'.
+		 */
+		COMPONENT_TYPE {
+			@Override
+			public String asString() {
+				return "spring.integration.type";
+			}
+
+		}
+
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageReceiverContext.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2022 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.integration.support.management.observation;
+
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+
+import io.micrometer.observation.transport.ReceiverContext;
+
+/**
+ * The {@link ReceiverContext} extension for {@link Message} context.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class MessageReceiverContext extends ReceiverContext<Message<?>> {
+
+	private final Message<?> message;
+
+	private final String handlerName;
+
+	public MessageReceiverContext(Message<?> message, @Nullable String handlerName) {
+		super((carrier, key) -> carrier.getHeaders().get(key, String.class));
+		this.message = message;
+		this.handlerName = handlerName != null ? handlerName : "unknown";
+	}
+
+	@Override
+	public Message<?> getCarrier() {
+		return this.message;
+	}
+
+	public String getHandlerName() {
+		return this.handlerName;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageReceiverObservationConvention.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2022 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.integration.support.management.observation;
+
+import org.springframework.messaging.Message;
+
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationConvention;
+import io.micrometer.observation.transport.ReceiverContext;
+
+/**
+ * The {@link ReceiverContext} extension for {@link Message} context.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public interface MessageReceiverObservationConvention
+		extends ObservationConvention<MessageReceiverContext> {
+
+	@Override
+	default boolean supportsContext(Observation.Context context) {
+		return context instanceof MessageReceiverContext;
+	}
+
+	@Override
+	default String getContextualName(MessageReceiverContext context) {
+		return context.getHandlerName() + " receive";
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageSenderContext.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2022 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.integration.support.management.observation;
+
+import org.springframework.integration.support.MutableMessage;
+import org.springframework.messaging.Message;
+
+import io.micrometer.observation.transport.SenderContext;
+
+/**
+ * The {@link SenderContext} extension for {@link Message} context.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class MessageSenderContext extends SenderContext<MutableMessage<?>> {
+
+	public MessageSenderContext(MutableMessage<?> message) {
+		super((carrier, key, value) -> carrier.getHeaders().put(key, value));
+		setCarrier(message);
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptorTests.java

@@ -21,12 +21,12 @@ import static org.assertj.core.api.Assertions.assertThat;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
-import java.util.Objects;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -40,22 +40,30 @@ import org.springframework.integration.channel.ExecutorChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.config.GlobalChannelInterceptor;
+import org.springframework.integration.handler.BridgeHandler;
+import org.springframework.integration.support.MutableMessage;
+import org.springframework.integration.support.MutableMessageBuilder;
+import org.springframework.integration.support.management.observation.IntegrationObservation;
+import org.springframework.integration.support.management.observation.MessageSenderContext;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.GenericMessage;
-import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
+import io.micrometer.common.KeyValues;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.observation.DefaultMeterObservationHandler;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import io.micrometer.core.tck.MeterRegistryAssert;
 import io.micrometer.observation.Observation;
 import io.micrometer.observation.ObservationHandler;
 import io.micrometer.observation.ObservationRegistry;
 import io.micrometer.observation.tck.TestObservationRegistry;
 import io.micrometer.observation.tck.TestObservationRegistryAssert;
-import io.micrometer.observation.transport.ReceiverContext;
-import io.micrometer.observation.transport.SenderContext;
 import io.micrometer.tracing.Span;
 import io.micrometer.tracing.TraceContext;
 import io.micrometer.tracing.Tracer;
@@ -79,6 +87,9 @@ public class ObservationPropagationChannelInterceptorTests {
 	@Autowired
 	ObservationRegistry observationRegistry;
 
+	@Autowired
+	MeterRegistry meterRegistry;
+
 	@Autowired
 	SimpleTracer simpleTracer;
 
@@ -199,36 +210,29 @@ public class ObservationPropagationChannelInterceptorTests {
 	}
 
 	@Test
-	@SuppressWarnings({ "unchecked", "rawtypes" })
-	void observationContextPropagatedOverDirectChannel() throws InterruptedException {
-		CountDownLatch handleLatch = new CountDownLatch(1);
-		this.testTracingChannel.subscribe(m -> {
-			// This would be the instrumentation code on the receiver side
-			// We would need to check if Zipkin wouldn't require us to create the receiving span and then an additional one for the user code...
-			ReceiverContext<Message<?>> receiverContext =
-					new ReceiverContext<>((carrier, key) -> carrier.getHeaders().get(key, String.class));
-			receiverContext.setCarrier(m);
+	void observationContextPropagatedOverExecutorChannel() {
+		BridgeHandler handler = new BridgeHandler();
+		handler.registerObservationRegistry(this.observationRegistry);
+		handler.setBeanName("testBridge");
+		this.testTracingChannel.subscribe(handler);
 
-			// ...and this would be the user's code
-			Observation.createNotStarted("user.code", receiverContext, this.observationRegistry)
-					.observe(() -> {
-						// Let's assume that this is the user code
-					});
-			handleLatch.countDown();
-		});
+		QueueChannel replyChannel = new QueueChannel();
 
-		// This would be the instrumentation code on the sender side (user's code would call e.g. MessageTemplate and this code
-		// would lay in MessageTemplate)
-		// We need to mutate the carrier, so we need to use the builder not the message since message headers are immutable
-		SenderContext<MessageBuilder<String>> senderContext =
-				new SenderContext<>((carrier, key, value) -> Objects.requireNonNull(carrier).setHeader(key, value));
-		MessageBuilder<String> builder = MessageBuilder.withPayload("test");
-		senderContext.setCarrier(builder);
+		MutableMessage<String> message =
+				(MutableMessage<String>) MutableMessageBuilder.withPayload("test")
+						.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel)
+						.build();
 
-		Observation.createNotStarted("sending", senderContext, this.observationRegistry)
-				.observe(() -> this.testTracingChannel.send(builder.build()));
+		Observation.createNotStarted("sending", new MessageSenderContext(message), this.observationRegistry)
+				.observe(() -> this.testTracingChannel.send(message));
 
-		assertThat(handleLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		Message<?> receive = replyChannel.receive();
+
+		assertThat(receive).isNotNull()
+				.extracting(Message::getHeaders)
+				.asInstanceOf(InstanceOfAssertFactories.MAP)
+				.containsEntry("foo", "some foo value")
+				.containsEntry("bar", "some bar value");
 
 		TestObservationRegistryAssert.assertThat(this.observationRegistry)
 				.doesNotHaveAnyRemainingCurrentObservation();
@@ -236,12 +240,28 @@ public class ObservationPropagationChannelInterceptorTests {
 		TracerAssert.assertThat(this.simpleTracer)
 				.reportedSpans()
 				.hasSize(2)
-				.satisfies(simpleSpans -> SpansAssert.assertThat((Collection<FinishedSpan>) (Collection) simpleSpans)
+				.satisfies(simpleSpans -> assertSpans(simpleSpans)
 						.hasASpanWithName("sending")
-						.assertThatASpanWithNameEqualTo("user.code")
+						.assertThatASpanWithNameEqualTo("testBridge receive")
 						.hasTag("foo", "some foo value")
 						.hasTag("bar", "some bar value")
+						.hasTag("spring.integration.type", "handler")
+						.hasTag("spring.integration.name", "testBridge")
 						.hasKindEqualTo(Span.Kind.CONSUMER));
+
+
+		MeterRegistryAssert.assertThat(this.meterRegistry)
+				.hasTimerWithNameAndTags("spring.integration.handler",
+						KeyValues.of(IntegrationObservation.HandlerTags.COMPONENT_NAME.asString(), "testBridge",
+								IntegrationObservation.HandlerTags.COMPONENT_TYPE.asString(), "handler",
+								"error", "none"));
+
+		assertThat(this.meterRegistry.get("spring.integration.handler").timer().count()).isEqualTo(1);
+	}
+
+	@SuppressWarnings("unchecked")
+	private static SpansAssert assertSpans(Collection<? extends FinishedSpan> actual) {
+		return SpansAssert.assertThat((Collection<FinishedSpan>) actual);
 	}
 
 	@Configuration
@@ -254,17 +274,24 @@ public class ObservationPropagationChannelInterceptorTests {
 		}
 
 		@Bean
-		ObservationRegistry observationRegistry(Tracer tracer, Propagator propagator) {
+		MeterRegistry meterRegistry() {
+			return new SimpleMeterRegistry();
+		}
+
+		@Bean
+		ObservationRegistry observationRegistry(Tracer tracer, Propagator propagator, MeterRegistry meterRegistry) {
 			TestObservationRegistry observationRegistry = TestObservationRegistry.create();
-			observationRegistry.observationConfig().observationHandler(
-					// Composite will pick the first matching handler
-					new ObservationHandler.FirstMatchingCompositeObservationHandler(
-							// This is responsible for creating a child span on the sender side
-							new PropagatingSenderTracingObservationHandler<>(tracer, propagator),
-							// This is responsible for creating a span on the receiver side
-							new PropagatingReceiverTracingObservationHandler<>(tracer, propagator),
-							// This is responsible for creating a default span
-							new DefaultTracingObservationHandler(tracer)));
+			observationRegistry.observationConfig()
+					.observationHandler(new DefaultMeterObservationHandler(meterRegistry))
+					.observationHandler(
+							// Composite will pick the first matching handler
+							new ObservationHandler.FirstMatchingCompositeObservationHandler(
+									// This is responsible for creating a child span on the sender side
+									new PropagatingSenderTracingObservationHandler<>(tracer, propagator),
+									// This is responsible for creating a span on the receiver side
+									new PropagatingReceiverTracingObservationHandler<>(tracer, propagator),
+									// This is responsible for creating a default span
+									new DefaultTracingObservationHandler(tracer)));
 			return observationRegistry;
 		}
 
@@ -318,14 +345,13 @@ public class ObservationPropagationChannelInterceptorTests {
 					setter.set(carrier, "bar", "some bar value");
 				}
 
-
 				// This is called on the consumer side when the message is consumed
 				// Normally we would use tools like Extractor from tracing but for tests we are just manually creating a span
 				@Override
 				public <C> Span.Builder extract(C carrier, Getter<C> getter) {
 					String foo = getter.get(carrier, "foo");
 					String bar = getter.get(carrier, "bar");
-					return tracer.spanBuilder().kind(Span.Kind.CONSUMER).tag("foo", foo).tag("bar", bar);
+					return tracer.spanBuilder().tag("foo", foo).tag("bar", bar);
 				}
 			};
 		}

spring-integration-core/src/test/java/org/springframework/integration/support/management/observation/IntegrationObservabilityZipkinTests.java

@@ -0,0 +1,123 @@
+/*
+ * Copyright 2022 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.integration.support.management.observation;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.annotation.BridgeTo;
+import org.springframework.integration.annotation.EndpointId;
+import org.springframework.integration.annotation.Poller;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.channel.interceptor.ObservationPropagationChannelInterceptor;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.config.EnableIntegrationManagement;
+import org.springframework.integration.config.GlobalChannelInterceptor;
+import org.springframework.integration.support.MutableMessage;
+import org.springframework.integration.support.MutableMessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.ChannelInterceptor;
+
+import io.micrometer.common.KeyValues;
+import io.micrometer.core.tck.MeterRegistryAssert;
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationRegistry;
+import io.micrometer.tracing.Span;
+import io.micrometer.tracing.test.SampleTestRunner;
+import io.micrometer.tracing.test.simple.SpansAssert;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class IntegrationObservabilityZipkinTests extends SampleTestRunner {
+
+	@Override
+	public TracingSetup[] getTracingSetup() {
+		return new TracingSetup[]{ TracingSetup.IN_MEMORY_BRAVE, TracingSetup.ZIPKIN_BRAVE };
+	}
+
+	@Override
+	public SampleTestRunnerConsumer yourCode() {
+		return (bb, meterRegistry) -> {
+			ObservationRegistry observationRegistry = getObservationRegistry();
+			try (AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext()) {
+				applicationContext.registerBean(ObservationRegistry.class, () -> observationRegistry);
+				applicationContext.register(ObservationIntegrationTestConfiguration.class);
+				applicationContext.refresh();
+
+				PollableChannel queueChannel = applicationContext.getBean("queueChannel", PollableChannel.class);
+				PollableChannel replyChannel = new QueueChannel();
+
+				MutableMessage<String> testMessage =
+						(MutableMessage<String>) MutableMessageBuilder.withPayload("test data")
+								.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel)
+								.build();
+
+				Observation.createNotStarted("Test send", new MessageSenderContext(testMessage), observationRegistry)
+						.observe(() -> queueChannel.send(testMessage));
+
+				Message<?> receive = replyChannel.receive(10_000);
+				assertThat(receive).isNotNull()
+						.extracting("payload").isEqualTo("test data");
+			}
+
+			SpansAssert.assertThat(bb.getFinishedSpans())
+					.haveSameTraceId()
+					.hasASpanWithName("Test send", spanAssert -> spanAssert.hasKindEqualTo(Span.Kind.PRODUCER))
+					.hasASpanWithName("observedEndpoint receive", spanAssert -> spanAssert
+							.hasTag(IntegrationObservation.HandlerTags.COMPONENT_NAME.asString(), "observedEndpoint")
+							.hasTag(IntegrationObservation.HandlerTags.COMPONENT_TYPE.asString(), "handler")
+							.hasKindEqualTo(Span.Kind.CONSUMER))
+					.hasSize(2);
+
+			MeterRegistryAssert.assertThat(getMeterRegistry())
+					.hasTimerWithNameAndTags("spring.integration.handler",
+							KeyValues.of(
+									IntegrationObservation.HandlerTags.COMPONENT_NAME.asString(), "observedEndpoint",
+									IntegrationObservation.HandlerTags.COMPONENT_TYPE.asString(), "handler",
+									"error", "none"));
+		};
+	}
+
+
+	@Configuration
+	@EnableIntegration
+	@EnableIntegrationManagement
+	public static class ObservationIntegrationTestConfiguration {
+
+		@Bean
+		@GlobalChannelInterceptor
+		public ChannelInterceptor observationPropagationInterceptor(ObservationRegistry observationRegistry) {
+			return new ObservationPropagationChannelInterceptor(observationRegistry);
+		}
+
+		@Bean
+		@BridgeTo(poller = @Poller(fixedDelay = "100"))
+		@EndpointId("observedEndpoint")
+		public PollableChannel queueChannel() {
+			return new QueueChannel();
+		}
+
+	}
+
+}