✨ Supported multi instance leader elector

Author: Cairry

alert/alert.go

@@ -1,10 +1,16 @@
 package alert
 
 import (
+	"context"
 	"watchAlert/alert/consumer"
 	"watchAlert/alert/eval"
 	"watchAlert/alert/probing"
 	"watchAlert/internal/ctx"
+	"watchAlert/internal/global"
+	"watchAlert/pkg/client"
+	"watchAlert/pkg/tools"
+
+	"github.com/zeromicro/go-zero/core/logc"
 )
 
 var (
@@ -13,18 +19,199 @@ var (
 
 	ProductProbing probing.ProductProbing
 	ConsumeProbing probing.ConsumeProbing
+
+	// Leader 选举器
+	LeaderElector *tools.LeaderElector
+
+	// 消息订阅取消函数
+	subscriberCancels []context.CancelFunc
+
+	// 选举开关
+	leaderElectionEnabled bool
 )
 
 func Initialize(ctx *ctx.Context) {
 	// 初始化告警规则评估任务
 	AlertRule = eval.NewAlertRuleEval(ctx)
-	AlertRule.RestartAllEvals()
-
 	ConsumerWork = consumer.NewConsumerWork(ctx)
-	ConsumerWork.RestartAllConsumers()
 
 	// 初始化拨测任务
 	ConsumeProbing = probing.NewProbingConsumerTask(ctx)
 	ProductProbing = probing.NewProbingTask(ctx)
+
+	// 检查 Leader 选举是否启用
+	leaderElectionEnabled = global.Config.Server.EnableElection
+
+	if leaderElectionEnabled {
+		// 启用 Leader 选举模式
+		logc.Infof(ctx.Ctx, "Leader 选举已启用，开始选举流程...")
+		LeaderElector = tools.NewLeaderElector(
+			ctx.Ctx,
+			client.Redis,
+			loadRules,
+			unloadRules,
+		)
+		// 启动 Leader 选举
+		LeaderElector.Start()
+	} else {
+		loadRules()
+	}
+}
+
+// loadRules 加载所有规则(成为 Leader 时调用)
+func loadRules() {
+	logc.Infof(ctx.Ctx, "本节点为 Leader 节点，开始加载规则...")
+
+	// 重启所有告警规则评估器
+	AlertRule.RestartAllEvals()
+
+	// 重启所有故障中心消费者
+	ConsumerWork.RestartAllConsumers()
+
+	// 重启所有拨测任务
 	ProductProbing.RePushRule(&ConsumeProbing)
+
+	// 启动 Redis 消息订阅，监听规则变更
+	startMessageSubscribers()
+}
+
+// startMessageSubscribers 启动消息订阅器
+func startMessageSubscribers() {
+	subscriberCancels = make([]context.CancelFunc, 0)
+
+	// 订阅告警规则重载消息
+	subCtx1, cancel1 := context.WithCancel(ctx.Ctx)
+	subscriberCancels = append(subscriberCancels, cancel1)
+	go tools.SubscribeReloadMessages(subCtx1, client.Redis, tools.ChannelRuleReload, handleRuleReload)
+
+	// 订阅故障中心重载消息
+	subCtx2, cancel2 := context.WithCancel(ctx.Ctx)
+	subscriberCancels = append(subscriberCancels, cancel2)
+	go tools.SubscribeReloadMessages(subCtx2, client.Redis, tools.ChannelFaultCenterReload, handleFaultCenterReload)
+
+	// 订阅拨测规则重载消息
+	subCtx3, cancel3 := context.WithCancel(ctx.Ctx)
+	subscriberCancels = append(subscriberCancels, cancel3)
+	go tools.SubscribeReloadMessages(subCtx3, client.Redis, tools.ChannelProbingReload, handleProbingReload)
+}
+
+// stopMessageSubscribers 停止消息订阅器
+func stopMessageSubscribers() {
+	for _, cancel := range subscriberCancels {
+		cancel()
+	}
+	subscriberCancels = nil
+	logc.Infof(ctx.Ctx, "消息订阅器已停止")
+}
+
+// handleRuleReload 处理告警规则重载消息
+func handleRuleReload(msg tools.ReloadMessage) {
+
+	// 从数据库获取规则
+	rule := ctx.DB.Rule().GetRuleObject(msg.ID)
+	if rule.RuleId == "" {
+		logc.Errorf(ctx.Ctx, "规则不存在: %s", msg.ID)
+		return
+	}
+
+	switch msg.Action {
+	case tools.ActionCreate, tools.ActionEnable:
+		if rule.Enabled != nil && *rule.Enabled {
+			AlertRule.Submit(rule)
+			logc.Infof(ctx.Ctx, "[Leader] 已启动规则评估: %s", msg.Name)
+		}
+
+	case tools.ActionUpdate:
+		AlertRule.Stop(msg.ID)
+		if rule.Enabled != nil && *rule.Enabled {
+			AlertRule.Submit(rule)
+			logc.Infof(ctx.Ctx, "[Leader] 已重启规则评估: %s", msg.Name)
+		}
+
+	case tools.ActionDelete, tools.ActionDisable:
+		AlertRule.Stop(msg.ID)
+		logc.Infof(ctx.Ctx, "[Leader] 已停止规则评估: %s", msg.Name)
+	}
+}
+
+// handleFaultCenterReload 处理故障中心重载消息
+func handleFaultCenterReload(msg tools.ReloadMessage) {
+	fc, err := ctx.DB.FaultCenter().Get(msg.TenantID, msg.ID, "")
+	if err != nil {
+		logc.Errorf(ctx.Ctx, "故障中心不存在: %s, err: %v", msg.ID, err)
+		return
+	}
+
+	switch msg.Action {
+	case tools.ActionCreate, tools.ActionEnable:
+		ConsumerWork.Submit(fc)
+		logc.Infof(ctx.Ctx, "[Leader] 已启动故障中心消费: %s", msg.Name)
+
+	case tools.ActionUpdate:
+		ConsumerWork.Stop(msg.ID)
+		ConsumerWork.Submit(fc)
+		logc.Infof(ctx.Ctx, "[Leader] 已重启故障中心消费: %s", msg.Name)
+
+	case tools.ActionDelete, tools.ActionDisable:
+		ConsumerWork.Stop(msg.ID)
+		logc.Infof(ctx.Ctx, "[Leader] 已停止故障中心消费: %s", msg.Name)
+	}
+}
+
+// handleProbingReload 处理拨测规则重载消息
+func handleProbingReload(msg tools.ReloadMessage) {
+	rule, err := ctx.DB.Probing().Search(msg.TenantID, msg.ID)
+	if err != nil {
+		logc.Errorf(ctx.Ctx, "拨测规则不存在: %s, err: %v", msg.ID, err)
+		return
+	}
+	switch msg.Action {
+	case tools.ActionCreate, tools.ActionEnable:
+		if rule.Enabled != nil && *rule.Enabled {
+			ProductProbing.Add(rule)
+			ConsumeProbing.Add(rule)
+			logc.Infof(ctx.Ctx, "[Leader] 已启动拨测任务: %s", msg.Name)
+		}
+
+	case tools.ActionUpdate:
+		ProductProbing.Stop(msg.ID)
+		ConsumeProbing.Stop(msg.ID)
+		if rule.Enabled != nil && *rule.Enabled {
+			ProductProbing.Add(rule)
+			ConsumeProbing.Add(rule)
+			logc.Infof(ctx.Ctx, "[Leader] 已重启拨测任务: %s", msg.Name)
+		}
+
+	case tools.ActionDelete, tools.ActionDisable:
+		ProductProbing.Stop(msg.ID)
+		ConsumeProbing.Stop(msg.ID)
+		logc.Infof(ctx.Ctx, "[Leader] 已停止拨测任务: %s", msg.Name)
+	}
+}
+
+// unloadRules 卸载所有规则(失去 Leader 时调用)
+func unloadRules() {
+	logc.Infof(ctx.Ctx, "本节点失去 Leader 身份，停止所有任务...")
+
+	// 停止消息订阅
+	stopMessageSubscribers()
+
+	// 停止所有告警规则评估器
+	AlertRule.StopAllEvals()
+
+	// 停止所有故障中心消费者
+	ConsumerWork.StopAllConsumers()
+
+	// 停止所有拨测任务
+	ProductProbing.StopAllTasks()
+	ConsumeProbing.StopAllTasks()
+}
+
+// IsLeader 判断节点角色
+func IsLeader() bool {
+	if !leaderElectionEnabled {
+		return true
+	}
+
+	return LeaderElector != nil && LeaderElector.IsLeader()
 }

alert/consumer/consumer.go

@@ -3,8 +3,6 @@ package consumer
 import (
 	"context"
 	"fmt"
-	"github.com/zeromicro/go-zero/core/logc"
-	"golang.org/x/sync/errgroup"
 	"regexp"
 	"runtime/debug"
 	"sort"
@@ -14,6 +12,9 @@ import (
 	"watchAlert/alert/process"
 	"watchAlert/internal/ctx"
 	"watchAlert/internal/models"
+
+	"github.com/zeromicro/go-zero/core/logc"
+	"golang.org/x/sync/errgroup"
 )
 
 const (
@@ -30,6 +31,7 @@ type (
 		Stop(faultCenterId string)
 		Watch(ctx context.Context, faultCenter models.FaultCenter)
 		RestartAllConsumers()
+		StopAllConsumers()
 	}
 
 	Consume struct {
@@ -403,3 +405,24 @@ func (c *Consume) processSilenceRule(faultCenter models.FaultCenter) {
 		silenceCtx.PushAlertMute(*muteRule)
 	}
 }
+
+// StopAllConsumers 停止所有消费者
+func (c *Consume) StopAllConsumers() {
+	c.ctx.Mux.Lock()
+	defer c.ctx.Mux.Unlock()
+
+	count := len(c.ctx.ContextMap)
+	if count == 0 {
+		return
+	}
+
+	logc.Infof(c.ctx.Ctx, "停止 %d 个故障中心消费者...", count)
+
+	// 取消所有消费任务
+	for fcId, cancel := range c.ctx.ContextMap {
+		cancel()
+		delete(c.ctx.ContextMap, fcId)
+	}
+
+	logc.Infof(c.ctx.Ctx, "所有故障中心消费者已停止")
+}

alert/consumer/upgrader.go

@@ -1,14 +1,14 @@
 package consumer
 
 import (
-	"context"
 	"fmt"
-	"github.com/zeromicro/go-zero/core/logc"
 	"time"
 	"watchAlert/alert/mute"
 	"watchAlert/alert/process"
 	"watchAlert/internal/ctx"
 	"watchAlert/internal/models"
+
+	"github.com/zeromicro/go-zero/core/logc"
 )
 
 // alarmUpgrade 处理告警升级主入口
@@ -17,7 +17,7 @@ func alarmUpgrade(ctx *ctx.Context, faultCenter models.FaultCenter, alerts map[s
 		return nil
 	}
 
-	filterAlerts := filterAlertEvents(ctx.Ctx, faultCenter, alerts)
+	filterAlerts := filterAlertEvents(faultCenter, alerts)
 	currentTime := time.Now().Unix()
 
 	confirmAggregated := createAggregatedAlert(models.ConfirmStatus, faultCenter)
@@ -139,7 +139,7 @@ func getContent(number int) string {
 }
 
 // filterAlertEvents 过滤告警事件
-func filterAlertEvents(ctx context.Context, faultCenter models.FaultCenter, alerts map[string]*models.AlertCurEvent) []*models.AlertCurEvent {
+func filterAlertEvents(faultCenter models.FaultCenter, alerts map[string]*models.AlertCurEvent) []*models.AlertCurEvent {
 	var newEvents []*models.AlertCurEvent
 	for _, event := range alerts {
 		// 过滤掉 预告警, 待恢复 状态的事件
@@ -175,10 +175,15 @@ func isMutedEvent(event *models.AlertCurEvent, faultCenter models.FaultCenter) b
 // sendAggregatedAlert 发送聚合后的告警函数
 func sendAggregatedAlert(ctx *ctx.Context, faultCenter models.FaultCenter, aggregated *AggregatedAlert) error {
 	var noticeId string
-	if aggregated.Status == models.ConfirmStatus {
-		noticeId = faultCenter.GetUpgradeNoticeId(models.ConfirmStatus)
-	} else if aggregated.Status == models.HandleStatus {
-		noticeId = faultCenter.GetUpgradeNoticeId(models.HandleStatus)
+	switch aggregated.Status {
+	case models.ConfirmStatus:
+		if faultCenter.GetStrategy(models.ConfirmStatus).GetEnabled() {
+			noticeId = faultCenter.GetUpgradeNoticeId(models.ConfirmStatus)
+		}
+	case models.HandleStatus:
+		if faultCenter.GetStrategy(models.HandleStatus).GetEnabled() {
+			noticeId = faultCenter.GetUpgradeNoticeId(models.HandleStatus)
+		}
 	}
 
 	logc.Alert(ctx.Ctx, fmt.Sprintf("Aggregated alarm %s timeout fingerprints: %v, exceeded %d min",

alert/eval/eval.go

@@ -63,6 +63,7 @@ type (
 		Eval(ctx context.Context, rule models.AlertRule)
 		Recover(tenantId, ruleId string, eventCacheKey models.AlertEventCacheKey, faultCenterInfoKey models.FaultCenterInfoCacheKey, curFingerprints []string)
 		RestartAllEvals()
+		StopAllEvals()
 	}
 
 	// AlertRule 告警规则
@@ -412,3 +413,24 @@ func (t *AlertRule) getRuleList() ([]models.AlertRule, error) {
 	}
 	return ruleList, nil
 }
+
+// StopAllEvals 停止所有评估器
+func (t *AlertRule) StopAllEvals() {
+	t.ctx.Mux.Lock()
+	defer t.ctx.Mux.Unlock()
+
+	count := len(t.ctx.ContextMap)
+	if count == 0 {
+		return
+	}
+
+	logc.Infof(t.ctx.Ctx, "停止 %d 个规则评估器...", count)
+
+	// 取消所有评估任务
+	for ruleId, cancel := range t.ctx.ContextMap {
+		cancel()
+		delete(t.ctx.ContextMap, ruleId)
+	}
+
+	logc.Infof(t.ctx.Ctx, "所有规则评估器已停止")
+}