diff --git a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/reduce/AlarmGroupReduce.java b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/reduce/AlarmGroupReduce.java index 597ef9a58a..fac5f7b268 100644 --- a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/reduce/AlarmGroupReduce.java +++ b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/reduce/AlarmGroupReduce.java @@ -171,7 +171,7 @@ public class AlarmGroupReduce implements DisposableBean { .factory()); } - private void runCheckAndSendGroups() { + void runCheckAndSendGroups() { beforeCheckAndSendGroupsRun(); try { long now = System.currentTimeMillis(); @@ -179,7 +179,6 @@ public class AlarmGroupReduce implements DisposableBean { if (shouldSendGroup(cache, now)) { sendGroupAlert(cache); cache.setLastSendTime(now); - cache.getAlertFingerprints().clear(); } }); } catch (Exception e) { @@ -262,22 +261,19 @@ public class AlarmGroupReduce implements DisposableBean { return newCache; }); String fingerprint = alert.getFingerprint(); - // Check if this is a duplicate alert + // Preserve the original startAt when updating an alert that is still tracked SingleAlert existingAlert = cache.getAlertFingerprints().get(fingerprint); if (existingAlert != null) { - // Update existing alert timestamp alert.setStartAt(existingAlert.getStartAt()); - cache.getAlertFingerprints().put(fingerprint, alert); - return; } - - // Add new alert + // Add or update the alert. The cache retains every currently-active alert of the + // group, so the group status is always computed over the full member set rather + // than only the alerts received within the current send window. cache.getAlertFingerprints().put(fingerprint, alert); - + if (shouldSendGroupImmediately(cache)) { sendGroupAlert(cache); cache.setLastSendTime(System.currentTimeMillis()); - cache.getAlertFingerprints().clear(); } } @@ -288,21 +284,27 @@ public class AlarmGroupReduce implements DisposableBean { long now = System.currentTimeMillis(); String status = determineGroupStatus(cache.getAlertFingerprints().values()); - + + boolean hasResolvedAlert = cache.getAlertFingerprints().values().stream() + .anyMatch(alert -> CommonConstants.ALERT_STATUS_RESOLVED.equals(alert.getStatus())); + // For firing alerts, check repeat interval if (CommonConstants.ALERT_STATUS_FIRING.equals(status)) { AlertGroupConverge ruleConfig = groupDefines.get(cache.getGroupDefineName()); long repeatInterval = ruleConfig.getRepeatInterval() != null ? ruleConfig.getRepeatInterval() * MS_PER_SECOND : DEFAULT_REPEAT_INTERVAL; - - // Skip if within repeat interval - if (cache.getLastRepeatTime() > 0 + + // Skip if within repeat interval. The throttle only suppresses repeated firing + // notifications; it must never swallow a pending resolved transition, so we still + // send when the batch carries a member that has just recovered. + if (!hasResolvedAlert + && cache.getLastRepeatTime() > 0 && now - cache.getLastRepeatTime() < repeatInterval) { return; } cache.setLastRepeatTime(now); } - + GroupAlert groupAlert = GroupAlert.builder() .groupKey(cache.getGroupKey()) .groupLabels(cache.getGroupLabels()) @@ -313,6 +315,14 @@ public class AlarmGroupReduce implements DisposableBean { .build(); alarmInhibitReduce.inhibitAlarm(groupAlert); + + // The resolved members have now been emitted, so drop them from the group. Firing + // members are retained until they recover, keeping the group firing while any member + // is still active instead of flushing the whole cache after every send. + if (hasResolvedAlert) { + cache.getAlertFingerprints().values().removeIf( + alert -> CommonConstants.ALERT_STATUS_RESOLVED.equals(alert.getStatus())); + } } private boolean shouldSendGroup(GroupAlertCache cache, long now) { diff --git a/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/reduce/AlarmGroupReduceTest.java b/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/reduce/AlarmGroupReduceTest.java index f6bf2c3598..4611d52138 100644 --- a/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/reduce/AlarmGroupReduceTest.java +++ b/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/reduce/AlarmGroupReduceTest.java @@ -24,12 +24,14 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -38,10 +40,12 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.hertzbeat.alert.dao.AlertGroupConvergeDao; import org.apache.hertzbeat.common.config.VirtualThreadProperties; import org.apache.hertzbeat.common.entity.alerter.AlertGroupConverge; +import org.apache.hertzbeat.common.entity.alerter.GroupAlert; import org.apache.hertzbeat.common.entity.alerter.SingleAlert; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.MockitoAnnotations; @@ -146,6 +150,87 @@ class AlarmGroupReduceTest { assertEquals(1, maxConcurrent.get()); } + /** + * Regression for issue #4160 (Bug 1): while one member of a group is still firing, the + * recovery of another member must not flip the whole group to resolved. The group has to + * stay firing until every member has actually cleared. + */ + @Test + void whenOneMemberRecoversButAnotherStillFiring_groupMustNotResolve() { + AlertGroupConverge rule = groupRule(); + alarmGroupReduce.refreshGroupDefines(Collections.singletonList(rule)); + + alarmGroupReduce.processGroupAlert(alert("cpu", "firing", "host1")); + alarmGroupReduce.processGroupAlert(alert("mem", "firing", "host1")); + // First group send: both members firing. + alarmGroupReduce.runCheckAndSendGroups(); + + // CPU recovers, memory is still firing. + alarmGroupReduce.processGroupAlert(alert("cpu", "resolved", "host1")); + alarmGroupReduce.runCheckAndSendGroups(); + + ArgumentCaptor captor = ArgumentCaptor.forClass(GroupAlert.class); + verify(alarmInhibitReduce, atLeastOnce()).inhibitAlarm(captor.capture()); + List groups = captor.getAllValues(); + + // Memory never recovered, so no group push may ever carry a resolved group status. + assertTrue(groups.stream().noneMatch(g -> "resolved".equals(g.getStatus())), + "group wrongly resolved while a member alert was still firing"); + // The CPU recovery is still communicated, inside a group that stays firing. + assertTrue(groups.stream().anyMatch(g -> "firing".equals(g.getStatus()) + && g.getAlerts().stream().anyMatch( + a -> "cpu".equals(a.getFingerprint()) && "resolved".equals(a.getStatus()))), + "CPU recovery was not communicated within the still-firing group"); + } + + /** + * Regression for issue #4160 (Bug 2): a resolved transition that happens while the group is + * firing and inside the firing repeat-interval window must still be emitted. The firing + * throttle may only suppress repeated firing notifications, never a pending recovery. + */ + @Test + void whenMemberRecoversInsideRepeatInterval_recoveryMustStillBeEmitted() { + AlertGroupConverge rule = groupRule(); + alarmGroupReduce.refreshGroupDefines(Collections.singletonList(rule)); + + alarmGroupReduce.processGroupAlert(alert("cpu", "firing", "host1")); + alarmGroupReduce.processGroupAlert(alert("mem", "firing", "host1")); + // First send arms the firing repeat-interval throttle. + alarmGroupReduce.runCheckAndSendGroups(); + + // CPU keeps firing, memory recovers within the repeat interval. + alarmGroupReduce.processGroupAlert(alert("cpu", "firing", "host1")); + alarmGroupReduce.processGroupAlert(alert("mem", "resolved", "host1")); + alarmGroupReduce.runCheckAndSendGroups(); + + ArgumentCaptor captor = ArgumentCaptor.forClass(GroupAlert.class); + verify(alarmInhibitReduce, atLeastOnce()).inhibitAlarm(captor.capture()); + List groups = captor.getAllValues(); + + assertTrue(groups.stream().anyMatch(g -> g.getAlerts().stream().anyMatch( + a -> "mem".equals(a.getFingerprint()) && "resolved".equals(a.getStatus()))), + "memory recovery was silently dropped by the firing repeat-interval throttle"); + } + + private AlertGroupConverge groupRule() { + AlertGroupConverge rule = new AlertGroupConverge(); + rule.setName("test-rule"); + rule.setGroupLabels(Collections.singletonList("instance")); + rule.setGroupWait(0L); + rule.setGroupInterval(0L); + rule.setRepeatInterval(3600L); + return rule; + } + + private SingleAlert alert(String fingerprint, String status, String instance) { + return SingleAlert.builder() + .fingerprint(fingerprint) + .status(status) + .labels(createLabels("instance", instance)) + .annotations(new HashMap<>()) + .build(); + } + private Map createLabels(String... keyValues) { Map labels = new HashMap<>(); for (int i = 0; i < keyValues.length; i += 2) {