[bugfix] Alarm group convergence must not resolve while members are still firing (#4316)

Co-authored-by: aias00 <liuhongyu@apache.org>

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) {

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<GroupAlert> captor = ArgumentCaptor.forClass(GroupAlert.class);
+        verify(alarmInhibitReduce, atLeastOnce()).inhibitAlarm(captor.capture());
+        List<GroupAlert> 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<GroupAlert> captor = ArgumentCaptor.forClass(GroupAlert.class);
+        verify(alarmInhibitReduce, atLeastOnce()).inhibitAlarm(captor.capture());
+        List<GroupAlert> 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<String, String> createLabels(String... keyValues) {
         Map<String, String> labels = new HashMap<>();
         for (int i = 0; i < keyValues.length; i += 2) {