diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java index 90642506b8..1fd53704ca 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -103,7 +103,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel } } Message message = this.doReceive(timeout); - if (countsEnabled) { + if (countsEnabled && message != null) { getMetrics().afterReceive(); counted = true; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/AbstractMessageChannelMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/AbstractMessageChannelMetrics.java index 5fc123306e..3d0ecf5a6f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/AbstractMessageChannelMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/AbstractMessageChannelMetrics.java @@ -29,6 +29,7 @@ import io.micrometer.core.instrument.Timer; * Abstract base class for channel metrics implementations. * * @author Gary Russell + * * @since 4.2 * */ @@ -42,6 +43,10 @@ public abstract class AbstractMessageChannelMetrics implements ConfigurableMetri private final Counter errorCounter; + private final Counter receiveCounter; + + private final Counter receiveErrorCounter; + private volatile boolean fullStatsEnabled; /** @@ -49,25 +54,32 @@ public abstract class AbstractMessageChannelMetrics implements ConfigurableMetri * @param name the name. */ public AbstractMessageChannelMetrics(String name) { - this(name, null, null); + this(name, null, null, null, null); } /** - * Construct an instance with the provided name, timer and error counter. + * Construct an instance with the provided name, timer, error counter, receive counter + * and receive error counter. * A non-null timer requires a non-null error counter. When a timer is provided, * Micrometer metrics are used and the legacy metrics are not maintained. * @param name the name. * @param timer the timer. * @param errorCounter the error counter. + * @param receiveCounter the receive counter. + * @param receiveErrorCounter the receive error counter. * @since 5.0.2 */ - public AbstractMessageChannelMetrics(String name, Timer timer, Counter errorCounter) { + public AbstractMessageChannelMetrics(String name, Timer timer, Counter errorCounter, Counter receiveCounter, + Counter receiveErrorCounter) { + if (timer != null) { Assert.notNull(errorCounter, "'errorCounter' cannot be null if a timer is provided"); } this.name = name; this.timer = timer; this.errorCounter = errorCounter; + this.receiveCounter = receiveCounter; + this.receiveErrorCounter = receiveErrorCounter; } /** @@ -124,6 +136,26 @@ public abstract class AbstractMessageChannelMetrics implements ConfigurableMetri return this.errorCounter; } + /** + * Return the receive counter if Micrometer metrics are being used. + * @return the counter or null if Micrometer is not being used. + * @since 5.0.2 + */ + @Nullable + public Counter getReceiveCounter() { + return this.receiveCounter; + } + + /** + * Return the receive error counter if Micrometer metrics are being used. + * @return the counter or null if Micrometer is not being used. + * @since 5.0.2 + */ + @Nullable + public Counter getReceiveErrorCounter() { + return this.receiveErrorCounter; + } + public abstract int getSendCount(); public abstract long getSendCountLong(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/DefaultMessageChannelMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/DefaultMessageChannelMetrics.java index ecb7f19fc7..79b9fb8d2a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/DefaultMessageChannelMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/DefaultMessageChannelMetrics.java @@ -28,6 +28,7 @@ import io.micrometer.core.instrument.Timer; * @author Helena Edelson * @author Gary Russell * @author Ivan Krizsan + * * @since 2.0 */ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics { @@ -64,7 +65,7 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics * @param name the name. */ public DefaultMessageChannelMetrics(String name) { - this(name, null, null); + this(name, null, null, null, (Counter) null); } /** @@ -73,17 +74,21 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics * @param name the name. * @param timer a timer. * @param errorCounter a counter. + * @param receiveCounter a counter for receives. + * @param receiveErrorCounter a counter for receive errors. * @since 5.0.2 */ - public DefaultMessageChannelMetrics(String name, Timer timer, Counter errorCounter) { + public DefaultMessageChannelMetrics(String name, Timer timer, Counter errorCounter, Counter receiveCounter, + Counter receiveErrorCounter) { + this(name, new ExponentialMovingAverage(DEFAULT_MOVING_AVERAGE_WINDOW, 1000000.), - new ExponentialMovingAverageRate( - ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW, true), - new ExponentialMovingAverageRatio( - ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW, true), - new ExponentialMovingAverageRate( - ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW, true), - timer, errorCounter); + new ExponentialMovingAverageRate( + ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW, true), + new ExponentialMovingAverageRatio( + ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW, true), + new ExponentialMovingAverageRate( + ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW, true), + timer, errorCounter, receiveCounter, receiveErrorCounter); } /** @@ -100,7 +105,8 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics public DefaultMessageChannelMetrics(String name, ExponentialMovingAverage sendDuration, ExponentialMovingAverageRate sendErrorRate, ExponentialMovingAverageRatio sendSuccessRatio, ExponentialMovingAverageRate sendRate) { - this(name, sendDuration, sendErrorRate, sendSuccessRatio, sendRate, null, null); + + this(name, sendDuration, sendErrorRate, sendSuccessRatio, sendRate, null, null, null, null); } /** @@ -113,13 +119,17 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics * @param sendSuccessRatio an {@link ExponentialMovingAverageRatio} for calculating the success ratio. * @param sendRate an {@link ExponentialMovingAverageRate} for calculating the send rate. * @param timer a timer. - * @param errorCounter a counter. + * @param errorCounter a counter for sends. + * @param receiveCounter a counter for receives. + * @param receiveErrorCounter a counter for receive errors. * @since 5.0.2 */ public DefaultMessageChannelMetrics(String name, ExponentialMovingAverage sendDuration, ExponentialMovingAverageRate sendErrorRate, ExponentialMovingAverageRatio sendSuccessRatio, - ExponentialMovingAverageRate sendRate, Timer timer, Counter errorCounter) { - super(name, timer, errorCounter); + ExponentialMovingAverageRate sendRate, Timer timer, Counter errorCounter, Counter receiveCounter, + Counter receiveErrorCounter) { + + super(name, timer, errorCounter, receiveCounter, receiveErrorCounter); this.sendDuration = sendDuration; this.sendErrorRate = sendErrorRate; this.sendSuccessRatio = sendSuccessRatio; @@ -251,12 +261,22 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics @Override public void afterReceive() { - this.receiveCount.incrementAndGet(); + if (getReceiveCounter() != null) { + getReceiveCounter().increment(); + } + else { + this.receiveCount.incrementAndGet(); + } } @Override public void afterError() { - this.receiveErrorCount.incrementAndGet(); + if (getReceiveErrorCounter() != null) { + getReceiveErrorCounter().increment(); + } + else { + this.receiveErrorCount.incrementAndGet(); + } } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagementConfigurer.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagementConfigurer.java index 4e411b3fed..c0eb925734 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagementConfigurer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagementConfigurer.java @@ -236,7 +236,13 @@ public class IntegrationManagementConfigurer implements SmartInitializingSinglet @SuppressWarnings("unchecked") private void configureChannelMetrics(String name, MessageChannelMetrics bean) { - AbstractMessageChannelMetrics metrics = this.metricsFactory.createChannelMetrics(name); + AbstractMessageChannelMetrics metrics; + if (bean instanceof PollableChannelManagement) { + metrics = this.metricsFactory.createPollableChannelMetrics(name); + } + else { + metrics = this.metricsFactory.createChannelMetrics(name); + } Assert.state(metrics != null, "'metrics' must not be null"); ManagementOverrides overrides = bean.getOverrides(); Boolean enabled = PatternMatchUtils.smartMatch(name, this.enabledCountsPatterns); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/MetricsFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/MetricsFactory.java index 0b7bc3231e..5e188eb38e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/MetricsFactory.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/MetricsFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2018 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. @@ -16,7 +16,6 @@ package org.springframework.integration.support.management; - /** * Factories implementing this interface provide metric objects for message channels and * message handlers. @@ -34,6 +33,17 @@ public interface MetricsFactory { */ AbstractMessageChannelMetrics createChannelMetrics(String name); + /** + * Factory method to create an {@link AbstractMessageChannelMetrics} for + * a pollable channel. + * @param name the name. + * @return the metrics. + * @since 5.0.2 + */ + default AbstractMessageChannelMetrics createPollableChannelMetrics(String name) { + return createChannelMetrics(name); + } + /** * Factory method to create an {@link AbstractMessageHandlerMetrics}. * @param name the name. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsFactory.java index 6d81feffb1..de03dcb7ee 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsFactory.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2018 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. @@ -18,6 +18,11 @@ package org.springframework.integration.support.management.micrometer; import java.util.function.Function; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.SmartInitializingSingleton; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.integration.core.MessageSource; import org.springframework.integration.support.management.AbstractMessageChannelMetrics; import org.springframework.integration.support.management.AbstractMessageHandlerMetrics; import org.springframework.integration.support.management.DefaultMessageChannelMetrics; @@ -25,8 +30,11 @@ import org.springframework.integration.support.management.DefaultMessageHandlerM import org.springframework.integration.support.management.MessageSourceMetrics; import org.springframework.integration.support.management.MessageSourceMetricsConfigurer; import org.springframework.integration.support.management.MetricsFactory; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; +import io.micrometer.core.instrument.Gauge; import io.micrometer.core.instrument.MeterRegistry; /** @@ -37,27 +45,42 @@ import io.micrometer.core.instrument.MeterRegistry; * {@link MessageSourceMetrics}. * * @author Gary Russell + * * @since 5.0.2 + * * @see org.springframework.integration.support.management.IntegrationManagementConfigurer */ -public class MicrometerMetricsFactory implements MetricsFactory, MessageSourceMetricsConfigurer { +public class MicrometerMetricsFactory implements MetricsFactory, MessageSourceMetricsConfigurer, + ApplicationContextAware, SmartInitializingSingleton { private static final Function NO_TAGS = n -> new String[0]; private final MeterRegistry meterRegistry; + private ApplicationContext applicationContext; + private Function timerNameProvider = n -> n + ".timer"; private Function counterNameProvider = n -> n + ".counter"; private Function errorCounterNameProvider = n -> n + ".errorCounter"; + private Function receiveCounterNameProvider = n -> n + ".receive.counter"; + + private Function receiveErrorCounterNameProvider = n -> n + ".receive.errorCounter"; + private Function timerTagProvider = NO_TAGS; private Function counterTagProvider = NO_TAGS; private Function errorCounterTagProvider = NO_TAGS; + private Function receiveCounterTagProvider = NO_TAGS; + + private Function receiveErrorCounterTagProvider = NO_TAGS; + + private Function componentCountTagProvider = NO_TAGS; + /** * Construct an instance with the provided {@link MeterRegistry}. * @param meterRegistry the registry. @@ -67,6 +90,17 @@ public class MicrometerMetricsFactory implements MetricsFactory, MessageSourceMe this.meterRegistry = meterRegistry; } + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + @Override + public void afterSingletonsInstantiated() { + Assert.notNull(this.applicationContext, "An application context is required"); + registerComponentGauges(); + } + /** * Provide a function to generate a timer name for the bean name. * Default: "beanName.timer". @@ -97,6 +131,26 @@ public class MicrometerMetricsFactory implements MetricsFactory, MessageSourceMe this.errorCounterNameProvider = errorCounterNameProvider; } + /** + * Provide a function to generate a receive counter name for the bean name. + * Default: "beanName.counter". + * @param counterNameProvider the counterNameProvider to set + */ + public void setReceiveCounterNameProvider(Function counterNameProvider) { + Assert.notNull(counterNameProvider, "'counterNameProvider' cannot be null"); + this.receiveCounterNameProvider = counterNameProvider; + } + + /** + * Provide a function to generate a receive error counter name for the bean name. + * Default: "beanName.errorCounter". + * @param errorCounterNameProvider the counterNameProvider to set + */ + public void setReceiveErrorCounterNameProvider(Function errorCounterNameProvider) { + Assert.notNull(errorCounterNameProvider, "'errorCounterNameProvider' cannot be null"); + this.receiveErrorCounterNameProvider = errorCounterNameProvider; + } + /** * Provide a function to generate timer tags for the bean name. * Default: no tags. @@ -127,12 +181,53 @@ public class MicrometerMetricsFactory implements MetricsFactory, MessageSourceMe this.errorCounterTagProvider = counterTagProvider; } + /** + * Provide a function to generate receive counter tags for the bean name. + * Default: no tags. + * @param counterTagProvider the counterTagProvider to set + */ + public void setReceiveCounterTagProvider(Function counterTagProvider) { + Assert.notNull(counterTagProvider, "'counterTagProvider' cannot be null"); + this.receiveCounterTagProvider = counterTagProvider; + } + + /** + * Provide a function to generate receive error counter tags for the bean name. + * Default: no tags. + * @param counterTagProvider the counterTagProvider to set + */ + public void setReceiveErrorCounterTagProvider(Function counterTagProvider) { + Assert.notNull(counterTagProvider, "'counterTagProvider' cannot be null"); + this.receiveErrorCounterTagProvider = counterTagProvider; + } + + /** + * Provide a function to generate tags for component (channels, handlers, sources) gauges. + * Default: no tags. + * @param componentCountTagProvider the componentCountTagProvider to set + */ + public void setComponentCountTagProvider(Function componentCountTagProvider) { + this.componentCountTagProvider = componentCountTagProvider; + } + @Override public AbstractMessageChannelMetrics createChannelMetrics(String name) { return new DefaultMessageChannelMetrics(name, this.meterRegistry.timer(this.timerNameProvider.apply(name), this.timerTagProvider.apply(name)), this.meterRegistry.counter(this.errorCounterNameProvider.apply(name), - this.errorCounterTagProvider.apply(name))); + this.errorCounterTagProvider.apply(name)), null, null); + } + + @Override + public AbstractMessageChannelMetrics createPollableChannelMetrics(String name) { + return new DefaultMessageChannelMetrics(name, + this.meterRegistry.timer(this.timerNameProvider.apply(name), this.timerTagProvider.apply(name)), + this.meterRegistry.counter(this.errorCounterNameProvider.apply(name), + this.errorCounterTagProvider.apply(name)), + this.meterRegistry.counter(this.receiveCounterNameProvider.apply(name), + this.receiveCounterTagProvider.apply(name)), + this.meterRegistry.counter(this.receiveErrorCounterNameProvider.apply(name), + this.receiveErrorCounterTagProvider.apply(name))); } @Override @@ -149,4 +244,24 @@ public class MicrometerMetricsFactory implements MetricsFactory, MessageSourceMe this.counterTagProvider.apply(name))); } + private void registerComponentGauges() { + Gauge.Builder builder = Gauge.builder("spring.integration.channels", this, + (c) -> this.applicationContext.getBeansOfType(MessageChannel.class).size()); + builder.tags(this.componentCountTagProvider.apply("channels")) + .description("The Number of Message Channels") + .register(this.meterRegistry); + + builder = Gauge.builder("spring.integration.handlers", this, + (c) -> this.applicationContext.getBeansOfType(MessageHandler.class).size()); + builder.tags(this.componentCountTagProvider.apply("handlers")) + .description("The Number of Message Handlers") + .register(this.meterRegistry); + + builder = Gauge.builder("spring.integration.sources", this, + (c) -> this.applicationContext.getBeansOfType(MessageSource.class).size()); + builder.tags(this.componentCountTagProvider.apply("sources")) + .description("The number of Message Sources") + .register(this.meterRegistry); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsTests.java index 25de46ee5c..022c000b32 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsTests.java @@ -29,16 +29,21 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.AbstractPollableChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.config.EnableIntegrationManagement; import org.springframework.integration.core.MessageSource; import org.springframework.integration.endpoint.AbstractMessageSource; +import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; +import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringRunner; import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.Gauge; import io.micrometer.core.instrument.Meter; import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.Timer; @@ -46,6 +51,7 @@ import io.micrometer.core.instrument.simple.SimpleMeterRegistry; /** * @author Gary Russell + * * @since 5.0.2 * */ @@ -62,9 +68,16 @@ public class MicrometerMetricsTests { @Autowired private MessageSource source; + @Autowired + private QueueChannel queue; + + @Autowired + private PollableChannel badPoll; + @Test public void testSend() { - this.channel.send(new GenericMessage<>("foo")); + GenericMessage message = new GenericMessage<>("foo"); + this.channel.send(message); try { this.channel.send(this.source.receive()); fail("Expected exception"); @@ -72,26 +85,67 @@ public class MicrometerMetricsTests { catch (MessagingException e) { assertThat(e.getCause().getMessage()).isEqualTo("testErrorCount"); } + this.queue.send(message); + this.queue.send(message); + this.queue.receive(); + this.badPoll.send(message); + try { + this.badPoll.receive(); + fail("Expected exception"); + } + catch (RuntimeException e) { + assertThat(e.getMessage()).isEqualTo("badPoll"); + } List meters = this.meterRegistry.getMeters(); - int foundMeters = 0; + assertThat(meters.size()).isEqualTo(22); for (Meter meter : meters) { String name = meter.getId().getName(); switch (name) { case "channel.timer": case "micrometerMetricsTests.Config.service.serviceActivator.handler.timer": + case "queue.timer": assertThat(((Timer) meter).count()).isEqualTo(2L); - foundMeters++; + break; + case "badPoll.timer": + assertThat(((Timer) meter).count()).isEqualTo(1L); break; case "source.counter": case "channel.errorCounter": case "micrometerMetricsTests.Config.service.serviceActivator.handler.errorCounter": assertThat(((Counter) meter).count()).isEqualTo(1L); - foundMeters++; + break; + case "queue.receive.counter": + assertThat(((Counter) meter).count()).isEqualTo(1L); + break; + case "spring.integration.channels": + assertThat(((Gauge) meter).measure().iterator().next().getValue()).isEqualTo(5.0); + break; + case "spring.integration.handlers": + assertThat(((Gauge) meter).measure().iterator().next().getValue()).isEqualTo(2.0); + break; + case "spring.integration.sources": + assertThat(((Gauge) meter).measure().iterator().next().getValue()).isEqualTo(1.0); + break; + case "errorChannel.timer": + case "nullChannel.timer": + case "_org.springframework.integration.errorLogger.handler.timer": + assertThat(((Timer) meter).count()).isEqualTo(0L); + break; + case "queue.errorCounter": + case "nullChannel.errorCounter": + case "queue.receive.errorCounter": + case "_org.springframework.integration.errorLogger.handler.errorCounter": + case "errorChannel.errorCounter": + case "badPoll.errorCounter": + case "badPoll.receiveCounter": + assertThat(((Counter) meter).count()).isEqualTo(0L); + break; + case "badPoll.receiveErrorCounter": + assertThat(((Counter) meter).count()).isEqualTo(1L); break; default: } } - assertThat(foundMeters).isEqualTo(5); } @Configuration @@ -131,6 +185,28 @@ public class MicrometerMetricsTests { } }; } + + @Bean + public QueueChannel queue() { + return new QueueChannel(); + } + + @Bean + public PollableChannel badPoll() { + return new AbstractPollableChannel() { + + @Override + protected boolean doSend(Message message, long timeout) { + return true; + } + + @Override + protected Message doReceive(long timeout) { + throw new RuntimeException("badPoll"); + } + }; + } + } }