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 4a52e5567 fix(alert): keep cluster scope in native alert correlation
(#6197)
4a52e5567 is described below
commit 4a52e556728e15c43efe71a67286075a33e324c3
Author: zmuxuny <[email protected]>
AuthorDate: Mon Oct 12 00:18:44 2026 -0700
fix(alert): keep cluster scope in native alert correlation (#6197)
`AlertCorrelationScope.matches` (`:39-46`) compares `brokerName`,
`brokerAddr`, `clusterName` and `clusterId` only when both events carry the
key, and neither domain put a cluster key into its labels: BUSINESS samples
label `consumerGroup` (plus `topic`), CLUSTER samples label `brokerName` (plus
`brokerAddr`). On an Apache instance whose NameServer serves several clusters,
a broker-down alert in any cluster therefore suppressed consumer-lag
notifications in every other cluster of the [...]
Nine instances across three tests drive the real Apache cluster and
business collectors through the real processor, evaluation service and
suppression service. Removing the label enrichment breaks the `cluster-a` rows
of `correlatesRealCollectorEventsWithinTheirKnownClusterTest` and its
cross-cluster `false` row; removing the `enqueue` argument change breaks that
test's outbox verification; restoring the filter inside the loop breaks
`newerResolvedClusterIdentitySupersedesLegacyUnscop [...]
---
.../alert/AlertNotificationSuppressionService.java | 6 +-
.../ops/alert/NativeAlertEvaluationService.java | 10 +-
.../AlertNotificationSuppressionServiceTest.java | 18 +++
.../alert/NativeAlertClusterCorrelationTest.java | 180 +++++++++++++++++++++
4 files changed, 209 insertions(+), 5 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationSuppressionService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationSuppressionService.java
index d8a1da304..c4d765889 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationSuppressionService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationSuppressionService.java
@@ -52,9 +52,6 @@ public class AlertNotificationSuppressionService {
windowStart, event.getTime(), page, CANDIDATE_PAGE_SIZE));
List<SystemAlertVO> candidates = result.getItems();
for (SystemAlertVO candidate : candidates) {
- if (!AlertCorrelationScope.matches(event, candidate)) {
- continue;
- }
String incident = candidate.getFingerprint() == null
? legacyIncidentKey(candidate)
: candidate.getFingerprint();
@@ -67,6 +64,9 @@ public class AlertNotificationSuppressionService {
page++;
}
return latestByIncident.values().stream()
+ // A newer scoped event supersedes an older unscoped event for
the same incident,
+ // including events written before cluster identity was
carried into alert labels.
+ .filter(candidate -> AlertCorrelationScope.matches(event,
candidate))
// REMINDER is emitted only while the state stays FIRING, so
an incident whose
// latest in-window event is a REMINDER is still active; only
RESOLVED ends it.
.filter(candidate ->
"FIRING".equalsIgnoreCase(candidate.getTransition())
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationService.java
index 0de1fac79..b927b9fa0 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationService.java
@@ -24,6 +24,7 @@ import
org.apache.rocketmq.studio.common.domain.enums.AlertLevel;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.StringUtils;
import java.time.Duration;
import java.time.LocalDateTime;
@@ -68,6 +69,11 @@ public class NativeAlertEvaluationService {
}
LocalDateTime eventTime =
LocalDateTime.ofInstant(sample.collectedAt(), ZoneOffset.UTC);
+ Map<String, String> eventLabels = new TreeMap<>(sample.labels());
+ // Preserve incident fingerprints while carrying the authoritative
cluster into correlation.
+ if (StringUtils.hasText(sample.clusterId())) {
+ eventLabels.put("clusterId", sample.clusterId().trim());
+ }
SystemAlertVO event = SystemAlertVO.builder()
.level(level(rule.getSeverity()))
.title(rule.getName())
@@ -80,7 +86,7 @@ public class NativeAlertEvaluationService {
.transition(update.transition().name())
.instanceId(sample.instanceId())
.currentValue(update.state().currentValue())
- .labels(Map.copyOf(new TreeMap<>(sample.labels())))
+ .labels(Map.copyOf(eventLabels))
.build();
applyNotificationSuppression(event, sample.domain(),
update.transition());
@@ -92,7 +98,7 @@ public class NativeAlertEvaluationService {
alertRepository.markRuleTriggered(rule.getId(),
eventTime.toString());
}
if (!event.isNotificationSuppressed()) {
- notificationOutboxService.enqueue(event, rule, sample.labels());
+ notificationOutboxService.enqueue(event, rule, event.getLabels());
}
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationSuppressionServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationSuppressionServiceTest.java
index faf5a8e5c..10f9c12e2 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationSuppressionServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationSuppressionServiceTest.java
@@ -95,6 +95,24 @@ class AlertNotificationSuppressionServiceTest {
.isEmpty();
}
+ @Test
+ void newerResolvedClusterIdentitySupersedesLegacyUnscopedFiringTest() {
+ AlertRepository repository = mock(AlertRepository.class);
+ LocalDateTime now = LocalDateTime.now();
+ SystemAlertVO firing = event(1L, AlertDomain.CLUSTER, "FIRING",
"broker-a", now.minusMinutes(3));
+ firing.setFingerprint("stable-broker-incident");
+ SystemAlertVO resolved = event(2L, AlertDomain.CLUSTER, "RESOLVED",
"broker-a", now.minusMinutes(1));
+ resolved.setFingerprint(firing.getFingerprint());
+ resolved.setLabels(Map.of("brokerName", "broker-a", "clusterId",
"cluster-a"));
+ SystemAlertVO business =
SystemAlertVO.builder().id(3L).domain(AlertDomain.BUSINESS)
+ .transition("FIRING").instanceId("local").time(now)
+ .labels(Map.of("consumerGroup", "orders", "clusterId",
"cluster-b")).build();
+
when(repository.findAlertsPage(any())).thenReturn(PageResult.of(List.of(resolved,
firing), 2, 1, 100));
+
+ assertThat(new
AlertNotificationSuppressionService(repository).findSuppressingClusterAlert(business))
+ .isEmpty();
+ }
+
@Test
void doesNotSuppressAcrossDifferentBrokerScopesTest() {
AlertRepository repository = mock(AlertRepository.class);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertClusterCorrelationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertClusterCorrelationTest.java
new file mode 100644
index 000000000..5d590afe5
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertClusterCorrelationTest.java
@@ -0,0 +1,180 @@
+/*
+ * 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.ops.alert;
+
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.cluster.metrics.MetricSample;
+import org.apache.rocketmq.studio.cluster.metrics.MetricAvailability;
+import org.apache.rocketmq.studio.cluster.metrics.MetricSnapshotRepository;
+import
org.apache.rocketmq.studio.cluster.metrics.collectors.ApacheRocketMqBusinessMetricsCollector;
+import
org.apache.rocketmq.studio.cluster.metrics.collectors.ApacheRocketMqClusterMetricsCollector;
+import org.apache.rocketmq.studio.common.domain.PageResult;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.provider.InstanceProvider;
+import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.springframework.transaction.PlatformTransactionManager;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class NativeAlertClusterCorrelationTest {
+ @ParameterizedTest
+ @CsvSource(value = {"cluster-a, cluster-a", "NULL, legacy", "' ',
legacy"}, nullValues = "NULL")
+ void enrichesEventLabelsWithoutChangingTheExistingFingerprintTest(String
clusterId, String eventCluster) {
+ Map<String, String> identityLabels = Map.of("consumerGroup", "orders",
"clusterId", "legacy",
+ "clusterName", "friendly-name");
+ MetricSample sample = new MetricSample("consumer.lag.total",
AlertDomain.BUSINESS, "shared-nameserver",
+ clusterId, identityLabels, 100D, MetricAvailability.AVAILABLE,
Instant.now());
+ AlertRuleVO rule = AlertRuleVO.builder().id(2L).name("Orders
backlog").domain(AlertDomain.BUSINESS)
+
.metric("consumer.lag.total").operator(">").threshold(10).instanceId(sample.instanceId())
+
.consumerGroup("orders").enabled(true).consecutiveSamples(1).build();
+ AlertStateRepository states = mock(AlertStateRepository.class);
+ when(states.find(any())).thenReturn(Optional.empty());
+ when(states.save(any(), any())).thenReturn(true);
+ AlertRepository alerts = mock(AlertRepository.class);
+ when(alerts.saveAlert(any())).thenAnswer(invocation ->
invocation.getArgument(0));
+ NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
+ AlertNotificationSuppressionService suppression =
mock(AlertNotificationSuppressionService.class);
+
when(suppression.findSuppressingClusterAlert(any())).thenReturn(Optional.empty());
+
+ new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
+ mock(MetricSnapshotRepository.class), alerts, outbox,
suppression).evaluate(rule, sample);
+
+ org.mockito.ArgumentCaptor<SystemAlertVO> event =
org.mockito.ArgumentCaptor.forClass(SystemAlertVO.class);
+ verify(alerts).saveAlert(event.capture());
+ assertThat(event.getValue().getLabels()).containsEntry("clusterId",
eventCluster)
+ .containsEntry("clusterName", "friendly-name");
+ String originalFingerprint = AlertFingerprint.of(rule.getId(),
sample.instanceId(), identityLabels);
+
assertThat(event.getValue().getFingerprint()).isEqualTo(originalFingerprint);
+ verify(states).save(eq(new AlertStateKey(rule.getId(),
originalFingerprint)), any());
+ verify(outbox).enqueue(eq(event.getValue()), eq(rule),
eq(event.getValue().getLabels()));
+ }
+
+ @ParameterizedTest
+ @CsvSource(value = {
+ "cluster-a, cluster-b, false",
+ "cluster-a, cluster-a, true",
+ "cluster-a, NULL, true",
+ "NULL, cluster-b, true",
+ "NULL, NULL, true"
+ }, nullValues = "NULL")
+ void correlatesRealCollectorEventsWithinTheirKnownClusterTest(String
brokerCluster, String consumerCluster,
+ boolean expectedSuppression) throws Exception {
+ InstanceVO instance =
InstanceVO.builder().name("shared-nameserver").endpoint("localhost:9876")
+ .vendor(InstanceVendor.APACHE).build();
+ RuntimeAdminClientResolver resolver =
mock(RuntimeAdminClientResolver.class);
+ MQAdminExt admin = mock(MQAdminExt.class);
+ ClusterInfo topology = new ClusterInfo();
+ topology.setBrokerAddrTable(Map.of(
+ "broker-a", new BrokerData(brokerCluster, "broker-a", new
HashMap<>(Map.of(0L, "broker-a:10911"))),
+ "broker-b", new BrokerData("cluster-b", "broker-b", new
HashMap<>(Map.of(0L, "broker-b:10911")))));
+ when(admin.examineBrokerClusterInfo()).thenReturn(topology);
+ when(admin.fetchBrokerRuntimeStats("broker-a:10911")).thenThrow(new
IllegalStateException("offline"));
+ KVTable healthy = new KVTable();
+ healthy.setTable(new HashMap<>());
+
when(admin.fetchBrokerRuntimeStats("broker-b:10911")).thenReturn(healthy);
+ when(resolver.execute(eq(instance),
any(MqAdminExtFactory.AdminAction.class))).thenAnswer(invocation ->
+
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(admin));
+ List<MetricSample> clusterSamples = new
ApacheRocketMqClusterMetricsCollector(resolver).collect(instance);
+ assertThat(clusterSamples).filteredOn(sample ->
"broker.availability".equals(sample.metricKey())
+ &&
"broker-a".equals(sample.labels().get("brokerName")))
+ .singleElement().satisfies(sample ->
assertThat(sample.clusterId()).isEqualTo(brokerCluster));
+
+ InstanceProvider provider = mock(InstanceProvider.class);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("orders");
+ group.setClusterId(consumerCluster);
+ group.setConsumeStatsAvailable(true);
+ group.setTotalLag(100);
+
when(registry.byInstanceId(instance.getName())).thenReturn(Optional.of(provider));
+ when(provider.listConsumerGroups(instance.getName(),
null)).thenReturn(List.of(group));
+ when(provider.getGroupProgress(instance.getName(),
"orders")).thenReturn(List.of());
+ List<MetricSample> businessSamples = new
ApacheRocketMqBusinessMetricsCollector(registry).collect(instance);
+ assertThat(businessSamples).filteredOn(sample ->
"consumer.lag.total".equals(sample.metricKey()))
+ .singleElement().satisfies(sample ->
assertThat(sample.clusterId()).isEqualTo(consumerCluster));
+
+ AlertRuleVO clusterRule = AlertRuleVO.builder().id(1L).name("Broker
offline").domain(AlertDomain.CLUSTER)
+
.metric("broker.availability").operator("UNAVAILABLE").instanceId(instance.getName())
+
.brokerName("broker-a").enabled(true).consecutiveSamples(1).build();
+ AlertRuleVO businessRule = AlertRuleVO.builder().id(2L).name("Orders
backlog").domain(AlertDomain.BUSINESS)
+
.metric("consumer.lag.total").operator(">").threshold(10).instanceId(instance.getName())
+
.consumerGroup("orders").enabled(true).consecutiveSamples(1).build();
+ AlertService service = mock(AlertService.class);
+
when(service.listRules(AlertDomain.CLUSTER)).thenReturn(List.of(clusterRule));
+
when(service.listRules(AlertDomain.BUSINESS)).thenReturn(List.of(businessRule));
+ AlertStateRepository states = mock(AlertStateRepository.class);
+ when(states.find(any())).thenReturn(Optional.empty());
+ when(states.save(any(), any())).thenReturn(true);
+ List<SystemAlertVO> emitted = new ArrayList<>();
+ AlertRepository alerts = mock(AlertRepository.class);
+ when(alerts.saveAlert(any())).thenAnswer(invocation -> {
+ SystemAlertVO event = invocation.getArgument(0);
+ event.setId((long) emitted.size() + 1);
+ emitted.add(event);
+ return event;
+ });
+ when(alerts.findAlertsPage(any())).thenAnswer(invocation -> {
+ List<SystemAlertVO> candidates = emitted.stream().filter(event ->
event.getDomain() == AlertDomain.CLUSTER)
+ .toList();
+ return PageResult.of(candidates, candidates.size(), 1, 100);
+ });
+ NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
+ AlertNotificationSuppressionService suppression = new
AlertNotificationSuppressionService(alerts);
+ AlertStateMachine stateMachine = new AlertStateMachine();
+ NativeAlertProcessor processor = new NativeAlertProcessor(service,
+ new NativeAlertEvaluationService(new AlertRuleEvaluator(),
stateMachine, states,
+ mock(MetricSnapshotRepository.class), alerts, outbox,
suppression),
+ stateMachine, states, alerts, outbox, suppression,
mock(PlatformTransactionManager.class));
+
+ processor.process(clusterSamples);
+ processor.process(businessSamples);
+
+ assertThat(emitted).hasSize(2);
+ SystemAlertVO businessEvent = emitted.get(1);
+ assertThat(businessEvent.getDomain()).isEqualTo(AlertDomain.BUSINESS);
+
assertThat(businessEvent.isNotificationSuppressed()).isEqualTo(expectedSuppression);
+ if (expectedSuppression) {
+
assertThat(businessEvent.getSuppressionCauseAlertId()).isEqualTo(emitted.get(0).getId());
+ verify(outbox, never()).enqueue(eq(businessEvent),
eq(businessRule), any());
+ } else {
+ assertThat(businessEvent.getSuppressionCauseAlertId()).isNull();
+ verify(outbox).enqueue(eq(businessEvent), eq(businessRule), any());
+ }
+ }
+}
