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 <michel.jung89@gmail.com>
Co-authored-by: Michel Jung <michel.jung89@gmail.com>
This commit is contained in:
@@ -25,6 +25,8 @@ import java.util.concurrent.Semaphore;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.springframework.integration.core.MessageSelector;
|
||||
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 class QueueChannel extends AbstractPollableChannel implements QueueChanne
|
||||
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 class QueueChannel extends AbstractPollableChannel implements QueueChanne
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroy() {
|
||||
super.destroy();
|
||||
if (this.sizeGauge != null) {
|
||||
this.sizeGauge.remove();
|
||||
}
|
||||
if (this.remainingCapacityGauge != null) {
|
||||
this.remainingCapacityGauge.remove();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -285,8 +285,8 @@ public class IntegrationManagementConfigurer
|
||||
}
|
||||
|
||||
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 (!getOverrides(bean).loggingConfigured) {
|
||||
@@ -299,7 +299,7 @@ public class IntegrationManagementConfigurer
|
||||
@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);
|
||||
|
||||
@@ -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 class MicrometerMetricsTests {
|
||||
.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 @@ public class MicrometerMetricsTests {
|
||||
|
||||
@Bean
|
||||
public QueueChannel queue() {
|
||||
return new QueueChannel();
|
||||
return new QueueChannel(10);
|
||||
}
|
||||
|
||||
@Bean
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user