From 2918a1f1cd08554b2cae2896e57634819e22a67a Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 4 Nov 2026 16:45:36 +0100 Subject: [PATCH] Instrument pollable consumers with Spring Integration metrics Make DefaultPollableMessageSource an IntegrationManagement component so it receives a MetricsCaptor / ObservationRegistry. poll() now records spring.integration.receive for received messages and spring.integration.send (or an IntegrationObservation.HANDLER observation) around the handler invocation. Register the pollable bean under its concrete type so the IntegrationManagementConfigurer can discover it. Resolves #3328 Signed-off-by: Oleg Zhurakousky --- .../binder/DefaultPollableMessageSource.java | 134 +++++++++++++++++- .../BindableFunctionProxyFactory.java | 3 +- .../DefaultPollableMessageSourceTests.java | 27 ++++ 3 files changed, 161 insertions(+), 3 deletions(-) diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java index 6842fe089a..71e9054164 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java @@ -18,8 +18,11 @@ import java.util.ArrayList; import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.function.BiConsumer; +import io.micrometer.observation.ObservationRegistry; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.apache.commons.logging.Log; @@ -27,6 +30,7 @@ import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.NameMatchMethodPointcutAdvisor; +import org.springframework.beans.factory.BeanNameAware; import org.springframework.cloud.function.context.catalog.FunctionTypeUtils; import org.springframework.context.Lifecycle; import org.springframework.core.AttributeAccessor; @@ -42,6 +46,15 @@ import org.springframework.integration.support.DefaultErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageUtils; +import org.springframework.integration.support.management.IntegrationManagement; +import org.springframework.integration.support.management.metrics.CounterFacade; +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.DefaultMessageReceiverObservationConvention; +import org.springframework.integration.support.management.observation.IntegrationObservation; +import org.springframework.integration.support.management.observation.MessageReceiverContext; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -66,7 +79,7 @@ * @since 2.0 * */ -public class DefaultPollableMessageSource implements PollableMessageSource, Lifecycle { +public class DefaultPollableMessageSource implements PollableMessageSource, Lifecycle, IntegrationManagement, BeanNameAware { private static final Log log = LogFactory.getLog(DefaultPollableMessageSource.class); @@ -98,6 +111,18 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life private boolean running; + private final Set meters = ConcurrentHashMap.newKeySet(); + + private String beanName; + + private MetricsCaptor metricsCaptor; + + private ObservationRegistry observationRegistry = ObservationRegistry.NOOP; + + private CounterFacade receiveCounter; + + private TimerFacade successTimer; + /** * @param messageConverter instance of {@link SmartMessageConverter}. Can be null. */ @@ -106,6 +131,45 @@ public DefaultPollableMessageSource( this.messageConverter = messageConverter; } + @Override + public void setBeanName(String name) { + this.beanName = name; + } + + @Override + public String getComponentName() { + return this.beanName; + } + + @Override + public String getComponentType() { + return "pollable-message-source"; + } + + @Override + public void registerMetricsCaptor(MetricsCaptor captor) { + this.metricsCaptor = captor; + } + + @Override + public void registerObservationRegistry(ObservationRegistry registry) { + Assert.notNull(registry, "'observationRegistry' must not be null"); + this.observationRegistry = registry; + } + + @Override + public boolean isObserved() { + return !ObservationRegistry.NOOP.equals(this.observationRegistry); + } + + @Override + public void destroy() { + this.meters.forEach(MeterFacade::remove); + this.meters.clear(); + this.receiveCounter = null; + this.successTimer = null; + } + public void setSource(MessageSource source) { ProxyFactory pf = new ProxyFactory(source); @@ -235,10 +299,25 @@ public boolean poll(MessageHandler handler) { @Override public boolean poll(MessageHandler handler, ParameterizedTypeReference type) { - Message message = this.receive(type); + Message message; + try { + message = this.receive(type); + } + catch (RuntimeException ex) { + if (this.metricsCaptor != null) { + receiveCounter(false, ex.getClass().getSimpleName()).increment(); + } + throw ex; + } if (message == null) { return false; } + if (this.metricsCaptor != null) { + if (this.receiveCounter == null) { + this.receiveCounter = receiveCounter(true, "none"); + } + this.receiveCounter.increment(); + } AcknowledgmentCallback ackCallback = StaticMessageHeaderAccessor.getAcknowledgmentCallback(message); @@ -361,6 +440,33 @@ private Message receive(ParameterizedTypeReference type) { } private void doHandleMessage(MessageHandler handler, Message message) { + if (isObserved()) { + IntegrationObservation.HANDLER.observation(null, + DefaultMessageReceiverObservationConvention.INSTANCE, + () -> new MessageReceiverContext(message, getComponentName()), + this.observationRegistry) + .observe(() -> invokeHandler(handler, message)); + } + else if (this.metricsCaptor != null) { + SampleFacade sample = this.metricsCaptor.start(); + try { + invokeHandler(handler, message); + if (this.successTimer == null) { + this.successTimer = sendTimer(true, "none"); + } + sample.stop(this.successTimer); + } + catch (RuntimeException ex) { + sample.stop(sendTimer(false, ex.getClass().getSimpleName())); + throw ex; + } + } + else { + invokeHandler(handler, message); + } + } + + private void invokeHandler(MessageHandler handler, Message message) { try { handler.handleMessage(message); } @@ -369,6 +475,30 @@ private void doHandleMessage(MessageHandler handler, Message message) { } } + private CounterFacade receiveCounter(boolean success, String exception) { + CounterFacade counter = this.metricsCaptor.counterBuilder(RECEIVE_COUNTER_NAME) + .tag("name", getComponentName()) + .tag("type", "source") + .tag("result", success ? "success" : "failure") + .tag("exception", exception) + .description("Messages received") + .build(); + this.meters.add(counter); + return counter; + } + + private TimerFacade sendTimer(boolean success, String exception) { + TimerFacade timer = this.metricsCaptor.timerBuilder(SEND_TIMER_NAME) + .tag("type", "handler") + .tag("name", getComponentName()) + .tag("result", success ? "success" : "failure") + .tag("exception", exception) + .description("Send processing time") + .build(); + this.meters.add(timer); + return timer; + } + /** * If there's a retry template, it will set the attribute holder via the listener. If * there's no retry template, but there's an error channel, we create a new attribute diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java index 943d8ace76..c961264bd1 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java @@ -186,7 +186,8 @@ private void createInput(String name) { if (this.supportedBindableFeatures.isPollable()) { PollableMessageSource pollableSource = (PollableMessageSource) getBindingTargetFactory(PollableMessageSource.class).createInput(name); if (context != null && !context.containsBean(name)) { - context.registerBean(name, PollableMessageSource.class, () -> pollableSource); + // register with the concrete type so that bean type matching (e.g., IntegrationManagement) works + context.registerBean(name, (Class) pollableSource.getClass(), () -> pollableSource); } this.inputHolders.put(name, new BoundTargetHolder(pollableSource, true)); } diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSourceTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSourceTests.java index 8ef5255115..65200d2fdf 100644 --- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSourceTests.java +++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSourceTests.java @@ -20,9 +20,11 @@ import java.util.List; import java.util.concurrent.atomic.AtomicInteger; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import org.junit.jupiter.api.Test; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.support.management.micrometer.MicrometerMetricsCaptor; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -30,6 +32,7 @@ import org.springframework.messaging.support.MessageBuilder; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** * Verifies that {@link DefaultPollableMessageSource} invokes the full @@ -42,6 +45,30 @@ class DefaultPollableMessageSourceTests { private final AtomicInteger receiveCount = new AtomicInteger(); + @Test + void pollRecordsSpringIntegrationMetrics() { + SimpleMeterRegistry registry = new SimpleMeterRegistry(); + DefaultPollableMessageSource source = newSource(); + source.setBeanName("pollableInput-in-0"); + source.registerMetricsCaptor(new MicrometerMetricsCaptor(registry)); + source.setSource(this::message); + + assertThat(source.poll(noopHandler())).isTrue(); + assertThatThrownBy(() -> source.poll(m -> { + throw new IllegalStateException("boom"); + })).isInstanceOf(RuntimeException.class); + + assertThat(registry.find("spring.integration.receive") + .tag("name", "pollableInput-in-0").tag("type", "source").counter().count()).isEqualTo(2.0); + assertThat(registry.find("spring.integration.send").tag("type", "handler") + .tag("name", "pollableInput-in-0").tag("result", "success").timer().count()).isEqualTo(1); + assertThat(registry.find("spring.integration.send").tag("type", "handler") + .tag("name", "pollableInput-in-0").tag("result", "failure").timer().count()).isEqualTo(1); + + source.destroy(); + assertThat(registry.find("spring.integration.send").timers()).isEmpty(); + } + @Test void successfulPollInvokesFullInterceptorLifecycleInOrder() { DefaultPollableMessageSource source = newSource();