🚀 Optimized alarm event grouping and processing logic

Author: Cairry

alert/consumer/consumer.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"regexp"
 	"runtime/debug"
-	"sort"
 	"sync"
 	"time"
 	"watchAlert/alert/mute"
@@ -23,6 +22,10 @@ const (
 
 	// 默认处理时间间隔
 	DefaultProcessTime = 1
+
+	// 状态前缀
+	RecoverStatePrefix = "Recover_"
+	FiringStatePrefix  = "Firing_"
 )
 
 type (
@@ -46,12 +49,12 @@ type (
 
 	RulesGroup struct {
 		RuleID string // 规则组 ID
-		Groups []EventsGroup
+		Groups map[string]EventsGroup
 	}
 
 	AlertGroups struct {
-		Rules []RulesGroup // 告警事件列表, 根据规则划分组
-		lock  sync.Mutex
+		Rules map[string]RulesGroup
+		lock  sync.RWMutex
 	}
 )
 
@@ -62,41 +65,37 @@ func (ag *AlertGroups) AddAlert(stateId string, alert *models.AlertCurEvent, fau
 
 	// 获取通知对象 ID 列表 用于事件分组
 	noticeObjIds := ag.getNoticeId(alert, faultCenter)
+	if len(noticeObjIds) == 0 {
+		return // 如果没有通知对象ID，则跳过
+	}
 
 	for _, noticeObjId := range noticeObjIds {
-		// 查找 Rule 位置
-		rulePos := ag.getRuleNodePos(stateId)
-
-		// Rule 存在时的处理，找到对应的规则组
-		if rulePos < len(ag.Rules) && ag.Rules[rulePos].RuleID == stateId {
-			rule := &ag.Rules[rulePos]
-
-			// 查找 Group 位置
-			groupPos := ag.getGroupNodePos(rule, noticeObjId)
-
-			if groupPos < len(rule.Groups) && (rule.Groups)[groupPos].NoticeID == noticeObjId {
-				// 追加事件
-				(rule.Groups)[groupPos].Events = append((rule.Groups)[groupPos].Events, alert)
-			} else {
-				// 实例化新的 EventGroup
-				rule.Groups = append(rule.Groups, EventsGroup{
-					NoticeID: noticeObjId,
-					Events:   []*models.AlertCurEvent{alert},
-				})
-			}
-			continue
-		} else {
-			// 实例化新的 RuleGroup
-			ag.Rules = append(ag.Rules, RulesGroup{
+		// 检查 Rule 是否存在
+		rule, exists := ag.Rules[stateId]
+		if !exists {
+			// 创建新的 RuleGroup
+			ag.Rules[stateId] = RulesGroup{
 				RuleID: stateId,
-				Groups: []EventsGroup{
-					{
-						NoticeID: noticeObjId,
-						Events:   []*models.AlertCurEvent{alert},
-					},
-				},
-			})
+				Groups: make(map[string]EventsGroup),
+			}
+			rule = ag.Rules[stateId]
 		}
+
+		// 检查 Group 是否存在
+		group, groupExists := rule.Groups[noticeObjId]
+		if !groupExists {
+			// 创建新的 EventsGroup
+			rule.Groups[noticeObjId] = EventsGroup{
+				NoticeID: noticeObjId,
+				Events:   []*models.AlertCurEvent{},
+			}
+			group = rule.Groups[noticeObjId]
+		}
+
+		// 添加事件到对应组
+		group.Events = append(group.Events, alert)
+		// 更新 group 映射
+		ag.Rules[stateId].Groups[noticeObjId] = group
 	}
 }
 
@@ -120,32 +119,6 @@ func (ag *AlertGroups) getNoticeId(alert *models.AlertCurEvent, faultCenter mode
 	return faultCenter.NoticeIds
 }
 
-// getRuleNodePos 获取 Rule 点位
-func (ag *AlertGroups) getRuleNodePos(ruleId string) int {
-	// Rules 切片排序
-	sort.Slice(ag.Rules, func(i, j int) bool {
-		return ag.Rules[i].RuleID < ag.Rules[j].RuleID
-	})
-
-	// 查找Rule位置
-	return sort.Search(len(ag.Rules), func(i int) bool {
-		return ag.Rules[i].RuleID >= ruleId
-	})
-}
-
-// getGroupNodePos 获取 Event 点位
-func (ag *AlertGroups) getGroupNodePos(rule *RulesGroup, groupId string) int {
-	// Groups 切片排序
-	sort.Slice(rule.Groups, func(i, j int) bool {
-		return rule.Groups[i].NoticeID < rule.Groups[j].NoticeID
-	})
-
-	// 查找Group位置
-	return sort.Search(len(rule.Groups), func(i int) bool {
-		return (rule.Groups)[i].NoticeID >= groupId
-	})
-}
-
 func NewConsumerWork(ctx *ctx.Context) ConsumeInterface {
 	return &Consume{
 		ctx: ctx,
@@ -213,14 +186,16 @@ func (c *Consume) executeTask(faultCenter models.FaultCenter, taskChan chan stru
 	// 获取故障中心的所有告警事件
 	data, err := c.ctx.Redis.Alert().GetAllEvents(models.BuildAlertEventCacheKey(faultCenter.TenantId, faultCenter.ID))
 	if err != nil {
-		logc.Error(c.ctx.Ctx, fmt.Sprintf("从 Redis 中获取事件信息错误, faultCenterKey: %s, err: %s", models.BuildAlertEventCacheKey(faultCenter.TenantId, faultCenter.ID), err.Error()))
+		logc.Errorf(c.ctx.Ctx, "从 Redis 中获取事件信息错误, faultCenterKey: %s, err: %s", models.BuildAlertEventCacheKey(faultCenter.TenantId, faultCenter.ID), err.Error())
 		return
 	}
 
 	// 事件过滤
 	filterEvents := c.filterAlertEvents(faultCenter, data)
 	// 事件分组
-	var alertGroups AlertGroups
+	alertGroups := AlertGroups{
+		Rules: make(map[string]RulesGroup),
+	}
 	c.alarmGrouping(faultCenter, &alertGroups, filterEvents)
 	// 发送事件
 	c.sendAlerts(faultCenter, &alertGroups)
@@ -297,13 +272,10 @@ func (c *Consume) alarmGrouping(faultCenter models.FaultCenter, alertGroups *Ale
 	for _, alert := range alerts {
 		// 状态+规则 = 状态 ID
 		var stateId string
-		switch alert.IsRecovered {
-		case true:
-			stateId = "Recover_" + alert.RuleId
-		case false:
-			stateId = "Firing_" + alert.RuleId
-		default:
-			stateId = "Unknown_" + alert.RuleId
+		if alert.IsRecovered {
+			stateId = RecoverStatePrefix + alert.RuleId
+		} else {
+			stateId = FiringStatePrefix + alert.RuleId
 		}
 
 		alertGroups.AddAlert(stateId, alert, faultCenter)
@@ -335,7 +307,7 @@ func (c *Consume) processAlertGroup(faultCenter models.FaultCenter, noticeId str
 	g.Go(func() error { return process.HandleAlert(c.ctx, "alarm", faultCenter, noticeId, alerts) })
 
 	if err := g.Wait(); err != nil {
-		logc.Error(c.ctx.Ctx, fmt.Sprintf("Alert group processing failed: %v", err))
+		logc.Errorf(c.ctx.Ctx, "Alert group processing failed: %v", err)
 	}
 }
 

alert/eval/query.go

@@ -145,7 +145,7 @@ func metrics(ctx *ctx.Context, datasourceId, datasourceType string, rule models.
 				// 更新恢复时最新值
 				cache, err := ctx.Redis.Alert().GetEventFromCache(event.TenantId, event.FaultCenterId, event.Fingerprint)
 				if err == nil {
-					if !cache.IsRecovered || cache.Status != models.StateRecovered {
+					if !cache.IsRecovered && cache.Status != models.StateRecovered {
 						event.Labels["value"] = v.GetValue()
 						process.PushEventToFaultCenter(ctx, &event)
 					}

alert/process/handle.go

@@ -128,20 +128,17 @@ func withRuleGroupByAlerts(ctx *ctx.Context, timeInt int64, alerts []*models.Ale
 		return alerts
 	}
 
-	var aggregatedAlert *models.AlertCurEvent
 	for i := range alerts {
 		alert := alerts[i]
-		if !strings.Contains(alert.Annotations, "聚合") {
-			alert.Annotations += fmt.Sprintf("\n聚合 %d 条告警\n", len(alerts))
-		}
-		aggregatedAlert = alert
-
 		if !alert.IsRecovered {
 			alert.LastSendTime = timeInt
 			ctx.Redis.Alert().PushAlertEvent(alert)
 		}
 	}
 
+	aggregatedAlert := alerts[0]
+	aggregatedAlert.Annotations += fmt.Sprintf("\n聚合 %d 条消息，详情请前往 WatchAlert 查看\n", len(alerts))
+
 	return []*models.AlertCurEvent{aggregatedAlert}
 }
 

internal/models/alert_current_event.go

@@ -77,9 +77,6 @@ func (alert *AlertCurEvent) TransitionStatus(newStatus AlertStatus) error {
 		return err
 	}
 
-	// 更新状态
-	alert.Status = newStatus
-
 	return nil
 }
 
@@ -120,19 +117,20 @@ func (alert *AlertCurEvent) validateTransition(newState AlertStatus) error {
 func (alert *AlertCurEvent) handleStateTransition(newState AlertStatus) error {
 	now := time.Now().Unix()
 
+	// 更新状态
+	alert.Status = newState
+
 	switch newState {
 	case StatePreAlert:
 		alert.FirstTriggerTime = now
 		alert.LastEvalTime = now
+
 	case StateAlerting:
 	case StateRecovered:
-		if alert.IsRecovered == true && alert.Status == StateRecovered {
-			return nil
-		}
-
 		alert.LastSendTime = 0
 		alert.RecoverTime = now
 		alert.IsRecovered = true
+
 	case StateSilenced:
 	}