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 e4960b708 fix(metrics): preserve cloud group failure diagnostic scope
(#6199)
e4960b708 is described below
commit e4960b708bf81af20a384e8c731651b1fc62beeb
Author: zmuxuny <[email protected]>
AuthorDate: Mon Oct 12 00:19:12 2026 -0700
fix(metrics): preserve cloud group failure diagnostic scope (#6199)
The group-level failure branch in
`CloudRocketMqBusinessMetricsCollector.java:107-109` built its three
UNAVAILABLE samples through the four-argument `unavailable(...)` helper at
`:119-123`, which hardcodes a null cluster id even though the group's cluster
is known at that point. `NativeAlertRuleScopeMatcher.java:38` selects on
`matchesSelector(rule.getClusterName(), sample.clusterId())` and
`NativeAlertRuleTestService.java:77-79` filters samples through that matcher
before returning t [...]
Seven new instances drive the real `TencentInstanceProvider` through the
collector, the rule-test service and `AlertRuleEvaluator`, asserting first that
the provider maps the injected `ClusterIdV4` onto `group.getClusterId()`.
Passing null instead of the cluster id empties
`clusterScopedRuleTestReportsTheKnownGroupOutageTest` for all three metrics.
`wholeInstanceFailureRemainsUnscopedTest`,
`unscopedRuleStillReportsFailureWithoutAProviderClusterIdTest`,
`failureDoesNotAppearInAnotherC [...]
---
.../CloudRocketMqBusinessMetricsCollector.java | 13 +-
.../tencent/TencentCloudLagFailureScopeTest.java | 151 +++++++++++++++++++++
2 files changed, 160 insertions(+), 4 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 8503c84a3..ae98df3a6 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
@@ -105,9 +105,9 @@ public class CloudRocketMqBusinessMetricsCollector
implements BusinessMetricsCol
} catch (RuntimeException error) {
log.warn("Failed to collect cloud consumer lag for group {} on
instance {}: {}", group.getName(),
instance.getName(), error.getMessage());
- return List.of(unavailable(CONSUMER_LAG_TOTAL, instance, labels,
collectedAt),
- unavailable(CONSUMER_LAG_MAX_QUEUE, instance, labels,
collectedAt),
- unavailable(TOPIC_BACKLOG_TOTAL, instance, labels,
collectedAt));
+ return List.of(unavailable(CONSUMER_LAG_TOTAL, instance,
group.getClusterId(), labels, collectedAt),
+ unavailable(CONSUMER_LAG_MAX_QUEUE, instance,
group.getClusterId(), labels, collectedAt),
+ unavailable(TOPIC_BACKLOG_TOTAL, instance,
group.getClusterId(), labels, collectedAt));
}
}
@@ -119,7 +119,12 @@ public class CloudRocketMqBusinessMetricsCollector
implements BusinessMetricsCol
private static MetricSample unavailable(String metric, InstanceVO
instance, Map<String, String> labels,
Instant collectedAt) {
- return new MetricSample(metric, AlertDomain.BUSINESS,
instance.getName(), null, labels, null,
+ return unavailable(metric, instance, null, labels, collectedAt);
+ }
+
+ private static MetricSample unavailable(String metric, InstanceVO
instance, String clusterId,
+ Map<String, String> labels, Instant collectedAt) {
+ return new MetricSample(metric, AlertDomain.BUSINESS,
instance.getName(), clusterId, labels, null,
MetricAvailability.UNAVAILABLE, collectedAt);
}
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentCloudLagFailureScopeTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentCloudLagFailureScopeTest.java
new file mode 100644
index 000000000..454a3121e
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentCloudLagFailureScopeTest.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.provider.tencent;
+
+import com.tencentcloudapi.trocket.v20230308.TrocketClient;
+import com.tencentcloudapi.trocket.v20230308.models.ConsumeGroupItem;
+import
com.tencentcloudapi.trocket.v20230308.models.DescribeConsumerGroupListResponse;
+import
com.tencentcloudapi.trocket.v20230308.models.DescribeTopicListByGroupResponse;
+import com.tencentcloudapi.trocket.v20230308.models.SubscriptionData;
+import org.apache.rocketmq.studio.cluster.metrics.MetricAvailability;
+import
org.apache.rocketmq.studio.cluster.metrics.collectors.CloudRocketMqBusinessMetricsCollector;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.InstanceRepository;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.ops.alert.AlertDomain;
+import org.apache.rocketmq.studio.ops.alert.AlertRuleEvaluator;
+import org.apache.rocketmq.studio.ops.alert.AlertRuleTestResultVO;
+import org.apache.rocketmq.studio.ops.alert.AlertRuleVO;
+import org.apache.rocketmq.studio.ops.alert.NativeAlertRuleTestService;
+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.ValueSource;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class TencentCloudLagFailureScopeTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"consumer.lag.total", "consumer.lag.max_queue",
"topic.backlog.total"})
+ void clusterScopedRuleTestReportsTheKnownGroupOutageTest(String metric)
throws Exception {
+ NativeAlertRuleTestService service =
serviceWithFailedProgress("legacy-cluster");
+
+ AlertRuleTestResultVO result = service.test(rule(metric,
"legacy-cluster"));
+
+ assertThat(result.samples()).singleElement().satisfies(sample -> {
+ assertThat(sample.availability()).isEqualTo("UNAVAILABLE");
+ assertThat(sample.currentValue()).isNull();
+ assertThat(sample.conditionMet()).isFalse();
+ assertThat(sample.labels()).containsEntry("consumerGroup",
"orders");
+ });
+ }
+
+ @Test
+ void failureDoesNotAppearInAnotherClustersRuleTest() throws Exception {
+ assertThat(serviceWithFailedProgress("legacy-cluster")
+ .test(rule("consumer.lag.total",
"other-cluster")).samples()).isEmpty();
+ }
+
+ @Test
+ void unscopedRuleStillReportsFailureWithoutAProviderClusterIdTest() throws
Exception {
+
assertThat(serviceWithFailedProgress(null).test(rule("consumer.lag.total",
null)).samples())
+ .singleElement().satisfies(sample ->
assertThat(sample.availability()).isEqualTo("UNAVAILABLE"));
+ }
+
+ @Test
+ void successfulClusterScopedRuleStillReportsItsMeasuredLagTest() throws
Exception {
+ var result = serviceWithProgress("legacy-cluster", false)
+ .test(rule("consumer.lag.total", "legacy-cluster"));
+
+ assertThat(result.samples()).singleElement().satisfies(sample -> {
+ assertThat(sample.availability()).isEqualTo("AVAILABLE");
+ assertThat(sample.currentValue()).isEqualTo(80D);
+ assertThat(sample.conditionMet()).isTrue();
+ });
+ }
+
+ @Test
+ void wholeInstanceFailureRemainsUnscopedTest() {
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ when(registry.byInstanceId("tencent-prod"))
+ .thenThrow(new BusinessException(502, "provider unavailable"));
+ InstanceVO instance =
InstanceVO.builder().name("tencent-prod").vendor(InstanceVendor.TENCENT).build();
+
+ assertThat(new
CloudRocketMqBusinessMetricsCollector(registry).collect(instance))
+ .hasSize(3).allSatisfy(sample -> {
+ assertThat(sample.clusterId()).isNull();
+ assertThat(sample.labels()).isEmpty();
+ assertThat(sample.value()).isNull();
+
assertThat(sample.availability()).isEqualTo(MetricAvailability.UNAVAILABLE);
+ });
+ }
+
+ private static AlertRuleVO rule(String metric, String cluster) {
+ return
AlertRuleVO.builder().domain(AlertDomain.BUSINESS).instanceId("tencent-prod")
+ .consumerGroup("orders").clusterName(cluster).metric(metric)
+ .operator(">").threshold(50).consecutiveSamples(1).build();
+ }
+
+ private static NativeAlertRuleTestService serviceWithFailedProgress(String
cluster) throws Exception {
+ return serviceWithProgress(cluster, true);
+ }
+
+ private static NativeAlertRuleTestService serviceWithProgress(String
cluster, boolean failure) throws Exception {
+ InstanceRepository instances = mock(InstanceRepository.class);
+ InstanceVO instance =
InstanceVO.builder().name("tencent-prod").vendor(InstanceVendor.TENCENT)
+
.cloudInstanceId("rmq-test").regionId("ap-chengdu").credentialId(1L).build();
+
when(instances.findByIdentifier("tencent-prod")).thenReturn(Optional.of(instance));
+ TrocketClient client = mock(TrocketClient.class);
+ TencentClientFactory factory = mock(TencentClientFactory.class);
+ when(factory.call(any(Long.class), anyString(),
any())).thenAnswer(invocation ->
+
invocation.<TencentClientFactory.TencentCall<Object>>getArgument(2).execute(client));
+ ConsumeGroupItem item = new ConsumeGroupItem();
+ item.setConsumerGroup("orders");
+ // Real Tencent mapping retains the source cluster of a migrated
consumer group.
+ item.setClusterIdV4(cluster);
+ DescribeConsumerGroupListResponse groups = new
DescribeConsumerGroupListResponse();
+ groups.setData(new ConsumeGroupItem[] {item});
+ groups.setTotalCount(1L);
+ when(client.DescribeConsumerGroupList(any())).thenReturn(groups);
+ if (failure) {
+ when(client.DescribeTopicListByGroup(any())).thenThrow(new
BusinessException(502, "progress unavailable"));
+ } else {
+ SubscriptionData subscription = new SubscriptionData();
+ subscription.setTopic("orders-topic");
+ subscription.setConsumerLag(80L);
+ DescribeTopicListByGroupResponse progress = new
DescribeTopicListByGroupResponse();
+ progress.setData(new SubscriptionData[] {subscription});
+ progress.setTotalCount(1L);
+ when(client.DescribeTopicListByGroup(any())).thenReturn(progress);
+ }
+ TencentInstanceProvider provider = new
TencentInstanceProvider(factory, instances);
+ assertThat(provider.listConsumerGroups("tencent-prod",
null)).singleElement()
+ .satisfies(group ->
assertThat(group.getClusterId()).isEqualTo(cluster));
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+
when(registry.byInstanceId("tencent-prod")).thenReturn(Optional.of(provider));
+ return new NativeAlertRuleTestService(instances, List.of(),
+ List.of(new CloudRocketMqBusinessMetricsCollector(registry)),
new AlertRuleEvaluator());
+ }
+}
