Add observation for message channels

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.channel;
 
 import java.util.ArrayDeque;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.Deque;
@@ -26,6 +27,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 
+import io.micrometer.observation.ObservationRegistry;
+
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.core.OrderComparator;
 import org.springframework.core.log.LogAccessor;
@@ -34,13 +37,18 @@
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.history.MessageHistory;
+import org.springframework.integration.support.MutableMessage;
 import org.springframework.integration.support.management.IntegrationManagedResource;
 import org.springframework.integration.support.management.IntegrationManagement;
 import org.springframework.integration.support.management.TrackableComponent;
 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.DefaultMessageSenderObservationConvention;
+import org.springframework.integration.support.management.observation.IntegrationObservation;
+import org.springframework.integration.support.management.observation.MessageSenderContext;
+import org.springframework.integration.support.management.observation.MessageSenderObservationConvention;
 import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
@@ -75,22 +83,27 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 
 	protected final Set<MeterFacade> meters = ConcurrentHashMap.newKeySet(); // NOSONAR
 
-	private volatile boolean shouldTrack = false;
+	private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
+
+	@Nullable
+	private MessageSenderObservationConvention observationConvention;
 
-	private volatile Class<?>[] datatypes = new Class<?>[0];
+	private boolean shouldTrack = false;
 
-	private volatile String fullChannelName;
+	private Class<?>[] datatypes = new Class<?>[0];
 
-	private volatile MessageConverter messageConverter;
+	private MessageConverter messageConverter;
 
-	private volatile boolean loggingEnabled = true;
+	private boolean loggingEnabled = true;
 
 	private MetricsCaptor metricsCaptor;
 
 	private TimerFacade successTimer;
 
 	private TimerFacade failureTimer;
 
+	private volatile String fullChannelName;
+
 	@Override
 	public String getComponentType() {
 		return "channel";
@@ -138,10 +151,7 @@ public void setLoggingEnabled(boolean loggingEnabled) {
 	 * @see #setMessageConverter(MessageConverter)
 	 */
 	public void setDatatypes(Class<?>... datatypes) {
-		this.datatypes =
-				(datatypes != null && datatypes.length > 0)
-						? datatypes
-						: new Class<?>[0];
+		this.datatypes = Arrays.copyOf(datatypes, datatypes.length);
 	}
 
 	/**
@@ -192,6 +202,10 @@ public void setMessageConverter(MessageConverter messageConverter) {
 		this.messageConverter = messageConverter;
 	}
 
+	public void setObservationConvention(@Nullable MessageSenderObservationConvention observationConvention) {
+		this.observationConvention = observationConvention;
+	}
+
 	/**
 	 * Return a read-only list of the configured interceptors.
 	 */
@@ -224,6 +238,12 @@ public ManagementOverrides getOverrides() {
 		return this.managementOverrides;
 	}
 
+	@Override
+	public void registerObservationRegistry(ObservationRegistry observationRegistry) {
+		Assert.notNull(observationRegistry, "'observationRegistry' must not be null");
+		this.observationRegistry = observationRegistry;
+	}
+
 	@Override
 	protected void onInit() {
 		super.onInit();
@@ -276,15 +296,14 @@ public boolean send(Message<?> message) {
 	 * Send a message on this channel. If the channel is at capacity, this
 	 * method will block until either the timeout occurs or the sending thread
 	 * is interrupted. If the specified timeout is 0, the method will return
-	 * immediately. If less than zero, it will block indefinitely (see
-	 * {@link #send(Message)}).
+	 * immediately. If less than zero, it will block indefinitely (see {@link #send(Message)}).
 	 * @param messageArg the Message to send
 	 * @param timeout the timeout in milliseconds
 	 * @return <code>true</code> if the message is sent successfully,
 	 * <code>false</code> if the message cannot be sent within the allotted
 	 * time or the sending thread is interrupted.
 	 */
-	@Override // NOSONAR complexity
+	@Override
 	public boolean send(Message<?> messageArg, long timeout) {
 		Assert.notNull(messageArg, "message must not be null");
 		Assert.notNull(messageArg.getPayload(), "message payload must not be null");
@@ -293,11 +312,45 @@ public boolean send(Message<?> messageArg, long timeout) {
 			message = MessageHistory.write(message, this, getMessageBuilderFactory());
 		}
 
+		if (!ObservationRegistry.NOOP.equals(this.observationRegistry)) {
+			return sendWithObservation(message, timeout);
+		}
+		else if (this.metricsCaptor != null) {
+			return sendWithMetrics(message, timeout);
+		}
+		else {
+			return sendInternal(message, timeout);
+		}
+	}
+
+	private boolean sendWithObservation(Message<?> message, long timeout) {
+		MutableMessage<?> messageToSend = MutableMessage.of(message);
+		MessageSenderContext context = new MessageSenderContext(messageToSend, getComponentName());
+		return IntegrationObservation.PRODUCER.observation(
+						this.observationConvention,
+						DefaultMessageSenderObservationConvention.INSTANCE,
+						() -> context,
+						this.observationRegistry)
+				.observe(() -> sendInternal(messageToSend, timeout));
+	}
+
+	private boolean sendWithMetrics(Message<?> message, long timeout) {
+		SampleFacade sample = this.metricsCaptor.start();
+		try {
+			boolean sent = sendInternal(message, timeout);
+			sample.stop(sendTimer(sent));
+			return sent;
+		}
+		catch (RuntimeException ex) {
+			sample.stop(buildSendTimer(false, ex.getClass().getSimpleName()));
+			throw ex;
+		}
+	}
+
+	private boolean sendInternal(Message<?> message, long timeout) {
 		Deque<ChannelInterceptor> interceptorStack = null;
 		boolean sent = false;
-		boolean metricsProcessed = false;
 		ChannelInterceptorList interceptorList = this.interceptors;
-		SampleFacade sample = null;
 		try {
 			message = convertPayloadIfNecessary(message);
 			boolean debugEnabled = this.loggingEnabled && this.logger.isDebugEnabled();
@@ -311,14 +364,8 @@ public boolean send(Message<?> messageArg, long timeout) {
 					return false;
 				}
 			}
-			if (this.metricsCaptor != null) {
-				sample = this.metricsCaptor.start();
-			}
+
 			sent = doSend(message, timeout);
-			if (sample != null) {
-				sample.stop(sendTimer(sent));
-			}
-			metricsProcessed = true;
 
 			if (debugEnabled) {
 				logger.debug("postSend (sent=" + sent + ") on channel '" + this + "', message: " + message);
@@ -330,9 +377,6 @@ public boolean send(Message<?> messageArg, long timeout) {
 			return sent;
 		}
 		catch (Exception ex) {
-			if (!metricsProcessed && sample != null) {
-				sample.stop(buildSendTimer(false, ex.getClass().getSimpleName()));
-			}
 			if (interceptorStack != null) {
 				interceptorList.afterSendCompletion(message, this, sent, ex, interceptorStack);
 			}
@@ -411,7 +455,7 @@ private Message<?> convertPayloadIfNecessary(Message<?> message) {
 	 * accepted or the blocking thread is interrupted.
 	 * @param message The message.
 	 * @param timeout The timeout.
-	 * @return true if the send was successful.
+	 * @return true if the {@code send} was successful.
 	 */
 	protected abstract boolean doSend(Message<?> message, long timeout);
 

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

@@ -113,8 +113,7 @@ public boolean equals(Object obj) {
 		if (this == obj) {
 			return true;
 		}
-		if (obj instanceof MutableMessage<?>) {
-			MutableMessage<?> other = (MutableMessage<?>) obj;
+		if (obj instanceof MutableMessage<?> other) {
 			UUID thisId = this.headers.getId();
 			UUID otherId = other.headers.getId();
 			return (ObjectUtils.nullSafeEquals(thisId, otherId) &&
@@ -123,4 +122,18 @@ public boolean equals(Object obj) {
 		return false;
 	}
 
+	/**
+	 * Build a new {@link MutableMessage} based on the provided message
+	 * if that one is not already a {@link MutableMessage}.
+	 * @param message the message to build from.
+	 * @return new {@link MutableMessage}.
+	 * @since 6.0
+	 */
+	public static MutableMessage<?> of(Message<?> message) {
+		if (message instanceof MutableMessage) {
+			return (MutableMessage<?>) message;
+		}
+		return new MutableMessage<>(message.getPayload(), message.getHeaders());
+	}
+
 }

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

@@ -0,0 +1,46 @@
+/*
+ * 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 MessageSenderObservationConvention} implementation.
+ * Provides low cardinalities as a {@link IntegrationObservation.ProducerTags} values.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class DefaultMessageSenderObservationConvention implements MessageSenderObservationConvention {
+
+	/**
+	 * A shared singleton instance for {@link DefaultMessageSenderObservationConvention}.
+	 */
+	public static final DefaultMessageSenderObservationConvention INSTANCE =
+			new DefaultMessageSenderObservationConvention();
+
+
+	@Override
+	public KeyValues getLowCardinalityKeyValues(MessageSenderContext context) {
+		return KeyValues
+				// See IntegrationObservation.ProducerTags.COMPONENT_NAME - to avoid class tangle
+				.of("spring.integration.name", context.getProducerName())
+				// See IntegrationObservation.ProducerTags.COMPONENT_TYPE - to avoid class tangle
+				.and("spring.integration.type", "producer");
+	}
+}

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

@@ -71,6 +71,27 @@ public KeyName[] getLowCardinalityKeyNames() {
 			return GatewayTags.values();
 		}
 
+	},
+
+	/**
+	 * Observation for message producers, e.g. channels.
+	 */
+	PRODUCER {
+		@Override
+		public String getPrefix() {
+			return "spring.integration.";
+		}
+
+		@Override
+		public Class<DefaultMessageSenderObservationConvention> getDefaultConvention() {
+			return DefaultMessageSenderObservationConvention.class;
+		}
+
+		@Override
+		public KeyName[] getLowCardinalityKeyNames() {
+			return ProducerTags.values();
+		}
+
 	};
 
 	/**
@@ -141,4 +162,33 @@ public String asString() {
 
 	}
 
+	/**
+	 * Key names for message producer observations.
+	 */
+	public enum ProducerTags implements KeyName {
+
+		/**
+		 * Name of the message handler component.
+		 */
+		COMPONENT_NAME {
+			@Override
+			public String asString() {
+				return "spring.integration.name";
+			}
+
+		},
+
+		/**
+		 * Type of the component - 'producer'.
+		 */
+		COMPONENT_TYPE {
+			@Override
+			public String asString() {
+				return "spring.integration.type";
+			}
+
+		}
+
+	}
+
 }

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

@@ -30,9 +30,23 @@
  */
 public class MessageSenderContext extends SenderContext<MutableMessage<?>> {
 
-	public MessageSenderContext(MutableMessage<?> message) {
+	private final MutableMessage<?> message;
+
+	private final String producerName;
+
+	public MessageSenderContext(MutableMessage<?> message, String producerName) {
 		super((carrier, key, value) -> carrier.getHeaders().put(key, value));
-		setCarrier(message);
+		this.message = message;
+		this.producerName = producerName;
+	}
+
+	@Override
+	public MutableMessage<?> getCarrier() {
+		return this.message;
+	}
+
+	public String getProducerName() {
+		return this.producerName;
 	}
 
 }