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 eda9884d6 fix(metrics): preserve unavailable broker runtime fields
(#4521)
eda9884d6 is described below
commit eda9884d6be13432813545ea3b91c86537daa2c2
Author: zmuxuny <[email protected]>
AuthorDate: Mon Sep 21 17:29:26 2026 +0800
fix(metrics): preserve unavailable broker runtime fields (#4521)
ApacheRocketMqClusterMetricsCollector previously omitted
broker.disk.usage_ratio, broker.jvm.heap.usage_ratio and
broker.send_queue.usage_ratio when the Broker runtime table lacked those
fields, and alert reconciliation then treated the missing sample as resolved.
Each field is now emitted as an UNAVAILABLE sample (metricOrUnavailable) while
broker availability itself stays AVAILABLE.
Fixes #4519
---
.../ApacheRocketMqClusterMetricsCollector.java | 24 +++++++++-------
.../ApacheRocketMqClusterMetricsCollectorTest.java | 33 ++++++++++++++++++++++
.../studio/ops/alert/NativeAlertProcessorTest.java | 33 ++++++++++++++++++++++
3 files changed, 80 insertions(+), 10 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollector.java
index df1435be8..9196ca196 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollector.java
@@ -113,16 +113,14 @@ public class ApacheRocketMqClusterMetricsCollector
implements ClusterMetricsColl
return;
}
samples.add(available(BROKER_AVAILABILITY, instance, clusterId,
labels, 1D, collectedAt));
- parseDiskUsage(runtime.getTable().get("commitLogDiskRatio"))
- .ifPresent(value ->
samples.add(available(BROKER_DISK_USAGE_RATIO, instance, clusterId,
- labels, value, collectedAt)));
- parseHeapUsage(runtime.getTable().get("jvmMemoryHeapUsed"),
runtime.getTable().get("jvmMemoryHeapMax"))
- .ifPresent(value ->
samples.add(available(BROKER_JVM_HEAP_USAGE_RATIO, instance, clusterId,
- labels, value, collectedAt)));
- parseUsageRatio(runtime.getTable().get("sendThreadPoolQueueSize"),
- runtime.getTable().get("sendThreadPoolQueueCapacity"))
- .ifPresent(value ->
samples.add(available(BROKER_SEND_QUEUE_USAGE_RATIO, instance, clusterId,
- labels, value, collectedAt)));
+ samples.add(metricOrUnavailable(BROKER_DISK_USAGE_RATIO, instance,
clusterId, labels,
+
parseDiskUsage(runtime.getTable().get("commitLogDiskRatio")), collectedAt));
+ samples.add(metricOrUnavailable(BROKER_JVM_HEAP_USAGE_RATIO,
instance, clusterId, labels,
+ parseHeapUsage(runtime.getTable().get("jvmMemoryHeapUsed"),
+ runtime.getTable().get("jvmMemoryHeapMax")),
collectedAt));
+ samples.add(metricOrUnavailable(BROKER_SEND_QUEUE_USAGE_RATIO,
instance, clusterId, labels,
+
parseUsageRatio(runtime.getTable().get("sendThreadPoolQueueSize"),
+
runtime.getTable().get("sendThreadPoolQueueCapacity")), collectedAt));
} catch (Exception error) {
log.warn("Failed to collect runtime metrics for broker {} on
instance {}: {}", brokerName,
instance.getName(), error.getMessage());
@@ -163,6 +161,12 @@ public class ApacheRocketMqClusterMetricsCollector
implements ClusterMetricsColl
}
}
+ private static MetricSample metricOrUnavailable(String key, InstanceVO
instance, String clusterId,
+ Map<String, String> labels, java.util.Optional<Double> value,
Instant collectedAt) {
+ return value.map(metric -> available(key, instance, clusterId, labels,
metric, collectedAt))
+ .orElseGet(() -> unavailable(key, instance, clusterId, labels,
collectedAt));
+ }
+
private static MetricSample available(String key, InstanceVO instance,
String clusterId,
Map<String, String> labels, double value, Instant collectedAt) {
return new MetricSample(key, AlertDomain.CLUSTER, instance.getName(),
clusterId, labels, value,
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollectorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollectorTest.java
index 8aebb17b3..2ffaf2a4c 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollectorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollectorTest.java
@@ -75,6 +75,29 @@ class ApacheRocketMqClusterMetricsCollectorTest {
.singleElement().extracting(MetricSample::value).isEqualTo(0.25D);
}
+ @Test
+ void emitsBrokerScopedUnavailableSamplesForMissingRuntimeFieldsTest()
throws Exception {
+ RuntimeAdminClientResolver resolver =
mock(RuntimeAdminClientResolver.class);
+ MQAdminExt admin = mock(MQAdminExt.class);
+ InstanceVO instance = apacheInstance();
+ ClusterInfo topology = new ClusterInfo();
+ topology.setBrokerAddrTable(Map.of("broker-a", new
BrokerData("cluster-a", "broker-a",
+ new HashMap<>(Map.of(0L, "broker-a:10911")))));
+ KVTable runtime = new KVTable();
+ runtime.setTable(new HashMap<>());
+ when(admin.examineBrokerClusterInfo()).thenReturn(topology);
+
when(admin.fetchBrokerRuntimeStats("broker-a:10911")).thenReturn(runtime);
+ when(resolver.execute(eq(instance),
any(MqAdminExtFactory.AdminAction.class)))
+ .thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1)
+ .apply(admin));
+
+ List<MetricSample> samples = new
ApacheRocketMqClusterMetricsCollector(resolver).collect(instance);
+
+ assertUnavailableBrokerMetric(samples, "broker.disk.usage_ratio");
+ assertUnavailableBrokerMetric(samples, "broker.jvm.heap.usage_ratio");
+ assertUnavailableBrokerMetric(samples,
"broker.send_queue.usage_ratio");
+ }
+
@Test
void recordsUnavailableNameserverWhenTopologyCollectionFailsTest() {
RuntimeAdminClientResolver resolver =
mock(RuntimeAdminClientResolver.class);
@@ -100,6 +123,16 @@ class ApacheRocketMqClusterMetricsCollectorTest {
.isEmpty();
}
+ private static void assertUnavailableBrokerMetric(List<MetricSample>
samples, String metricKey) {
+ assertThat(samples).filteredOn(sample ->
sample.metricKey().equals(metricKey))
+ .singleElement().satisfies(sample -> {
+
assertThat(sample.availability()).isEqualTo(MetricAvailability.UNAVAILABLE);
+ assertThat(sample.value()).isNull();
+ assertThat(sample.labels()).containsEntry("brokerName",
"broker-a")
+ .containsEntry("brokerAddr", "broker-a:10911");
+ });
+ }
+
private static InstanceVO apacheInstance() {
return
InstanceVO.builder().name("local").endpoint("localhost:9876").vendor(InstanceVendor.APACHE).build();
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
index 6124bf166..869f2bbd6 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
@@ -517,6 +517,39 @@ class NativeAlertProcessorTest {
verify(alerts, never()).saveAlert(any(SystemAlertVO.class));
}
+ @Test
+ void brokerScopedUnavailableSampleKeepsActiveFingerprintTest() {
+ AlertService service = mock(AlertService.class);
+ AlertRuleVO rule =
AlertRuleVO.builder().id(1L).domain(AlertDomain.CLUSTER).name("Broker disk")
+
.metric("broker.disk.usage_ratio").operator(">").threshold(0.8).enabled(true)
+
.instanceId("local").brokerName("broker-a").consecutiveSamples(1).build();
+ when(service.listRules(AlertDomain.CLUSTER)).thenReturn(List.of(rule));
+ Map<String, String> labels = Map.of("brokerName", "broker-a",
"brokerAddr", "broker-a:10911");
+ MetricSample unavailable = new MetricSample("broker.disk.usage_ratio",
AlertDomain.CLUSTER, "local",
+ "cluster-a", labels, null, MetricAvailability.UNAVAILABLE,
Instant.now());
+ AlertStateKey key = new AlertStateKey(rule.getId(),
AlertFingerprint.of(rule.getId(), "local", labels));
+ AlertRuleState firing = new AlertRuleState(AlertStateStatus.FIRING, 1,
0.9D,
+ unavailable.collectedAt().minusSeconds(60),
unavailable.collectedAt().minusSeconds(60),
+ unavailable.collectedAt().minusSeconds(60), null);
+ ActiveAlertState active = new ActiveAlertState(key, firing, "local",
labels);
+ AlertStateRepository states = mock(AlertStateRepository.class);
+ when(states.find(key)).thenReturn(Optional.of(firing));
+ when(states.save(eq(key), any(AlertRuleState.class))).thenReturn(true);
+ when(states.findActive(any(MetricCollectionScope.class),
eq(List.of(rule)))).thenReturn(List.of(active));
+ AlertRepository alerts = mock(AlertRepository.class);
+ NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
+
+ NativeAlertProcessor processor = new NativeAlertProcessor(service,
+ new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
+ mock(MetricSnapshotRepository.class), alerts, outbox,
suppression()),
+ new AlertStateMachine(), states, alerts, outbox,
suppression(), mockTxManager());
+ processor.processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.CLUSTER, "local",
+ java.util.Set.of("broker.disk.usage_ratio")),
List.of(unavailable));
+
+ verify(alerts, never()).saveAlert(any(SystemAlertVO.class));
+ verify(outbox, never()).enqueue(any(), any(), any());
+ }
+
@Test
void resolvesMissingMetricEvenWhenAnotherMetricSharesTheSameLabelsTest() {
AlertService service = mock(AlertService.class);
