Add observation to inbound endpoints (#3930)

* Add observation to inbound endpoints

* * Improve `Observation` logic in the `MessageProducerSupport` and `MessagingGatewaySupport`
* Clean up Javadocs in observation support classes
* Add test for `MessagingGatewaySupport` instrumentation into `IntegrationObservabilityZipkinTests`
* Add docs for `@EnableIntegrationManagement.observationPatterns()`
* The generated docs for metrics and spans looks OK: have both `Gateway` and `Handler` section now

* * Fix Checkstyle violations

* * Fix `IntegrationMBeanExporter` to filter out a `MessageProducer` from sources.
Marking a `MessageProducerSupport` with an `IntegrationInboundManagement`
causes it to be considered as a source for JMX.
Technically it might have a reason since it is indeed a source, but that's fully different story

* * No observation instrumentation by default

* * Fix "no observation by default" configuration logic

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

@@ -54,7 +54,7 @@ public @interface EnableIntegrationManagement {
 	 * <p>
 	 * Set this to false for disabling logging by default in all framework components that implement
 	 * {@link org.springframework.integration.support.management.IntegrationManagement}
-	 * (channels, message handlers etc). It turns off logging such as "PreSend on channel", "Received message" etc.
+	 * (channels, message handlers etc.). It 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
 	 * {@link org.springframework.integration.support.management.IntegrationManagement#setLoggingEnabled(boolean)}.
@@ -62,4 +62,15 @@ public @interface EnableIntegrationManagement {
 	 */
 	String defaultLoggingEnabled() default "true";
 
+	/**
+	 * Set simple pattern component names matching for observation registry injection.
+	 * @return simple pattern component names matching for observation registry injection.
+	 * None by default - no unconditional observation instrumentation.
+	 * Can be set to {@code *} to instrumentation all the integration components.
+	 * The pattern can start with {@code !} to negate the matching.
+	 * @since 6.0
+	 * @see org.springframework.integration.support.utils.PatternMatchUtils#smartMatch(String, String...)
+	 */
+	String[] observationPatterns() default { };
+
 }

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

@@ -75,7 +75,11 @@ public class IntegrationManagementConfiguration implements ImportAware, Environm
 				Boolean.parseBoolean(this.environment.resolvePlaceholders(
 						(String) this.attributes.get("defaultLoggingEnabled"))));
 		configurer.setMetricsCaptorProvider(metricsCaptorProvider);
-		configurer.setObservationRegistry(observationRegistryProvider);
+		String[] observationPatterns = (String[]) this.attributes.get("observationPatterns");
+		if (observationPatterns.length > 0) {
+			configurer.setObservationPatterns(observationPatterns);
+			configurer.setObservationRegistry(observationRegistryProvider);
+		}
 		return configurer;
 	}
 

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

@@ -19,6 +19,8 @@ package org.springframework.integration.config;
 import java.util.HashSet;
 import java.util.Set;
 
+import org.apache.commons.logging.Log;
+
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.ObjectProvider;
@@ -33,6 +35,7 @@ import org.springframework.integration.support.management.IntegrationManagement;
 import org.springframework.integration.support.management.IntegrationManagement.ManagementOverrides;
 import org.springframework.integration.support.management.metrics.MeterFacade;
 import org.springframework.integration.support.management.metrics.MetricsCaptor;
+import org.springframework.integration.support.utils.PatternMatchUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
@@ -80,6 +83,8 @@ public class IntegrationManagementConfigurer
 
 	private ObjectProvider<ObservationRegistry> observationRegistryProvider;
 
+	private String[] observationPatterns;
+
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
 		this.applicationContext = applicationContext;
@@ -98,7 +103,7 @@ public class IntegrationManagementConfigurer
 	 * Exception logging (debug or otherwise) is not affected by this setting.
 	 * <p>
 	 * It has been found that in high-volume messaging environments, calls to methods such as
-	 * {@link org.apache.commons.logging.Log#isDebugEnabled()} can be quite expensive
+	 * {@link 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
@@ -135,6 +140,16 @@ public class IntegrationManagementConfigurer
 		this.observationRegistryProvider = observationRegistryProvider;
 	}
 
+	/**
+	 * Set simple patterns for component names matching which has to be instrumented with a {@link ObservationRegistry}.
+	 * @param observationPatterns the simple patterns to use.
+	 * @since 6.0
+	 * @see PatternMatchUtils#smartMatch(String, String...)
+	 */
+	public void setObservationPatterns(String... observationPatterns) {
+		Assert.notEmpty(observationPatterns, "'observationPatterns' must not be empty");
+		this.observationPatterns = observationPatterns;
+	}
 
 	@Override
 	public void afterSingletonsInstantiated() {
@@ -171,19 +186,19 @@ public class IntegrationManagementConfigurer
 	private void registerComponentGauges() {
 		this.gauges.add(
 				this.metricsCaptor.gaugeBuilder("spring.integration.channels", this,
-						(c) -> this.applicationContext.getBeansOfType(MessageChannel.class).size())
+								(c) -> this.applicationContext.getBeansOfType(MessageChannel.class).size())
 						.description("The number of message channels")
 						.build());
 
 		this.gauges.add(
 				this.metricsCaptor.gaugeBuilder("spring.integration.handlers", this,
-						(c) -> this.applicationContext.getBeansOfType(MessageHandler.class).size())
+								(c) -> this.applicationContext.getBeansOfType(MessageHandler.class).size())
 						.description("The number of message handlers")
 						.build());
 
 		this.gauges.add(
 				this.metricsCaptor.gaugeBuilder("spring.integration.sources", this,
-						(c) -> this.applicationContext.getBeansOfType(MessageSource.class).size())
+								(c) -> this.applicationContext.getBeansOfType(MessageSource.class).size())
 						.description("The number of message sources")
 						.build());
 	}
@@ -195,7 +210,10 @@ public class IntegrationManagementConfigurer
 		if (this.metricsCaptor != null) {
 			integrationManagement.registerMetricsCaptor(this.metricsCaptor);
 		}
-		if (this.observationRegistry != null) {
+		if (this.observationRegistry != null &&
+				Boolean.TRUE.equals(PatternMatchUtils.smartMatch(
+						integrationManagement.getComponentName(), this.observationPatterns))) {
+
 			integrationManagement.registerObservationRegistry(this.observationRegistry);
 		}
 	}

spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java

@@ -21,6 +21,7 @@ import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.context.IntegrationProperties;
 import org.springframework.integration.support.channel.ChannelResolverUtils;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.core.GenericMessagingTemplate;
@@ -80,6 +81,7 @@ public class MessagingTemplate extends GenericMessagingTemplate {
 	}
 
 	@Override
+	@Nullable
 	public Message<?> sendAndReceive(MessageChannel destination, Message<?> requestMessage) {
 		if (!this.throwExceptionOnLateReplySet) {
 			synchronized (this) {

spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java

@@ -31,7 +31,12 @@ import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.support.DefaultErrorMessageStrategy;
 import org.springframework.integration.support.ErrorMessageStrategy;
 import org.springframework.integration.support.ErrorMessageUtils;
+import org.springframework.integration.support.management.IntegrationInboundManagement;
 import org.springframework.integration.support.management.TrackableComponent;
+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.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -40,6 +45,7 @@ import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
+import io.micrometer.observation.ObservationRegistry;
 import reactor.core.publisher.Flux;
 
 /**
@@ -50,13 +56,19 @@ import reactor.core.publisher.Flux;
  * @author Artem Bilan
  * @author Gary Russell
  */
-public abstract class MessageProducerSupport extends AbstractEndpoint implements MessageProducer, TrackableComponent,
-		SmartInitializingSingleton, IntegrationPattern {
+public abstract class MessageProducerSupport extends AbstractEndpoint
+		implements MessageProducer, TrackableComponent,
+		SmartInitializingSingleton, IntegrationPattern, IntegrationInboundManagement {
 
 	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
 
 	private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
 
+	private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
+
+	@Nullable
+	private MessageReceiverObservationConvention observationConvention;
+
 	private MessageChannel outputChannel;
 
 	private String outputChannelName;
@@ -135,7 +147,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
 	/**
 	 * Configure the default timeout value to use for send operations.
 	 * May be overridden for individual messages.
-	 * @param sendTimeout the send timeout in milliseconds
+	 * @param sendTimeout the 'send' operation timeout in milliseconds
 	 * @see MessagingTemplate#setSendTimeout
 	 */
 	public void setSendTimeout(long sendTimeout) {
@@ -172,6 +184,22 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
 		return this.messagingTemplate;
 	}
 
+	/**
+	 * 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
+	public void registerObservationRegistry(ObservationRegistry observationRegistry) {
+		Assert.notNull(observationRegistry, "'observationRegistry' must not be null");
+		this.observationRegistry = observationRegistry;
+	}
+
 	@Override
 	public IntegrationPatternType getIntegrationPatternType() {
 		return IntegrationPatternType.inbound_channel_adapter;
@@ -216,14 +244,18 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
 		}
 	}
 
-	protected void sendMessage(Message<?> messageArg) {
-		Message<?> message = messageArg;
+	protected void sendMessage(Message<?> message) {
 		if (message == null) {
 			throw new MessagingException("cannot send a null message");
 		}
-		message = trackMessageIfAny(message);
+
 		try {
-			this.messagingTemplate.send(getRequiredOutputChannel(), message);
+			IntegrationObservation.HANDLER.observation(
+							this.observationConvention,
+							DefaultMessageReceiverObservationConvention.INSTANCE,
+							() -> new MessageReceiverContext(message, getComponentName()),
+							this.observationRegistry)
+					.observe(() -> this.messagingTemplate.send(getRequiredOutputChannel(), trackMessageIfAny(message)));
 		}
 		catch (RuntimeException ex) {
 			if (!sendErrorMessageIfNecessary(message, ex)) {

spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java

@@ -53,6 +53,10 @@ import org.springframework.integration.support.management.metrics.MeterFacade;
 import org.springframework.integration.support.management.metrics.MetricsCaptor;
 import org.springframework.integration.support.management.metrics.SampleFacade;
 import org.springframework.integration.support.management.metrics.TimerFacade;
+import org.springframework.integration.support.management.observation.DefaultMessageRequestReplyReceiverObservationConvention;
+import org.springframework.integration.support.management.observation.IntegrationObservation;
+import org.springframework.integration.support.management.observation.MessageRequestReplyReceiverContext;
+import org.springframework.integration.support.management.observation.MessageRequestReplyReceiverObservationConvention;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -62,10 +66,12 @@ import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.core.MessagePostProcessor;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
 
+import io.micrometer.observation.ObservationRegistry;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
 
@@ -87,7 +93,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 
 	private static final long DEFAULT_TIMEOUT = 1000L;
 
-	protected final MessagingTemplate messagingTemplate; // NOSONAR
+	protected final ConvertingMessagingTemplate messagingTemplate; // NOSONAR
 
 	private final SimpleMessageConverter messageConverter = new SimpleMessageConverter();
 
@@ -130,6 +136,11 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 
 	private TimerFacade successTimer;
 
+	private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
+
+	@Nullable
+	private MessageRequestReplyReceiverObservationConvention observationConvention;
+
 	private volatile AbstractEndpoint replyMessageCorrelator;
 
 	private volatile boolean initialized;
@@ -152,7 +163,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 	 * @see #setErrorOnTimeout
 	 */
 	public MessagingGatewaySupport(boolean errorOnTimeout) {
-		MessagingTemplate template = new MessagingTemplate();
+		ConvertingMessagingTemplate template = new ConvertingMessagingTemplate();
 		template.setMessageConverter(this.messageConverter);
 		template.setSendTimeout(DEFAULT_TIMEOUT);
 		template.setReceiveTimeout(this.replyTimeout);
@@ -354,6 +365,18 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 		this.metricsCaptor = metricsCaptorToRegister;
 	}
 
+	@Override
+	public void registerObservationRegistry(ObservationRegistry observationRegistry) {
+		Assert.notNull(observationRegistry, "'observationRegistry' must not be null");
+		this.observationRegistry = observationRegistry;
+	}
+
+	public void setObservationConvention(
+			@Nullable MessageRequestReplyReceiverObservationConvention observationConvention) {
+
+		this.observationConvention = observationConvention;
+	}
+
 	@Override
 	protected void onInit() {
 		Assert.state(!(this.requestChannelName != null && this.requestChannel != null),
@@ -494,16 +517,16 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 
 	@Nullable
 	protected Object sendAndReceive(Object object) {
-		return doSendAndReceive(object, true);
+		return sendAndReceive(object, true);
 	}
 
 	@Nullable
 	protected Message<?> sendAndReceiveMessage(Object object) {
-		return (Message<?>) doSendAndReceive(object, false);
+		return (Message<?>) sendAndReceive(object, false);
 	}
 
-	@Nullable // NOSONAR
-	private Object doSendAndReceive(Object object, boolean shouldConvert) {
+	@Nullable
+	private Object sendAndReceive(Object object, boolean shouldConvert) {
 		initializeIfNecessary();
 		Assert.notNull(object, "request must not be null");
 		MessageChannel channel = getRequestChannel();
@@ -515,36 +538,28 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 
 		Object reply;
 		Message<?> requestMessage = null;
-		SampleFacade sample = null;
 		try {
-			if (this.metricsCaptor != null) {
-				sample = this.metricsCaptor.start();
+			requestMessage = convertToRequestMessage(object, shouldConvert);
+			Message<?> replyMessage;
+
+			if (this.observationRegistry != ObservationRegistry.NOOP) {
+				replyMessage = sendAndReceiveWithObservation(channel, object, requestMessage);
 			}
-			if (shouldConvert) {
-				reply = this.messagingTemplate.convertSendAndReceive(channel, object, Object.class,
-						this.historyWritingPostProcessor);
+			else if (this.metricsCaptor != null) {
+				replyMessage = sendAndReceiveWithMetrics(channel, object, requestMessage);
 			}
 			else {
-				requestMessage = (object instanceof Message<?>)
-						? (Message<?>) object : this.requestMapper.toMessage(object);
-				Assert.state(requestMessage != null, () -> "request mapper resulted in no message for " + object);
-				requestMessage = this.historyWritingPostProcessor.postProcessMessage(requestMessage);
-				reply = this.messagingTemplate.sendAndReceive(channel, requestMessage);
+				replyMessage = doSendAndReceive(channel, object, requestMessage);
 			}
 
-			if (reply == null && this.errorOnTimeout) {
-				throwMessageTimeoutException(object, "No reply received within timeout");
-			}
-			if (sample != null) {
-				sample.stop(sendTimer());
+			reply = replyMessage;
+			if (shouldConvert) {
+				reply = this.messagingTemplate.getMessageConverter().fromMessage(replyMessage, Object.class);
 			}
 		}
 		catch (Throwable ex) { // NOSONAR (catch throwable)
 			logger.debug(() -> "failure occurred in gateway sendAndReceive: " + ex.getMessage());
 			reply = ex;
-			if (sample != null) {
-				sample.stop(buildSendTimer(false, ex.getClass().getSimpleName()));
-			}
 		}
 
 		if (reply instanceof Throwable || reply instanceof ErrorMessage) {
@@ -557,6 +572,57 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 		return reply;
 	}
 
+	private Message<?> convertToRequestMessage(Object object, boolean shouldConvert) {
+		if (shouldConvert) {
+			return this.messagingTemplate.doConvert(object, null, this.historyWritingPostProcessor);
+		}
+		else {
+			Message<?> requestMessage = (object instanceof Message<?>)
+					? (Message<?>) object : this.requestMapper.toMessage(object);
+			Assert.state(requestMessage != null, () -> "request mapper resulted in no message for " + object);
+			return this.historyWritingPostProcessor.postProcessMessage(requestMessage);
+		}
+	}
+
+	private Message<?> sendAndReceiveWithObservation(MessageChannel requestChannel, Object object,
+			Message<?> requestMessage) {
+
+		MessageRequestReplyReceiverContext context =
+				new MessageRequestReplyReceiverContext(requestMessage, getComponentName());
+
+		return IntegrationObservation.GATEWAY.observation(this.observationConvention,
+						DefaultMessageRequestReplyReceiverObservationConvention.INSTANCE,
+						() -> context, this.observationRegistry)
+				.observe(() -> {
+					Message<?> replyMessage = doSendAndReceive(requestChannel, object, requestMessage);
+					context.setResponse(replyMessage);
+					return replyMessage;
+				});
+	}
+
+	private Message<?> sendAndReceiveWithMetrics(MessageChannel requestChannel, Object object,
+			Message<?> requestMessage) {
+
+		SampleFacade sample = this.metricsCaptor.start();
+		try {
+			Message<?> replyMessage = doSendAndReceive(requestChannel, object, requestMessage);
+			sample.stop(sendTimer());
+			return replyMessage;
+		}
+		catch (Throwable ex) {
+			sample.stop(buildSendTimer(false, ex.getClass().getSimpleName()));
+			throw ex;
+		}
+	}
+
+	private Message<?> doSendAndReceive(MessageChannel requestChannel, Object object, Message<?> requestMessage) {
+		Message<?> replyMessage = this.messagingTemplate.sendAndReceive(requestChannel, requestMessage);
+		if (replyMessage == null && this.errorOnTimeout) {
+			throwMessageTimeoutException(object, "No reply received within timeout");
+		}
+		return replyMessage;
+	}
+
 	@Nullable
 	private Object handleSendAndReceiveError(Object object, @Nullable Message<?> requestMessage, Throwable error,
 			boolean shouldConvert) {
@@ -646,23 +712,23 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 		}
 
 		return Mono.defer(() -> {
-			Object originalReplyChannelHeader = requestMessage.getHeaders().getReplyChannel();
-			Object originalErrorChannelHeader = requestMessage.getHeaders().getErrorChannel();
+					Object originalReplyChannelHeader = requestMessage.getHeaders().getReplyChannel();
+					Object originalErrorChannelHeader = requestMessage.getHeaders().getErrorChannel();
 
-			MonoReplyChannel replyChan = new MonoReplyChannel();
+					MonoReplyChannel replyChan = new MonoReplyChannel();
 
-			Message<?> messageToSend = MutableMessageBuilder.fromMessage(requestMessage)
-					.setReplyChannel(replyChan)
-					.setHeader(this.messagingTemplate.getSendTimeoutHeader(), null)
-					.setHeader(this.messagingTemplate.getReceiveTimeoutHeader(), null)
-					.setErrorChannel(replyChan)
-					.build();
+					Message<?> messageToSend = MutableMessageBuilder.fromMessage(requestMessage)
+							.setReplyChannel(replyChan)
+							.setHeader(this.messagingTemplate.getSendTimeoutHeader(), null)
+							.setHeader(this.messagingTemplate.getReceiveTimeoutHeader(), null)
+							.setErrorChannel(replyChan)
+							.build();
 
-			sendMessageForReactiveFlow(requestChannel, messageToSend);
+					sendMessageForReactiveFlow(requestChannel, messageToSend);
 
-			return buildReplyMono(requestMessage, replyChan.replyMono.asMono(), error, originalReplyChannelHeader,
-					originalErrorChannelHeader);
-		})
+					return buildReplyMono(requestMessage, replyChan.replyMono.asMono(), error, originalReplyChannelHeader,
+							originalErrorChannelHeader);
+				})
 				.onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t));
 	}
 
@@ -916,4 +982,19 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 
 	}
 
+	/**
+	 * The {@link MessagingTemplate} extension to increase {@link #doConvert(Object, Map, MessagePostProcessor)}
+	 * visibility to get access to the request message from an observation context.
+	 */
+	protected static class ConvertingMessagingTemplate extends MessagingTemplate {
+
+		@Override // NOSONAR Increase visibility
+		public Message<?> doConvert(Object payload, @Nullable Map<String, Object> headers,
+				@Nullable MessagePostProcessor postProcessor) {
+
+			return super.doConvert(payload, headers, postProcessor);
+		}
+
+	}
+
 }

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

@@ -0,0 +1,49 @@
+/*
+ * 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 MessageRequestReplyReceiverObservationConvention} implementation.
+ * Provides low cardinalities as a {@link IntegrationObservation.GatewayTags} values.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class DefaultMessageRequestReplyReceiverObservationConvention
+		implements MessageRequestReplyReceiverObservationConvention {
+
+	/**
+	 * A shared singleton instance for {@link DefaultMessageRequestReplyReceiverObservationConvention}.
+	 */
+	public static final DefaultMessageRequestReplyReceiverObservationConvention INSTANCE =
+			new DefaultMessageRequestReplyReceiverObservationConvention();
+
+	@Override
+	public KeyValues getLowCardinalityKeyValues(MessageRequestReplyReceiverContext context) {
+		return KeyValues
+				// See IntegrationObservation.GatewayTags.COMPONENT_NAME - to avoid class tangle
+				.of("spring.integration.name", context.getGatewayName())
+				// See IntegrationObservation.GatewayTags.COMPONENT_TYPE - to avoid class tangle
+				.and("spring.integration.type", "gateway")
+				// See IntegrationObservation.GatewayTags.OUTCOME - to avoid class tangle
+				.and("spring.integration.outcome", context.getError() != null ? "INTERNAL_ERROR" : "SUCCESS");
+	}
+
+}

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

@@ -21,6 +21,9 @@ import io.micrometer.observation.docs.ObservationDocumentation;
 
 /**
  * The {@link ObservationDocumentation} implementation for Spring Integration infrastructure.
+ * <p>
+ * NOTE: This class is mostly intended for observation docs generation, so any string literals,
+ * even if they are the same, cannot be extracted into constants or super methods.
  *
  * @author Artem Bilan
  *
@@ -47,6 +50,27 @@ public enum IntegrationObservation implements ObservationDocumentation {
 			return HandlerTags.values();
 		}
 
+	},
+
+	/**
+	 * Observation for inbound message gateways.
+	 */
+	GATEWAY {
+		@Override
+		public String getPrefix() {
+			return "spring.integration.";
+		}
+
+		@Override
+		public Class<DefaultMessageRequestReplyReceiverObservationConvention> getDefaultConvention() {
+			return DefaultMessageRequestReplyReceiverObservationConvention.class;
+		}
+
+		@Override
+		public KeyName[] getLowCardinalityKeyNames() {
+			return GatewayTags.values();
+		}
+
 	};
 
 	/**
@@ -78,4 +102,43 @@ public enum IntegrationObservation implements ObservationDocumentation {
 
 	}
 
+	/**
+	 * Key names for message handler observations.
+	 */
+	public enum GatewayTags implements KeyName {
+
+		/**
+		 * Name of the message gateway component.
+		 */
+		COMPONENT_NAME {
+			@Override
+			public String asString() {
+				return "spring.integration.name";
+			}
+
+		},
+
+		/**
+		 * Type of the component - 'gateway'.
+		 */
+		COMPONENT_TYPE {
+			@Override
+			public String asString() {
+				return "spring.integration.type";
+			}
+
+		},
+
+		/**
+		 * Outcome of the request/reply execution.
+		 */
+		OUTCOME {
+			@Override
+			public String asString() {
+				return "spring.integration.outcome";
+			}
+		},
+
+	}
+
 }

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

@@ -16,14 +16,11 @@
 
 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.
+ * A {@link MessageReceiverContext}-based {@link ObservationConvention} contract.
  *
  * @author Artem Bilan
  *

spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageRequestReplyReceiverContext.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.RequestReplyReceiverContext;
+
+/**
+ * The {@link RequestReplyReceiverContext} extension for a {@link Message} contract with inbound gateways.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class MessageRequestReplyReceiverContext extends RequestReplyReceiverContext<Message<?>, Message<?>> {
+
+	private final Message<?> message;
+
+	private final String gatewayName;
+
+	public MessageRequestReplyReceiverContext(Message<?> message, @Nullable String gatewayName) {
+		super((carrier, key) -> carrier.getHeaders().get(key, String.class));
+		this.message = message;
+		this.gatewayName = gatewayName != null ? gatewayName : "unknown";
+	}
+
+	@Override
+	public Message<?> getCarrier() {
+		return this.message;
+	}
+
+	public String getGatewayName() {
+		return this.gatewayName;
+	}
+
+}

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

@@ -0,0 +1,47 @@
+/*
+ * 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.observation.Observation;
+import io.micrometer.observation.ObservationConvention;
+
+/**
+ * A {@link MessageRequestReplyReceiverContext}-based {@link ObservationConvention} contract.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public interface MessageRequestReplyReceiverObservationConvention
+		extends ObservationConvention<MessageRequestReplyReceiverContext> {
+
+	@Override
+	default String getName() {
+		return "spring.integration.gateway";
+	}
+
+	@Override
+	default boolean supportsContext(Observation.Context context) {
+		return context instanceof MessageRequestReplyReceiverContext;
+	}
+
+	@Override
+	default String getContextualName(MessageRequestReplyReceiverContext context) {
+		return context.getGatewayName() + " process";
+	}
+
+}

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

@@ -32,18 +32,17 @@ import org.springframework.integration.channel.interceptor.ObservationPropagatio
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.config.EnableIntegrationManagement;
 import org.springframework.integration.config.GlobalChannelInterceptor;
+import org.springframework.integration.gateway.MessagingGatewaySupport;
 import org.springframework.integration.handler.BridgeHandler;
 import org.springframework.integration.handler.advice.HandleMessageAdvice;
-import org.springframework.integration.support.MutableMessage;
-import org.springframework.integration.support.MutableMessageBuilder;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.messaging.support.GenericMessage;
 
 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;
@@ -70,18 +69,11 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner {
 				applicationContext.register(ObservationIntegrationTestConfiguration.class);
 				applicationContext.refresh();
 
-				PollableChannel queueChannel = applicationContext.getBean("queueChannel", PollableChannel.class);
-				PollableChannel replyChannel = new QueueChannel();
+				TestMessagingGatewaySupport messagingGateway =
+						applicationContext.getBean(TestMessagingGatewaySupport.class);
 
-				MutableMessage<String> message =
-						(MutableMessage<String>) MutableMessageBuilder.withPayload("test data")
-								.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel)
-								.build();
+				Message<?> receive = messagingGateway.process(new GenericMessage<>("test data"));
 
-				Observation.createNotStarted("Test send", () -> new MessageSenderContext(message), observationRegistry)
-						.observe(() -> queueChannel.send(message));
-
-				Message<?> receive = replyChannel.receive(10_000);
 				assertThat(receive).isNotNull()
 						.extracting("payload").isEqualTo("test data");
 				var configuration = applicationContext.getBean(ObservationIntegrationTestConfiguration.class);
@@ -91,7 +83,11 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner {
 
 			SpansAssert.assertThat(bb.getFinishedSpans())
 					.haveSameTraceId()
-					.hasASpanWithName("Test send", spanAssert -> spanAssert.hasKindEqualTo(Span.Kind.PRODUCER))
+					.hasASpanWithName("testInboundGateway process", spanAssert -> spanAssert
+							.hasTag(IntegrationObservation.GatewayTags.COMPONENT_NAME.asString(), "testInboundGateway")
+							.hasTag(IntegrationObservation.GatewayTags.COMPONENT_TYPE.asString(), "gateway")
+							.hasTagWithKey("test.message.id")
+							.hasKindEqualTo(Span.Kind.SERVER))
 					.hasASpanWithName("observedEndpoint receive", spanAssert -> spanAssert
 							.hasTag(IntegrationObservation.HandlerTags.COMPONENT_NAME.asString(), "observedEndpoint")
 							.hasTag(IntegrationObservation.HandlerTags.COMPONENT_TYPE.asString(), "handler")
@@ -110,7 +106,7 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner {
 
 	@Configuration
 	@EnableIntegration
-	@EnableIntegrationManagement
+	@EnableIntegrationManagement(observationPatterns = { "observedEndpoint", "testInboundGateway" })
 	public static class ObservationIntegrationTestConfiguration {
 
 		CountDownLatch observedHandlerLatch = new CountDownLatch(1);
@@ -121,6 +117,22 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner {
 			return new ObservationPropagationChannelInterceptor(observationRegistry);
 		}
 
+		@Bean
+		TestMessagingGatewaySupport testInboundGateway(PollableChannel queueChannel) {
+			TestMessagingGatewaySupport messagingGatewaySupport = new TestMessagingGatewaySupport();
+			messagingGatewaySupport.setObservationConvention(
+					new DefaultMessageRequestReplyReceiverObservationConvention() {
+
+						@Override
+						public KeyValues getHighCardinalityKeyValues(MessageRequestReplyReceiverContext context) {
+							return KeyValues.of("test.message.id", context.getCarrier().getHeaders().getId().toString());
+						}
+
+					});
+			messagingGatewaySupport.setRequestChannel(queueChannel);
+			return messagingGatewaySupport;
+		}
+
 		@Bean
 		public PollableChannel queueChannel() {
 			return new QueueChannel();
@@ -149,4 +161,13 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner {
 
 	}
 
+	private static class TestMessagingGatewaySupport extends MessagingGatewaySupport {
+
+		@Nullable
+		Message<?> process(Message<?> request) {
+			return sendAndReceiveMessage(request);
+		}
+
+	}
+
 }

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java

@@ -26,6 +26,7 @@ import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 import javax.lang.model.SourceVersion;
@@ -268,6 +269,7 @@ public class IntegrationMBeanExporter extends MBeanExporter
 				.stream()
 				// If the source is proxied, we have to extract the target to expose as an MBean.
 				// The MetadataMBeanInfoAssembler does not support JDK dynamic proxies.
+				.filter(Predicate.not(MessageProducer.class::isInstance))
 				.map(this::extractTarget)
 				.map(IntegrationInboundManagement.class::cast)
 				.forEach(src -> this.sources.put(src.getComponentName(), src));

src/reference/asciidoc/metrics.adoc

@@ -150,7 +150,13 @@ registry.config().meterFilter(MeterFilter.deny(id ->
 
 Starting with version 6.0, Spring Integration utilizes a Micrometer Observation abstraction which can handle metrics as well as https://micrometer.io/docs/tracing[tracing] via appropriate `ObservationHandler` configuration.
 
-The observation handling is enabled on the `IntegrationManagement` components whenever an `ObservationRegistry` bean is present in the application context.
+The observation handling is enabled on the `IntegrationManagement` components whenever an `ObservationRegistry` bean is present in the application context and an `@EnableIntegrationManagement` is configured.
+To customize what set of components should be instrumented, an `observationPatterns()` attribute is exposed on the `@EnableIntegrationManagement` annotation.
+See its javadocs for a pattern matching algorithm.
+
+IMPORTANT: By default, none of the `IntegrationManagement` components are instrumented with an `ObservationRegistry` bean.
+Can be configured as `*` to match all components.
+
 The meters are not gathered in this case independently, but delegated to an appropriate `ObservationHandler` configured on the provided `ObservationRegistry`.
 
 An observation production on the `IntegrationManagement` components can be customized via `ObservationConvention` configuration.