This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 7f97f3070 fix(metrics): avoid deriving queue lag from cloud aggregates
(#6184)
7f97f3070 is described below
commit 7f97f3070b93855e8c1477435db1a11b6491434c
Author: zmuxuny <[email protected]>
AuthorDate: Mon Oct 12 00:17:32 2026 -0700
fix(metrics): avoid deriving queue lag from cloud aggregates (#6184)
`CloudRocketMqBusinessMetricsCollector.java:91-96` published
`max(diffTotal)` over the provider's progress rows as an AVAILABLE
`consumer.lag.max_queue`. Those rows are subscription-level aggregates rather
than queues: the Aliyun and Tencent converters fill
`brokerOffset`/`consumerOffset` with `QueueProgressVO.UNKNOWN_OFFSET`
(`AliyunConverters.java:212-213` and `:226-227`,
`TencentInstanceProvider.java:548-549`) and put the topic-level lag into
`diffTotal`. A topic aggregate of 80 th [...]
Six tests plus two parameterized vendor instances cover the change, driving
the real evaluator and state machine rather than mocks. Reverting to AVAILABLE
makes `doesNotPublishTopicTotalsAsQueueMaximumTest` evaluate 80 > 50 as met and
makes `unsupportedQueueMaximumDoesNotClearAnActiveAlertTest` lose its `NONE`
transition; `collectsAggregateLagWithoutInventingQueueMaximumTest`,
`emptyProgressStillDoesNotProvideAQueueMaximumTest` and
`failedGroupProgressRemainsUnavailableTest` keep `con [...]
Known gap, tracked for a maintainer follow-up:
`NativeAlertMetricCatalogService.java:41-44` still lists
`consumer.lag.max_queue` under `BUSINESS_CLOUD`, so `validate()` (`:79-83`)
keeps accepting cloud queue rules and the rule editor keeps offering a metric
that can now only ever produce UNSUPPORTED —
`docs/studio-native-alerting-design.md:147` requires the catalog to be
capability-aware and hide unsupported metrics, and the metric table row at
`:142` still names Aliyun and Tencent as [...]
---
.../CloudRocketMqBusinessMetricsCollector.java | 9 +-
.../CloudRocketMqBusinessMetricsCollectorTest.java | 142 ++++++++++++++++++++-
2 files changed, 140 insertions(+), 11 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqBusinessMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqBusinessMetricsCollector.java
index 4cc1d6161..8503c84a3 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqBusinessMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqBusinessMetricsCollector.java
@@ -88,12 +88,13 @@ public class CloudRocketMqBusinessMetricsCollector
implements BusinessMetricsCol
try {
List<QueueProgressVO> progress =
provider.getGroupProgress(instance.getName(), group.getName());
double totalLag = progress.stream().mapToDouble(row ->
Math.max(0L, row.getDiffTotal())).sum();
- double maxQueueLag = progress.stream().mapToDouble(row ->
Math.max(0L, row.getDiffTotal())).max()
- .orElse(0D);
List<MetricSample> samples = new ArrayList<>();
samples.add(available(CONSUMER_LAG_TOTAL, instance,
group.getClusterId(), labels, totalLag, collectedAt));
- samples.add(available(CONSUMER_LAG_MAX_QUEUE, instance,
group.getClusterId(), labels, maxQueueLag,
- collectedAt));
+ // Aliyun and Tencent expose topic/group aggregates, not
individual queues.
+ // Keep a non-numeric sample so an existing queue alert is not
falsely cleared.
+ samples.add(new MetricSample(CONSUMER_LAG_MAX_QUEUE,
AlertDomain.BUSINESS, instance.getName(),
+ group.getClusterId(), labels, null,
MetricAvailability.UNSUPPORTED, collectedAt,
+ "CLOUD_QUEUE_LAG_UNSUPPORTED"));
progress.stream().filter(row -> row.getTopic() != null &&
!row.getTopic().isBlank())
.collect(java.util.stream.Collectors.groupingBy(QueueProgressVO::getTopic,
java.util.stream.Collectors.summingLong(
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqBusinessMetricsCollectorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqBusinessMetricsCollectorTest.java
index c7ea40b4d..43dcbb536 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqBusinessMetricsCollectorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqBusinessMetricsCollectorTest.java
@@ -25,8 +25,18 @@ import
org.apache.rocketmq.studio.instance.group.QueueProgressVO;
import org.apache.rocketmq.studio.provider.InstanceProvider;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.apache.rocketmq.studio.ops.alert.AlertDomain;
+import org.apache.rocketmq.studio.ops.alert.AlertRuleEvaluator;
+import org.apache.rocketmq.studio.ops.alert.AlertRuleVO;
import java.util.List;
+import java.time.Instant;
+import org.apache.rocketmq.studio.ops.alert.AlertRuleState;
+import org.apache.rocketmq.studio.ops.alert.AlertStateMachine;
+import org.apache.rocketmq.studio.ops.alert.AlertStateStatus;
+import org.apache.rocketmq.studio.ops.alert.AlertStateTransition;
import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
@@ -35,7 +45,7 @@ import static org.mockito.Mockito.when;
class CloudRocketMqBusinessMetricsCollectorTest {
@Test
- void collectsTotalAndMaximumLagFromACloudProviderTest() {
+ void collectsAggregateLagWithoutInventingQueueMaximumTest() {
InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
InstanceProvider provider = mock(InstanceProvider.class);
InstanceVO instance =
InstanceVO.builder().name("aliyun").vendor(InstanceVendor.ALIYUN).build();
@@ -45,19 +55,137 @@ class CloudRocketMqBusinessMetricsCollectorTest {
when(registry.byInstanceId("aliyun")).thenReturn(Optional.of(provider));
when(provider.listConsumerGroups("aliyun",
null)).thenReturn(List.of(group));
when(provider.getGroupProgress("aliyun", "orders")).thenReturn(List.of(
-
QueueProgressVO.builder().topic("orders-topic").diffTotal(12).build(),
QueueProgressVO.builder()
- .topic("orders-topic").diffTotal(30).build()));
+
QueueProgressVO.builder().topic("orders-topic").broker("topic:orders-topic")
+
.brokerOffset(-1).consumerOffset(-1).diffTotal(12).build(),
QueueProgressVO.builder()
+ .topic("payments-topic").broker("topic:payments-topic")
+
.brokerOffset(-1).consumerOffset(-1).diffTotal(30).build()));
List<MetricSample> samples = new
CloudRocketMqBusinessMetricsCollector(registry).collect(instance);
- assertThat(samples).hasSize(3).allSatisfy(sample ->
assertThat(sample.availability())
- .isEqualTo(MetricAvailability.AVAILABLE));
+ assertThat(samples).hasSize(4);
+ assertThat(samples).filteredOn(sample ->
!sample.metricKey().equals("consumer.lag.max_queue"))
+ .allSatisfy(sample ->
assertThat(sample.availability()).isEqualTo(MetricAvailability.AVAILABLE));
assertThat(samples).filteredOn(sample ->
sample.metricKey().equals("consumer.lag.total"))
.singleElement().extracting(MetricSample::value).isEqualTo(42D);
assertThat(samples).filteredOn(sample ->
sample.metricKey().equals("consumer.lag.max_queue"))
-
.singleElement().extracting(MetricSample::value).isEqualTo(30D);
+ .singleElement().satisfies(sample -> {
+
assertThat(sample.availability()).isEqualTo(MetricAvailability.UNSUPPORTED);
+ assertThat(sample.value()).isNull();
+ assertThat(sample.clusterId()).isEqualTo("cloud-a");
+
assertThat(sample.unavailableReason()).isEqualTo("CLOUD_QUEUE_LAG_UNSUPPORTED");
+ });
assertThat(samples).filteredOn(sample ->
sample.metricKey().equals("topic.backlog.total"))
-
.singleElement().extracting(MetricSample::value).isEqualTo(42D);
+
.extracting(MetricSample::value).containsExactlyInAnyOrder(12D, 30D);
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = InstanceVendor.class, names = {"ALIYUN", "TENCENT"})
+ void doesNotPublishTopicTotalsAsQueueMaximumTest(InstanceVendor vendor) {
+ QueueProgressVO aggregate =
QueueProgressVO.builder().topic("orders-topic")
+ .broker("topic:orders-topic").queueId(0)
+
.brokerOffset(QueueProgressVO.UNKNOWN_OFFSET).consumerOffset(QueueProgressVO.UNKNOWN_OFFSET)
+ .diffTotal(80).build();
+ List<MetricSample> samples = collectCloudProgress(vendor,
List.of(aggregate));
+ MetricSample maximum = samples.stream().filter(sample ->
sample.metricKey().equals("consumer.lag.max_queue"))
+ .findFirst().orElseThrow();
+ AlertRuleVO rule = AlertRuleVO.builder().domain(AlertDomain.BUSINESS)
+
.metric("consumer.lag.max_queue").operator(">").threshold(50).build();
+
+ // Two queues of 40 each have topic lag 80, but neither queue breaches
50.
+ assertThat(new AlertRuleEvaluator().evaluate(rule,
maximum).conditionMet()).isFalse();
+
assertThat(maximum.availability()).isEqualTo(MetricAvailability.UNSUPPORTED);
+ assertThat(maximum.value()).isNull();
+ assertThat(maximum.labels()).containsEntry("consumerGroup", "orders");
+ assertThat(samples).filteredOn(sample ->
sample.metricKey().equals("consumer.lag.total"))
+ .singleElement().satisfies(sample -> {
+ assertThat(sample.value()).isEqualTo(80D);
+
assertThat(sample.availability()).isEqualTo(MetricAvailability.AVAILABLE);
+ });
+ assertThat(samples).filteredOn(sample ->
sample.metricKey().equals("topic.backlog.total"))
+
.singleElement().extracting(MetricSample::value).isEqualTo(80D);
+ }
+
+ @Test
+ void doesNotPublishAliyunTotalFallbackAsQueueMaximumTest() {
+ QueueProgressVO aggregate =
QueueProgressVO.builder().broker("total").queueId(0)
+
.brokerOffset(QueueProgressVO.UNKNOWN_OFFSET).consumerOffset(QueueProgressVO.UNKNOWN_OFFSET)
+ .diffTotal(80).build();
+ List<MetricSample> samples =
collectCloudProgress(InstanceVendor.ALIYUN, List.of(aggregate));
+
+ assertThat(samples).filteredOn(sample ->
sample.metricKey().equals("consumer.lag.max_queue"))
+ .singleElement().satisfies(sample -> {
+
assertThat(sample.availability()).isEqualTo(MetricAvailability.UNSUPPORTED);
+ assertThat(sample.value()).isNull();
+ });
+ assertThat(samples).filteredOn(sample ->
sample.metricKey().equals("consumer.lag.total"))
+
.singleElement().extracting(MetricSample::value).isEqualTo(80D);
+ assertThat(samples).noneMatch(sample ->
sample.metricKey().equals("topic.backlog.total"));
+ }
+
+ @Test
+ void unsupportedQueueMaximumDoesNotClearAnActiveAlertTest() {
+ List<MetricSample> samples =
collectCloudProgress(InstanceVendor.TENCENT, List.of(
+
QueueProgressVO.builder().topic("orders-topic").broker("topic:orders-topic")
+
.brokerOffset(QueueProgressVO.UNKNOWN_OFFSET).consumerOffset(QueueProgressVO.UNKNOWN_OFFSET)
+ .diffTotal(0).build()));
+ MetricSample maximum = samples.stream().filter(sample ->
sample.metricKey().equals("consumer.lag.max_queue"))
+ .findFirst().orElseThrow();
+ AlertRuleVO rule = AlertRuleVO.builder().domain(AlertDomain.BUSINESS)
+
.metric("consumer.lag.max_queue").operator(">").threshold(50).build();
+ Instant previously = maximum.collectedAt().minusSeconds(60);
+ AlertRuleState firing = new AlertRuleState(AlertStateStatus.FIRING, 1,
80D,
+ previously, previously, previously, null);
+
+ var update = new AlertStateMachine().advance(firing,
+ new AlertRuleEvaluator().evaluate(rule, maximum), 1,
maximum.collectedAt());
+
+ assertThat(update.transition()).isEqualTo(AlertStateTransition.NONE);
+ assertThat(update.state().status()).isEqualTo(AlertStateStatus.FIRING);
+ }
+
+ @Test
+ void emptyProgressStillDoesNotProvideAQueueMaximumTest() {
+ List<MetricSample> samples =
collectCloudProgress(InstanceVendor.ALIYUN, List.of());
+
+ assertThat(samples).filteredOn(sample ->
sample.metricKey().equals("consumer.lag.max_queue"))
+ .singleElement().satisfies(sample -> {
+
assertThat(sample.availability()).isEqualTo(MetricAvailability.UNSUPPORTED);
+ assertThat(sample.value()).isNull();
+ });
+ assertThat(samples).filteredOn(sample ->
sample.metricKey().equals("consumer.lag.total"))
+ .singleElement().extracting(MetricSample::value).isEqualTo(0D);
+ }
+
+ @Test
+ void failedGroupProgressRemainsUnavailableTest() {
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ InstanceProvider provider = mock(InstanceProvider.class);
+ InstanceVO instance =
InstanceVO.builder().name("cloud").vendor(InstanceVendor.TENCENT).build();
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("orders");
+ when(registry.byInstanceId("cloud")).thenReturn(Optional.of(provider));
+ when(provider.listConsumerGroups("cloud",
null)).thenReturn(List.of(group));
+ when(provider.getGroupProgress("cloud", "orders")).thenThrow(new
IllegalStateException("unavailable"));
+
+ assertThat(new
CloudRocketMqBusinessMetricsCollector(registry).collect(instance))
+ .hasSize(3).allSatisfy(sample -> {
+
assertThat(sample.availability()).isEqualTo(MetricAvailability.UNAVAILABLE);
+ assertThat(sample.value()).isNull();
+ assertThat(sample.labels()).containsEntry("consumerGroup",
"orders");
+ });
+ }
+
+ private static List<MetricSample> collectCloudProgress(InstanceVendor
vendor, List<QueueProgressVO> progress) {
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ InstanceProvider provider = mock(InstanceProvider.class);
+ InstanceVO instance =
InstanceVO.builder().name("cloud").vendor(vendor).build();
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("orders");
+ group.setClusterId("cloud-a");
+ when(registry.byInstanceId("cloud")).thenReturn(Optional.of(provider));
+ when(provider.listConsumerGroups("cloud",
null)).thenReturn(List.of(group));
+ when(provider.getGroupProgress("cloud",
"orders")).thenReturn(progress);
+ return new
CloudRocketMqBusinessMetricsCollector(registry).collect(instance);
}
@Test
