From f23f3ea63de0f37471b8f2f2b2d267b8ab22ceac Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 28 Jul 2020 16:59:35 -0400 Subject: [PATCH] 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 Co-authored-by: Michel Jung --- .../integration/channel/QueueChannel.java | 39 +++++++++++++++++++ .../IntegrationManagementConfigurer.java | 6 +-- .../micrometer/MicrometerMetricsTests.java | 16 ++++++-- src/reference/asciidoc/metrics.adoc | 14 +++++++ 4 files changed, 69 insertions(+), 6 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java index 5211f7973a..09f5e89661 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java @@ -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(); + } + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java index 43d43d89de..fecbeb45f1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java @@ -285,8 +285,8 @@ public class IntegrationManagementConfigurer } private void injectCaptor() { - Map managed = this.applicationContext - .getBeansOfType(IntegrationManagement.class); + Map managed = + this.applicationContext.getBeansOfType(IntegrationManagement.class); for (Entry 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); 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 e7053fd93e..6ab50407ac 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 @@ -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 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 diff --git a/src/reference/asciidoc/metrics.adoc b/src/reference/asciidoc/metrics.adoc index 4a4381701a..7eeead7f4a 100644 --- a/src/reference/asciidoc/metrics.adoc +++ b/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:` +* `description`: `The size of queue channel` + +and + +* `name`: `spring.integration.channel.queue.remaining.capacity` +* `tag`: `type:channel` +* `tag`: `name:` +* `description`: `The remaining.capacity of queue channel` + [[mgmt-channel-features]] ==== `MessageChannel` Metric Features