diff --git a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/notice/AlertNoticeDispatch.java b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/notice/AlertNoticeDispatch.java index b294e817cd8..c6cfa4084a8 100644 --- a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/notice/AlertNoticeDispatch.java +++ b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/notice/AlertNoticeDispatch.java @@ -18,10 +18,16 @@ package org.apache.hertzbeat.alert.notice; import com.google.common.collect.Maps; +import java.time.LocalDateTime; +import java.util.Collection; +import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.concurrent.RejectedExecutionException; +import java.util.function.Function; +import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; import org.apache.hertzbeat.alert.AlerterWorkerPool; import org.apache.hertzbeat.alert.config.AlertSseManager; @@ -29,7 +35,9 @@ import org.apache.hertzbeat.common.entity.alerter.NoticeReceiver; import org.apache.hertzbeat.common.entity.alerter.NoticeRule; import org.apache.hertzbeat.common.entity.alerter.NoticeTemplate; +import org.apache.hertzbeat.common.entity.alerter.SingleAlert; import org.apache.hertzbeat.alert.service.NoticeConfigService; +import org.apache.hertzbeat.common.constants.CommonConstants; import org.apache.hertzbeat.common.util.JsonUtil; import org.apache.hertzbeat.plugin.PostAlertPlugin; import org.apache.hertzbeat.plugin.Plugin; @@ -124,6 +132,7 @@ public void dispatchAlarm(GroupAlert groupAlert) { private void sendNotify(GroupAlert alert) { matchNoticeRulesByAlert(alert).ifPresent(noticeRules -> noticeRules.forEach(rule -> { NoticeTemplate noticeTemplate = getOneTemplateById(rule.getTemplateId()); + GroupAlert noticeAlert = scopeAlertToRule(alert, rule); rule.getReceiverId().forEach(receiverId -> { NoticeReceiver receiver = getOneReceiverById(receiverId); if (receiver == null || receiver.getType() == null) { @@ -133,7 +142,7 @@ private void sendNotify(GroupAlert alert) { try { workerPool.executeNotify(receiver.getType(), () -> { try { - sendNoticeMsg(receiver, noticeTemplate, alert); + sendNoticeMsg(receiver, noticeTemplate, noticeAlert); } catch (AlertNoticeException e) { log.warn("DispatchTask sendNoticeMsg error, message: {}", e.getMessage()); } @@ -145,4 +154,100 @@ private void sendNotify(GroupAlert alert) { }); })); } + + private GroupAlert scopeAlertToRule(GroupAlert alert, NoticeRule rule) { + if (rule.isFilterAll() || rule.getLabels() == null || rule.getLabels().isEmpty() + || alert.getAlerts() == null) { + return alert; + } + List matchingAlerts = alert.getAlerts().stream() + .filter(singleAlert -> singleAlert.getLabels() != null + && rule.getLabels().entrySet().stream().allMatch(label -> + Objects.equals(label.getValue(), singleAlert.getLabels().get(label.getKey())))) + .toList(); + Map commonLabels = extractCommonAttributes(matchingAlerts, SingleAlert::getLabels); + Map commonAnnotations = + extractCommonAttributes(matchingAlerts, SingleAlert::getAnnotations); + Map groupLabels = extractGroupLabels(alert.getGroupLabels(), commonLabels); + String groupKey = Objects.equals(groupLabels, alert.getGroupLabels()) + ? alert.getGroupKey() : generateGroupKey(groupLabels); + return GroupAlert.builder() + .id(alert.getId()) + .groupKey(groupKey) + .status(determineGroupStatus(matchingAlerts)) + .groupLabels(groupLabels) + .commonLabels(commonLabels) + .commonAnnotations(commonAnnotations) + .alertFingerprints(matchingAlerts.stream() + .map(SingleAlert::getFingerprint) + .filter(Objects::nonNull) + .toList()) + .creator(alert.getCreator()) + .modifier(alert.getModifier()) + .gmtCreate(firstTime(matchingAlerts, SingleAlert::getGmtCreate, alert.getGmtCreate())) + .gmtUpdate(lastTime(matchingAlerts, SingleAlert::getGmtUpdate, alert.getGmtUpdate())) + .alerts(matchingAlerts) + .build(); + } + + private Map extractCommonAttributes( + Collection alerts, + Function> attributes) { + if (alerts.isEmpty()) { + return new HashMap<>(0); + } + Map firstAttributes = attributes.apply(alerts.iterator().next()); + Map common = + firstAttributes == null ? new HashMap<>(0) : new HashMap<>(firstAttributes); + for (SingleAlert alert : alerts) { + Map current = attributes.apply(alert); + common.keySet().removeIf(key -> + current == null || !current.containsKey(key) + || !Objects.equals(common.get(key), current.get(key))); + } + return common; + } + + private Map extractGroupLabels( + Map originalGroupLabels, + Map commonLabels) { + Map groupLabels = new HashMap<>(); + if (originalGroupLabels != null) { + originalGroupLabels.keySet().forEach(key -> { + if (commonLabels.containsKey(key)) { + groupLabels.put(key, commonLabels.get(key)); + } + }); + } + return groupLabels; + } + + private String generateGroupKey(Map groupLabels) { + return groupLabels.entrySet().stream() + .sorted(Map.Entry.comparingByKey()) + .map(entry -> entry.getKey() + ":" + entry.getValue()) + .collect(Collectors.joining(",")); + } + + private String determineGroupStatus(List alerts) { + return alerts.stream().anyMatch(alert -> + CommonConstants.ALERT_STATUS_FIRING.equals(alert.getStatus())) + ? CommonConstants.ALERT_STATUS_FIRING : CommonConstants.ALERT_STATUS_RESOLVED; + } + + private LocalDateTime firstTime( + List alerts, + Function time, + LocalDateTime fallback) { + return alerts.stream().map(time).filter(Objects::nonNull) + .min(LocalDateTime::compareTo).orElse(fallback); + } + + private LocalDateTime lastTime( + List alerts, + Function time, + LocalDateTime fallback) { + return alerts.stream().map(time).filter(Objects::nonNull) + .max(LocalDateTime::compareTo).orElse(fallback); + } } diff --git a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/NoticeConfigServiceImpl.java b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/NoticeConfigServiceImpl.java index eab8ad014ee..8469315c5d9 100644 --- a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/NoticeConfigServiceImpl.java +++ b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/NoticeConfigServiceImpl.java @@ -246,11 +246,7 @@ public List getReceiverFilterRule(GroupAlert alert) { CacheFactory.setNoticeCache(rules); } - // The temporary rule is to forward all, and then implement more matching rules: alarm status selection, monitoring type selection, etc. - // TODO: This matches an already-grouped alert against notice rules (group-then-route). It cannot fully - // separate alerts that were grouped together but should reach different receivers, so a rule matched by - // one alert still notifies the whole group. The ideal design is route-then-group (like Alertmanager): - // route each single alert by its labels first, then group per receiver. Tracked as a follow-up to #3852. + // Match grouped alerts here; dispatch scopes each notification to the single alerts matching its rule. return rules.stream() .filter(rule -> { if (!rule.isFilterAll()) { diff --git a/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/notice/AlertNoticeDispatchTest.java b/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/notice/AlertNoticeDispatchTest.java index ff4c92ef4cb..1dd407589fe 100644 --- a/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/notice/AlertNoticeDispatchTest.java +++ b/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/notice/AlertNoticeDispatchTest.java @@ -17,31 +17,38 @@ package org.apache.hertzbeat.alert.notice; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertAll; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyByte; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.time.LocalDateTime; import java.util.Collections; import java.util.List; +import java.util.Map; import org.apache.hertzbeat.alert.AlerterWorkerPool; import org.apache.hertzbeat.alert.config.AlertSseManager; import org.apache.hertzbeat.alert.service.NoticeConfigService; import org.apache.hertzbeat.common.entity.alerter.GroupAlert; import org.apache.hertzbeat.common.entity.alerter.NoticeReceiver; +import org.apache.hertzbeat.common.entity.alerter.NoticeRule; +import org.apache.hertzbeat.common.entity.alerter.NoticeTemplate; +import org.apache.hertzbeat.common.entity.alerter.SingleAlert; import org.apache.hertzbeat.plugin.runner.PluginRunner; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; -import org.apache.hertzbeat.common.entity.alerter.SingleAlert; -import org.apache.hertzbeat.common.entity.alerter.NoticeTemplate; /** * Test case for Alert Notice Dispatch @@ -180,4 +187,155 @@ void testDispatchAlarmUsesTypedNotifyExecution() { verify(alertNotifyHandler).send(eq(receiver), eq(template), eq(alert)); verify(emitterManager).broadcast(any(String.class)); } + + @Test + void testDispatchAlarmRecomputesNoticeFromAlertsMatchingRuleLabels() { + LocalDateTime matchingCreated = LocalDateTime.of(2026, 7, 30, 10, 0); + LocalDateTime matchingUpdated = LocalDateTime.of(2026, 7, 30, 10, 5); + SingleAlert matchingAlert = SingleAlert.builder() + .fingerprint("matching") + .labels(Map.of("department", "algorithm", "service", "checkout", "severity", "warning")) + .annotations(Map.of("summary", "algorithm summary", "runbook", "shared runbook")) + .content("matching-content") + .status("resolved") + .gmtCreate(matchingCreated) + .gmtUpdate(matchingUpdated) + .build(); + SingleAlert unrelatedAlert = SingleAlert.builder() + .fingerprint("unrelated") + .labels(Map.of("department", "infra", "service", "checkout", "severity", "critical")) + .annotations(Map.of("summary", "infra summary", "runbook", "shared runbook")) + .content("unrelated-content") + .status("firing") + .gmtCreate(matchingCreated.minusHours(1)) + .gmtUpdate(matchingUpdated.plusHours(1)) + .build(); + GroupAlert groupedAlert = GroupAlert.builder() + .id(2L) + .groupKey("department:infra,service:checkout") + .status("firing") + .groupLabels(Map.of("department", "infra", "service", "checkout")) + .commonLabels(Map.of("service", "checkout")) + .commonAnnotations(Map.of("runbook", "shared runbook")) + .alertFingerprints(List.of("matching", "unrelated")) + .gmtCreate(matchingCreated.minusHours(1)) + .gmtUpdate(matchingUpdated.plusHours(1)) + .alerts(List.of(matchingAlert, unrelatedAlert)) + .build(); + NoticeTemplate template = NoticeTemplate.builder().id(1L).build(); + NoticeRule rule = NoticeRule.builder() + .filterAll(false) + .labels(Map.of("department", "algorithm")) + .receiverId(List.of(1L)) + .templateId(1L) + .build(); + + when(alertStoreHandler.store(groupedAlert)).thenReturn(groupedAlert); + when(noticeConfigService.getReceiverFilterRule(groupedAlert)).thenReturn(List.of(rule)); + when(noticeConfigService.getReceiverById(1L)).thenReturn(receiver); + when(noticeConfigService.getOneTemplateById(1L)).thenReturn(template); + doAnswer(invocation -> { + Runnable task = invocation.getArgument(1); + task.run(); + return null; + }).when(workerPool).executeNotify(anyByte(), any(Runnable.class)); + + alertNoticeDispatch.dispatchAlarm(groupedAlert); + + ArgumentCaptor noticeAlert = ArgumentCaptor.forClass(GroupAlert.class); + verify(alertNotifyHandler).send(eq(receiver), eq(template), noticeAlert.capture()); + GroupAlert scopedAlert = noticeAlert.getValue(); + assertAll( + () -> assertEquals(List.of(matchingAlert), scopedAlert.getAlerts()), + () -> assertEquals(List.of("matching"), scopedAlert.getAlertFingerprints()), + () -> assertEquals("resolved", scopedAlert.getStatus()), + () -> assertEquals( + Map.of("department", "algorithm", "service", "checkout"), + scopedAlert.getGroupLabels()), + () -> assertEquals( + Map.of("department", "algorithm", "service", "checkout", "severity", "warning"), + scopedAlert.getCommonLabels()), + () -> assertEquals( + Map.of("summary", "algorithm summary", "runbook", "shared runbook"), + scopedAlert.getCommonAnnotations()), + () -> assertEquals("department:algorithm,service:checkout", scopedAlert.getGroupKey()), + () -> assertEquals(matchingCreated, scopedAlert.getGmtCreate()), + () -> assertEquals(matchingUpdated, scopedAlert.getGmtUpdate()), + () -> assertEquals(2, groupedAlert.getAlerts().size()), + () -> assertEquals("firing", groupedAlert.getStatus()), + () -> assertEquals(Map.of("service", "checkout"), groupedAlert.getCommonLabels())); + } + + @Test + void testDispatchAlarmScopesMultipleRulesForTheSameReceiverIndependently() { + SingleAlert algorithmAlert = SingleAlert.builder() + .fingerprint("algorithm") + .labels(Map.of("department", "algorithm", "service", "checkout")) + .annotations(Map.of("summary", "algorithm firing", "runbook", "algorithm runbook")) + .status("firing") + .build(); + SingleAlert algorithmResolvedAlert = SingleAlert.builder() + .fingerprint("algorithm-resolved") + .labels(Map.of("department", "algorithm", "service", "checkout")) + .annotations(Map.of("summary", "algorithm resolved", "runbook", "algorithm runbook")) + .status("resolved") + .build(); + SingleAlert infrastructureAlert = SingleAlert.builder() + .fingerprint("infra") + .labels(Map.of("department", "infra", "service", "checkout")) + .annotations(Map.of("summary", "infra summary")) + .status("resolved") + .build(); + GroupAlert groupedAlert = GroupAlert.builder() + .status("firing") + .groupLabels(Map.of("service", "checkout")) + .alerts(List.of(algorithmAlert, algorithmResolvedAlert, infrastructureAlert)) + .build(); + NoticeTemplate algorithmTemplate = NoticeTemplate.builder().id(1L).build(); + NoticeTemplate infrastructureTemplate = NoticeTemplate.builder().id(2L).build(); + NoticeRule algorithmRule = NoticeRule.builder() + .filterAll(false) + .labels(Map.of("department", "algorithm")) + .receiverId(List.of(1L)) + .templateId(1L) + .build(); + NoticeRule infrastructureRule = NoticeRule.builder() + .filterAll(false) + .labels(Map.of("department", "infra")) + .receiverId(List.of(1L)) + .templateId(2L) + .build(); + + when(alertStoreHandler.store(groupedAlert)).thenReturn(groupedAlert); + when(noticeConfigService.getReceiverFilterRule(groupedAlert)) + .thenReturn(List.of(algorithmRule, infrastructureRule)); + when(noticeConfigService.getReceiverById(1L)).thenReturn(receiver); + when(noticeConfigService.getOneTemplateById(1L)).thenReturn(algorithmTemplate); + when(noticeConfigService.getOneTemplateById(2L)).thenReturn(infrastructureTemplate); + doAnswer(invocation -> { + Runnable task = invocation.getArgument(1); + task.run(); + return null; + }).when(workerPool).executeNotify(anyByte(), any(Runnable.class)); + + alertNoticeDispatch.dispatchAlarm(groupedAlert); + + ArgumentCaptor templates = ArgumentCaptor.forClass(NoticeTemplate.class); + ArgumentCaptor alerts = ArgumentCaptor.forClass(GroupAlert.class); + verify(alertNotifyHandler, times(2)).send(eq(receiver), templates.capture(), alerts.capture()); + assertAll( + () -> assertEquals(List.of(algorithmTemplate, infrastructureTemplate), templates.getAllValues()), + () -> assertEquals( + List.of("algorithm", "algorithm-resolved"), + alerts.getAllValues().get(0).getAlertFingerprints()), + () -> assertEquals(List.of("infra"), alerts.getAllValues().get(1).getAlertFingerprints()), + () -> assertEquals("firing", alerts.getAllValues().get(0).getStatus()), + () -> assertEquals("resolved", alerts.getAllValues().get(1).getStatus()), + () -> assertEquals( + Map.of("runbook", "algorithm runbook"), + alerts.getAllValues().get(0).getCommonAnnotations()), + () -> assertEquals( + Map.of("summary", "infra summary"), + alerts.getAllValues().get(1).getCommonAnnotations())); + } }