Add gauges for queue channel size (#3349)

* Add gauges for queue channel size

The `QueueChannel` provides a current size and remaining capacity metrics

* Add Micrometer gauges into `QueueChannel` to expose the current values
of the size and remaining capacity

**Cherry-pick to 5.3.x, 5.2.x & 5.1.x**

* * Revert `@SuppressWarnings("unchecked")` for test
* Document new gauges for queue channel

* * Fix IntegrationManagementConfigurer for NPE on `metricsCaptor`

* Fix wording in meter descriptions

Co-authored-by: Michel Jung <[email protected]>

Co-authored-by: Michel Jung <[email protected]>

Author: 2 people

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

@@ -26,6 +26,8 @@
 
 import org.springframework.integration.core.MessageSelector;
 import org.springframework.integration.support.management.QueueChannelManagement;
+import org.springframework.integration.support.management.metrics.GaugeFacade;
+import org.springframework.integration.support.management.metrics.MetricsCaptor;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
@@ -49,6 +51,12 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne
 
 	protected final Semaphore queueSemaphore = new Semaphore(0); // NOSONAR final
 
+	@Nullable
+	private GaugeFacade sizeGauge;
+
+	@Nullable
+	private GaugeFacade remainingCapacityGauge;
+
 	/**
 	 * Create a channel with the specified queue.
 	 *
@@ -79,6 +87,26 @@ public QueueChannel() {
 		this(new LinkedBlockingQueue<>());
 	}
 
+	@Override
+	public void registerMetricsCaptor(MetricsCaptor metricsCaptor) {
+		super.registerMetricsCaptor(metricsCaptor);
+		this.sizeGauge =
+				metricsCaptor.gaugeBuilder("spring.integration.channel.queue.size", this,
+						(channel) -> getQueueSize())
+						.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+						.tag("type", "channel")
+						.description("The size of the queue channel")
+						.build();
+
+		this.remainingCapacityGauge =
+				metricsCaptor.gaugeBuilder("spring.integration.channel.queue.remaining.capacity", this,
+						(channel) -> getRemainingCapacity())
+						.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+						.tag("type", "channel")
+						.description("The remaining capacity of the queue channel")
+						.build();
+	}
+
 	@Override
 	protected boolean doSend(Message<?> message, long timeout) {
 		Assert.notNull(message, "'message' must not be null");
@@ -207,4 +235,15 @@ public int getRemainingCapacity() {
 		}
 	}
 
+	@Override
+	public void destroy() {
+		super.destroy();
+		if (this.sizeGauge != null) {
+			this.sizeGauge.remove();
+		}
+		if (this.remainingCapacityGauge != null) {
+			this.remainingCapacityGauge.remove();
+		}
+	}
+
 }

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

@@ -288,8 +288,8 @@ public void afterSingletonsInstantiated() {
 	}
 
 	private void injectCaptor() {
-		Map<String, IntegrationManagement> managed = this.applicationContext
-				.getBeansOfType(IntegrationManagement.class);
+		Map<String, IntegrationManagement> managed =
+				this.applicationContext.getBeansOfType(IntegrationManagement.class);
 		for (Entry<String, IntegrationManagement> entry : managed.entrySet()) {
 			IntegrationManagement bean = entry.getValue();
 			if (!bean.getOverrides().loggingConfigured) {
@@ -302,7 +302,7 @@ private void injectCaptor() {
 	@Override
 	public Object postProcessAfterInitialization(Object bean, String name) throws BeansException {
 		if (this.singletonsInstantiated) {
-			if (bean instanceof IntegrationManagement) {
+			if (this.metricsCaptor != null && bean instanceof IntegrationManagement) {
 				((IntegrationManagement) bean).registerMetricsCaptor(this.metricsCaptor);
 			}
 			return doConfigureMetrics(bean, name);

spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -94,7 +94,7 @@ public class MicrometerMetricsTests {
 
 	@SuppressWarnings("unchecked")
 	@Test
-	public void testSend() {
+	public void testMicrometerMetrics() {
 		GenericMessage<String> message = new GenericMessage<>("foo");
 		this.channel.send(message);
 		assertThatExceptionOfType(MessagingException.class)
@@ -162,6 +162,16 @@ public void testSend() {
 				.tag("result", "success")
 				.counter().count()).isEqualTo(1);
 
+		this.queue.send(message);
+
+		assertThat(registry.get("spring.integration.channel.queue.size")
+				.tag("name", "queue")
+				.gauge().value()).isEqualTo(2d);
+
+		assertThat(registry.get("spring.integration.channel.queue.remaining.capacity")
+				.tag("name", "queue")
+				.gauge().value()).isEqualTo(8d);
+
 		assertThat(registry.get("spring.integration.send")
 				.tag("name", "nullChannel")
 				.tag("result", "success")
@@ -260,7 +270,7 @@ protected Object doReceive() {
 
 		@Bean
 		public QueueChannel queue() {
-			return new QueueChannel();
+			return new QueueChannel(10);
 		}
 
 		@Bean

src/reference/asciidoc/metrics.adoc

@@ -165,6 +165,20 @@ It is possible to customize the names and tags of `Meters` created by integratio
 The https://github.com/spring-projects/spring-integration/blob/master/spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerCustomMetricsTests.java[MicrometerCustomMetricsTests] test case shows a simple example of how to do that.
 You can also further customize the meters by overloading the `build()` methods on builder subclasses.
 
+Starting with version 5.1.13, the `QueueChannel` exposes Micrometer gauges for queue size and remaining capacity:
+
+* `name`: `spring.integration.channel.queue.size`
+* `tag`: `type:channel`
+* `tag`: `name:<componentName>`
+* `description`: `The size of queue channel`
+
+and
+
+* `name`: `spring.integration.channel.queue.remaining.capacity`
+* `tag`: `type:channel`
+* `tag`: `name:<componentName>`
+* `description`: `The remaining.capacity of queue channel`
+
 [[mgmt-channel-features]]
 ==== `MessageChannel` Metric Features