feat(alert): A2 告警引擎分布式安全 — Redis 分布式锁(SetNX)

- Cache 接口新增 SetNX 方法(go-redis 原子 / redka check-then-set 单副本安全)
- AlertEngine 改造为支持分布式锁: tryAcquireLock/releaseLock/GetEngineStatus
- 多副本部署时仅 Leader 执行告警检查, 60s TTL 防死锁, fail-open 容错
- 新增 GET /alerts/engine/status 监控端点(is_leader/instance_id/lock_holder)
- 修复 GrafanaDashboard.tsx Panel.options 类型不匹配导致构建中断
- 同步未提交的前端类型安全迁移(140+ 文件)
- ROADMAP/TDD_PLAN/PROJECT_EVALUATION 三文档同步更新

验证: go build 零错误, Docker 部署 healthy, 引擎状态端点正常,
锁竞争测试通过(非 Leader 跳过检查), 5/5 核心 API 回归通过
This commit is contained in:
attacker committed 2026-06-26 21:13:15 +08:00
1 parent 5ca0e6d83a
commit c68aef4444
150 files changed
+1038 -690

No files matched your search

+90 -5
View File
@@ -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