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 cd353fc7e [ISSUE #2951][ISSUE #2950][ISSUE #2952] fix(alert):
consolidate rule normalization and incident correlation (#2960)
cd353fc7e is described below
commit cd353fc7e8cdd126e1038edce6e661d06f70bd05
Author: shown <[email protected]>
AuthorDate: Fri Sep 4 14:22:22 2026 +0800
[ISSUE #2951][ISSUE #2950][ISSUE #2952] fix(alert): consolidate rule
normalization and incident correlation (#2960)
* [ISSUE #2951] Normalize native alert metric keys
* [ISSUE #2950] Report topic route skew once
* [ISSUE #2952] Correlate resolved legacy alert incidents
---
.../alert/AlertNotificationSuppressionService.java | 16 +++++++-
.../ops/alert/NativeAlertMetricCatalogService.java | 11 +++--
.../AlertNotificationSuppressionServiceTest.java | 17 ++++++++
.../alert/NativeAlertMetricCatalogServiceTest.java | 18 ++++++++
web/src/utils/topicRouteDiagnostics.test.ts | 25 +++++++++++
web/src/utils/topicRouteDiagnostics.ts | 48 ++++++++++------------
6 files changed, 104 insertions(+), 31 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 727921623..9609af9ec 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
@@ -24,6 +24,7 @@ import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Comparator;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -54,7 +55,8 @@ public class AlertNotificationSuppressionService {
if (!AlertCorrelationScope.matches(event, candidate)) {
continue;
}
- String incident = candidate.getFingerprint() == null ?
String.valueOf(candidate.getId())
+ String incident = candidate.getFingerprint() == null
+ ? legacyIncidentKey(candidate)
: candidate.getFingerprint();
latestByIncident.merge(incident, candidate, (left, right) ->
later(left, right) ? left : right);
}
@@ -69,6 +71,18 @@ public class AlertNotificationSuppressionService {
.max(Comparator.comparing(SystemAlertVO::getTime,
Comparator.nullsLast(Comparator.naturalOrder())));
}
+ private static String legacyIncidentKey(SystemAlertVO alert) {
+ Map<String, String> identityLabels = new LinkedHashMap<>();
+ if (alert.getLabels() != null) {
+ identityLabels.putAll(alert.getLabels());
+ }
+ long ruleId = alert.getRuleId() == null ? 0L : alert.getRuleId();
+ if (alert.getRuleId() == null && alert.getTitle() != null) {
+ identityLabels.put("__legacy_title", alert.getTitle());
+ }
+ return "legacy:" + AlertFingerprint.of(ruleId, alert.getInstanceId(),
identityLabels);
+ }
+
private static boolean later(SystemAlertVO left, SystemAlertVO right) {
if (left.getTime() == null) {
return false;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertMetricCatalogService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertMetricCatalogService.java
index ebb1cce24..3eed1b8cc 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertMetricCatalogService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertMetricCatalogService.java
@@ -68,13 +68,18 @@ public class NativeAlertMetricCatalogService {
}
public void validate(AlertRuleVO rule) {
- if (rule == null || rule.getMetric() == null ||
!NATIVE_METRICS.contains(rule.getMetric())) {
+ if (rule == null || rule.getMetric() == null) {
+ return;
+ }
+ String metric = rule.getMetric().trim();
+ rule.setMetric(metric);
+ if (!NATIVE_METRICS.contains(metric)) {
return;
}
boolean supported = list(rule.getInstanceId(),
rule.getDomain()).stream()
- .anyMatch(metric -> metric.key().equals(rule.getMetric()));
+ .anyMatch(candidate -> candidate.key().equals(metric));
if (!supported) {
- throw new BusinessException(400, "Native metric " +
rule.getMetric()
+ throw new BusinessException(400, "Native metric " + metric
+ " is not supported by the selected Studio instance");
}
}
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 99e250f30..e945a51a7 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
@@ -78,6 +78,23 @@ class AlertNotificationSuppressionServiceTest {
.isEmpty();
}
+ @Test
+ void
doesNotSuppressAfterALegacyIncidentWithoutFingerprintsHasResolvedTest() {
+ AlertRepository repository = mock(AlertRepository.class);
+ LocalDateTime now = LocalDateTime.now();
+ SystemAlertVO firing = event(1L, AlertDomain.CLUSTER, "FIRING",
"broker-1", now.minusMinutes(3));
+ firing.setTitle("Broker unavailable");
+ firing.setRuleId(7L);
+ SystemAlertVO resolved = event(2L, AlertDomain.CLUSTER, "RESOLVED",
"broker-1", now.minusMinutes(1));
+ resolved.setTitle("Broker unavailable");
+ resolved.setRuleId(7L);
+
when(repository.findAlertsPage(any())).thenReturn(PageResult.of(List.of(firing,
resolved), 2, 1, 100));
+
+ assertThat(new AlertNotificationSuppressionService(repository)
+ .findSuppressingClusterAlert(event(3L, AlertDomain.BUSINESS,
"FIRING", "broker-1", now)))
+ .isEmpty();
+ }
+
@Test
void doesNotSuppressAcrossDifferentBrokerScopesTest() {
AlertRepository repository = mock(AlertRepository.class);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertMetricCatalogServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertMetricCatalogServiceTest.java
index b04f4caac..b94bbd334 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertMetricCatalogServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertMetricCatalogServiceTest.java
@@ -41,4 +41,22 @@ class NativeAlertMetricCatalogServiceTest {
.instanceId("aliyun").metric("broker.availability").build()))
.hasMessageContaining("not supported");
}
+
+ @Test
+ void normalizesMetricKeysBeforeNativeAndCustomValidationTest() {
+ InstanceRepository repository = mock(InstanceRepository.class);
+
when(repository.findByIdentifier("apache")).thenReturn(Optional.of(InstanceVO.builder()
+ .name("apache").vendor(InstanceVendor.APACHE).build()));
+ NativeAlertMetricCatalogService service = new
NativeAlertMetricCatalogService(repository);
+ AlertRuleVO nativeRule =
AlertRuleVO.builder().domain(AlertDomain.CLUSTER)
+ .instanceId("apache").metric(" broker.disk.usage_ratio
").build();
+ AlertRuleVO customRule =
AlertRuleVO.builder().domain(AlertDomain.CLUSTER)
+ .instanceId("apache").metric(" custom.metric ").build();
+
+ service.validate(nativeRule);
+ service.validate(customRule);
+
+
assertThat(nativeRule.getMetric()).isEqualTo("broker.disk.usage_ratio");
+ assertThat(customRule.getMetric()).isEqualTo("custom.metric");
+ }
}
diff --git a/web/src/utils/topicRouteDiagnostics.test.ts
b/web/src/utils/topicRouteDiagnostics.test.ts
index fd05fdec4..320c970d3 100644
--- a/web/src/utils/topicRouteDiagnostics.test.ts
+++ b/web/src/utils/topicRouteDiagnostics.test.ts
@@ -147,6 +147,31 @@ describe('topic route diagnostics', () => {
);
});
+ it('reports topic-level queue skew once without blaming each broker', () => {
+ const diagnostics = analyzeTopicRoutes([
+ route({ brokerName: 'broker-a', writeQueues: 12, readQueues: 12 }),
+ route({
+ brokerName: 'broker-b',
+ brokerAddr: '10.0.1.1:10911',
+ masterAddr: '10.0.1.1:10911',
+ brokerAddrs: {
+ '0': '10.0.1.1:10911',
+ '1': '10.0.1.2:10911',
+ },
+ writeQueues: 2,
+ readQueues: 2,
+ }),
+ ]);
+
+ expect(diagnostics.status).toBe('warning');
+ expect(diagnostics.issues.filter((item) => item.code ===
'WRITE_QUEUE_SKEW')).toHaveLength(1);
+ expect(diagnostics.issues.filter((item) => item.code ===
'READ_QUEUE_SKEW')).toHaveLength(1);
+ expect(diagnostics.distributions).toEqual([
+ expect.objectContaining({ brokerName: 'broker-a', status: 'healthy',
issues: [] }),
+ expect.objectContaining({ brokerName: 'broker-b', status: 'healthy',
issues: [] }),
+ ]);
+ });
+
it('infers read and write permissions from legacy route payloads', () => {
const diagnostics = analyzeTopicRoutes([
route({
diff --git a/web/src/utils/topicRouteDiagnostics.ts
b/web/src/utils/topicRouteDiagnostics.ts
index d273dd15e..0a88208be 100644
--- a/web/src/utils/topicRouteDiagnostics.ts
+++ b/web/src/utils/topicRouteDiagnostics.ts
@@ -208,8 +208,6 @@ const collectAddressDuplicates = (routes: BrokerRoute[]):
Set<string> => {
const distributionIssues = (
route: BrokerRoute,
- writeSkew: RouteQueueSkew,
- readSkew: RouteQueueSkew,
duplicateAddresses: Set<string>,
): RouteDiagnosticIssue[] => {
const brokerName = route.brokerName || 'unknown';
@@ -305,30 +303,6 @@ const distributionIssues = (
);
}
- if (hasSkew(writeSkew)) {
- issues.push(
- issue(
- 'WRITE_QUEUE_SKEW',
- 'warning',
- '写队列分布不均',
- '不同 Broker 的写队列数差距较大，生产流量可能无法均匀分摊。',
- brokerName,
- ),
- );
- }
-
- if (hasSkew(readSkew)) {
- issues.push(
- issue(
- 'READ_QUEUE_SKEW',
- 'warning',
- '读队列分布不均',
- '不同 Broker 的读队列数差距较大，消费者负载可能无法均匀分摊。',
- brokerName,
- ),
- );
- }
-
if (routeAddresses(route).some((addr) => duplicateAddresses.has(addr))) {
issues.push(
issue(
@@ -421,7 +395,7 @@ export const analyzeTopicRoutes = (routes: BrokerRoute[]):
TopicRouteDiagnostics
const distributions = routes.map<RouteDistribution>((route, index) => {
const brokerIds = routeBrokerIds(route);
- const routeIssues = distributionIssues(route, writeSkew, readSkew,
duplicateAddresses);
+ const routeIssues = distributionIssues(route, duplicateAddresses);
return {
key: `${route.brokerName || 'broker'}-${index}`,
@@ -444,6 +418,26 @@ export const analyzeTopicRoutes = (routes: BrokerRoute[]):
TopicRouteDiagnostics
});
const issues = distributions.flatMap((distribution) => distribution.issues);
+ if (hasSkew(writeSkew)) {
+ issues.push(
+ issue(
+ 'WRITE_QUEUE_SKEW',
+ 'warning',
+ '写队列分布不均',
+ '不同 Broker 的写队列数差距较大，生产流量可能无法均匀分摊。',
+ ),
+ );
+ }
+ if (hasSkew(readSkew)) {
+ issues.push(
+ issue(
+ 'READ_QUEUE_SKEW',
+ 'warning',
+ '读队列分布不均',
+ '不同 Broker 的读队列数差距较大，消费者负载可能无法均匀分摊。',
+ ),
+ );
+ }
const writableBrokerCount = distributions.filter(
(distribution) => distribution.writable && distribution.writeQueues > 0,
).length;
