merge: 合并远程 A2 告警引擎分布式安全更新，保留本地日志中心等最新改动

backend/.aiops.args

@@ -0,0 +1,4 @@
+--mode
+local
+--port
+8080

backend/controller/alert_engine.go

@@ -1,32 +1,117 @@
 package controller
 
 import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"log"
 	"time"
 
 	"ezops/model"
+	ezredis "ezops/pkg/redis"
 
 	"gorm.io/gorm"
 )
 
-// AlertEngine 告警引擎 — 定时扫描 Agent 上报数据并触发告警 // Logan.Li
+// AlertEngine 告警引擎 — 定时扫描 Agent 上报数据并触发告警
+// 支持分布式锁：多副本部署时仅 Leader 执行检查，避免重复告警
 type AlertEngine struct {
-	db *gorm.DB
+	db         *gorm.DB
+	cache      ezredis.Cache // 可选 Redis 缓存（分布式锁）
+	instanceID string        // 当前实例唯一标识
 }
 
-func NewAlertEngine(db *gorm.DB) *AlertEngine {
-	return &AlertEngine{db: db}
+const (
+	alertEngineLockKey = "aiops:alert_engine:lock"
+	alertEngineLockTTL = 60 * time.Second // 锁 TTL：2x tick 间隔，保证处理时间内不释放
+)
+
+// NewAlertEngine 创建告警引擎实例
+// cache 参数用于分布式锁（多副本部署）。nil 时为单实例模式，不竞争锁。
+func NewAlertEngine(db *gorm.DB, cache ezredis.Cache) *AlertEngine {
+	// 生成唯一实例 ID（随机 hex，防止多副本 PID/hostname 碰撞）
+	idBytes := make([]byte, 8)
+	_, _ = rand.Read(idBytes)
+	instanceID := fmt.Sprintf("ae-%s", hex.EncodeToString(idBytes))
+	return &AlertEngine{
+		db:         db,
+		cache:      cache,
+		instanceID: instanceID,
+	}
 }
 
-// Check 每 30 秒调用一次
+// Check 每 30 秒调用一次。多副本模式下通过分布式锁保证仅 Leader 执行。
+// 无 cache（单实例/缓存初始化失败）→ 直接执行
+// 有 cache（Docker/K8s 多副本）→ SetNX 原子抢锁
+// Redis 故障 → fail-open（继续执行，避免告警遗漏）
 func (e *AlertEngine) Check() {
+	if !e.tryAcquireLock() {
+		log.Printf("[AlertEngine] 非 Leader 副本（%s），跳过本轮检查", e.instanceID)
+		return
+	}
+	defer e.releaseLock()
+
 	e.checkResourceAlerts()
 	e.checkOfflineAlerts()
 	e.checkPortAlerts()
 	e.autoResolveAlerts()
 }
 
+// tryAcquireLock 尝试获取分布式锁，返回是否获得 Leader 角色
+func (e *AlertEngine) tryAcquireLock() bool {
+	if e.cache == nil {
+		return true // 单实例模式
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	acquired, err := e.cache.SetNX(ctx, alertEngineLockKey, e.instanceID, alertEngineLockTTL)
+	if err != nil {
+		log.Printf("[AlertEngine] 获取分布式锁失败（%s），fail-open 继续: %v", e.instanceID, err)
+		return true
+	}
+	return acquired
+}
+
+// releaseLock 释放分布式锁（仅当当前实例持有锁时才释放，防止误删其他副本的锁）
+func (e *AlertEngine) releaseLock() {
+	if e.cache == nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	val, err := e.cache.Get(ctx, alertEngineLockKey)
+	if err != nil {
+		return
+	}
+	if val == e.instanceID {
+		_ = e.cache.Del(ctx, alertEngineLockKey)
+	}
+}
+
+// GetEngineStatus 返回告警引擎运行状态（用于运维监控）
+func (e *AlertEngine) GetEngineStatus() map[string]interface{} {
+	isLeader := true
+	lockHolder := ""
+	if e.cache != nil {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		if val, err := e.cache.Get(ctx, alertEngineLockKey); err == nil {
+			lockHolder = val
+			isLeader = lockHolder == "" || lockHolder == e.instanceID
+		}
+	}
+	return map[string]interface{}{
+		"is_leader":     isLeader,
+		"instance_id":   e.instanceID,
+		"lock_holder":   lockHolder,
+		"distributed":   e.cache != nil,
+		"tick_interval": 30,
+		"lock_ttl_sec":  int(alertEngineLockTTL / time.Second),
+	}
+}
+
 // 检查 CPU/内存/磁盘/进程/TCP连接/端口 阈值
 func (e *AlertEngine) checkResourceAlerts() {
 	var rules []model.AgentRule

backend/pkg/redis/go_redis.go

@@ -49,6 +49,11 @@ func (r *goRedisCache) Set(ctx context.Context, key string, value string, expira
 	return r.client.Set(ctx, key, value, expiration).Err()
 }
 
+// SetNX 原子 SET if Not eXists — go-redis 原生支持，用于分布式锁
+func (r *goRedisCache) SetNX(ctx context.Context, key string, value string, expiration time.Duration) (bool, error) {
+	return r.client.SetNX(ctx, key, value, expiration).Result()
+}
+
 func (r *goRedisCache) Del(ctx context.Context, keys ...string) error {
 	return r.client.Del(ctx, keys...).Err()
 }

backend/pkg/redis/redis.go

@@ -51,6 +51,8 @@ type Cache interface {
 	Capabilities() Capabilities
 	Get(ctx context.Context, key string) (string, error)
 	Set(ctx context.Context, key string, value string, expiration time.Duration) error
+	// SetNX 原子「Set if Not eXists」— 用于分布式锁（多副本 Leader 选举）
+	SetNX(ctx context.Context, key string, value string, expiration time.Duration) (bool, error)
 	Del(ctx context.Context, keys ...string) error
 	Exists(ctx context.Context, keys ...string) (int64, error)
 	Expire(ctx context.Context, key string, expiration time.Duration) error

backend/pkg/redis/redka.go

@@ -65,6 +65,27 @@ func (r *redkaCache) Set(ctx context.Context, key string, value string, expirati
 	return nil
 }
 
+// SetNX check-then-set — Redka v0.2.0 无原子 SetNX，使用 Exists+Set 模拟。
+// 本地/redka 模式为单副本运行，不存在并发竞争，非原子实现安全。
+func (r *redkaCache) SetNX(ctx context.Context, key string, value string, expiration time.Duration) (bool, error) {
+	ok, err := r.db.Key().Exists(key)
+	if err != nil {
+		return false, err
+	}
+	if ok {
+		return false, nil
+	}
+	if err := r.db.Str().Set(key, value); err != nil {
+		return false, err
+	}
+	if expiration > 0 {
+		if _, err := r.db.Key().Expire(key, expiration); err != nil {
+			return false, err
+		}
+	}
+	return true, nil
+}
+
 func (r *redkaCache) Del(ctx context.Context, keys ...string) error {
 	for _, key := range keys {
 		if _, err := r.db.Key().Delete(key); err != nil {

backend/router/routes.go

@@ -496,6 +496,9 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c
 		kubernetesRouter.GET("/compare", k8sHandler.GetMultiClusterComparison)
 	}
 
+	// 告警引擎（分布式锁 — 多副本部署时仅 Leader 执行检查）
+	alertEngine := controller.NewAlertEngine(db, redisCache)
+
 	// 告警路由
 	alertHandler := controller.NewAlertHandler(db)
 	alerts := authorized.Group("/alerts")
@@ -532,6 +535,10 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c
 		alerts.PUT("/silences/:id", alertHandler.UpdateSilence)
 		alerts.DELETE("/silences/:id", alertHandler.DeleteSilence)
 		alerts.POST("/silences/:id/toggle", alertHandler.ToggleSilence)
+		// 告警引擎状态 — 查看当前 Leader 副本和分布式锁状态
+		alerts.GET("/engine/status", func(c *gin.Context) {
+			c.JSON(200, model.Response{Code: 200, Data: alertEngine.GetEngineStatus()})
+		})
 	}
 
 	// 流水线路由
@@ -943,8 +950,7 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c
 		siteGroup.GET("/stats", siteHandler.GetSiteStats)
 	}
 
-	// 启动告警引擎（每 30 秒检查一次）
-	alertEngine := controller.NewAlertEngine(db)
+	// 启动告警引擎（每 30 秒检查一次；alertEngine 在告警路由上方创建）
 	go func() {
 		ticker := time.NewTicker(30 * time.Second)
 		defer ticker.Stop()