package controller import ( "context" "crypto/rand" "encoding/hex" "encoding/json" "fmt" "log" "time" "ezops/model" ezredis "ezops/pkg/redis" "gorm.io/gorm" ) // AlertEngine 告警引擎 — 定时扫描 Agent 上报数据并触发告警 // 支持分布式锁:多副本部署时仅 Leader 执行检查,避免重复告警 type AlertEngine struct { db *gorm.DB cache ezredis.Cache // 可选 Redis 缓存(分布式锁) instanceID string // 当前实例唯一标识 } 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 秒调用一次。多副本模式下通过分布式锁保证仅 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 e.db.Where("enabled = ? AND metric IN ?", true, []string{"cpu", "mem", "disk", "service_process", "tcp_time_wait", "tcp_close_wait", "fd"}).Find(&rules) for _, rule := range rules { // 获取目标主机最新上报 var report model.AgentReport query := e.db.Model(&model.AgentReport{}) if rule.Hostname != "" { query = query.Where("hostname = ?", rule.Hostname) } query.Order("id DESC").First(&report) if report.ID == 0 { continue } var value float64 switch rule.Metric { case "cpu": value = report.CPUUsage case "mem": value = report.MemUsage case "disk": var disks []model.DiskMetric json.Unmarshal([]byte(report.DiskInfo), &disks) for _, d := range disks { if d.UsagePercent > value { value = d.UsagePercent } } case "tcp_time_wait": var netStat model.NetworkStatModel json.Unmarshal([]byte(report.NetInfo), &netStat) value = float64(netStat.TCPTimeWait) case "tcp_close_wait": var netStat model.NetworkStatModel json.Unmarshal([]byte(report.NetInfo), &netStat) value = float64(netStat.TCPCloseWait) case "service_process": // 进程不存在告警 — 检查 process_top_info 中是否包含指定进程名 var processes []model.ProcessMetric json.Unmarshal([]byte(report.ProcessTopInfo), &processes) found := false for _, p := range processes { if p.Name == rule.Hostname || rule.Hostname == "" { found = true break } } if found { continue // 进程存在,不触发 } value = 0 // 进程不存在时触发 case "fd": var fdStat model.FDMetric json.Unmarshal([]byte(report.FDInfo), &fdStat) value = fdStat.UsagePct } // 检查是否超阈值 if !e.compareOperator(rule.Operator, value, rule.Threshold) { continue } // 避免重复告警:如果最近 10 分钟已有同规则未恢复告警,不重复 var existing model.AgentAlert tenMinAgo := time.Now().Add(-10 * time.Minute) e.db.Where("rule_name = ? AND hostname = ? AND resolved = ? AND created_at > ?", rule.Name, report.Hostname, false, tenMinAgo).First(&existing) if existing.ID > 0 { continue } // 检查静默规则 — 匹配则不创建告警、不发送通知 if e.isSilenced(rule.Name, report.Hostname, rule.ID) { continue } // 创建告警 alert := model.AgentAlert{ RuleName: rule.Name, Hostname: report.Hostname, Metric: rule.Metric, Severity: "warning", Message: fmt.Sprintf("%s: %s %.1f%% %s %.1f%%", report.Hostname, rule.Metric, value, rule.Operator, rule.Threshold), Value: value, Threshold: rule.Threshold, } if value > rule.Threshold*1.5 { alert.Severity = "critical" } e.db.Create(&alert) // 发送告警通知 e.notifyAlert(&rule, report.Hostname, alert) // 更新规则最后触发时间 now := time.Now() e.db.Model(&rule).Update("last_triggered", now) } } // 检查主机离线(超过 3 分钟无上报)— 内置规则,无需手动创建。 // 以 servers 表为中心:遍历所有服务器,检查其 IP 上是否有 Agent 在最近 3 分钟内上报。 // 如果同一 IP 下有多个 Agent hostname,只要有一个在活跃上报,该服务器就算在线。 func (e *AlertEngine) checkOfflineAlerts() { offlineThreshold := 3 * time.Minute // 获取所有已登记的主机(去重 IP) var servers []model.Server e.db.Select("ip, hostname").Group("ip").Find(&servers) // 收集所有 agent_reports 中活跃上报的主机名(最近 3 分钟内有上报) threeMinAgo := time.Now().Add(-offlineThreshold) var activeHostnames []string e.db.Model(&model.AgentReport{}). Where("report_time >= ?", threeMinAgo). Distinct("hostname"). Pluck("hostname", &activeHostnames) // 对于每个已登记的服务器,检查是否有活跃 Agent for _, s := range servers { if s.IP == "" { continue } // 检查该 IP 对应的服务器是否有活跃 Agent 上报 hasActiveAgent := false for _, h := range activeHostnames { // 通过 hostname 精确匹配或 servers 表 IP 匹配 if h == s.Hostname { hasActiveAgent = true break } // 检查该 hostname 的最新上报——其源 IP 是否匹配此 server IP // (通过 servers 表中查找 hostname 对应的 IP) var matchedServer model.Server e.db.Where("hostname = ?", h).First(&matchedServer) if matchedServer.ID > 0 && matchedServer.IP == s.IP { hasActiveAgent = true break } } if hasActiveAgent { continue // 有活跃 Agent,在线 } // 该服务器离线:检查是否已有未恢复的离线告警 var existing model.AgentAlert e.db.Where("hostname = ? AND metric = ? AND resolved = ?", s.Hostname, "offline", false).First(&existing) if existing.ID == 0 { // 获取最后一次上报时间计算离线时长 var lastReport model.AgentReport e.db.Where("hostname = ?", s.Hostname).Order("id DESC").First(&lastReport) offlineMinutes := 3.0 var lastReportTime time.Time if lastReport.ID > 0 { offlineMinutes = time.Since(lastReport.ReportTime).Minutes() lastReportTime = lastReport.ReportTime } // 检查静默规则 if e.isSilenced("主机离线", s.Hostname, 0) { continue } alert := model.AgentAlert{ RuleName: "主机离线", Hostname: s.Hostname, Metric: "offline", Severity: "critical", Message: fmt.Sprintf("主机 %s (%s) 已超过 3 分钟无数据上报,可能已离线", s.Hostname, s.IP), Value: offlineMinutes, Threshold: 3, } e.db.Create(&alert) // 发送离线告警通知 e.notifyAlertRule("主机离线", s.Hostname, "offline", "critical", alert.Message, offlineMinutes, 3) // 记录离线上报时间用于恢复通知 if !lastReportTime.IsZero() { e.db.Model(&alert).Update("created_at", time.Now()) } } } } // 检查端口消失(之前有的监听端口现在没了) func (e *AlertEngine) checkPortAlerts() { var rules []model.AgentRule e.db.Where("enabled = ? AND metric = ?", true, "port_down").Find(&rules) for _, rule := range rules { if rule.Port == 0 { continue } // 获取目标主机最新上报 var report model.AgentReport query := e.db.Model(&model.AgentReport{}) if rule.Hostname != "" { query = query.Where("hostname = ?", rule.Hostname) } query.Order("id DESC").First(&report) if report.ID == 0 { continue } // 检查端口是否在最新上报中 var ports []model.PortMetric json.Unmarshal([]byte(report.PortInfo), &ports) portUp := false for _, p := range ports { if p.Port == rule.Port { portUp = true break } } if portUp { continue // 端口正常 } // 端口消失了,检查是否已有未恢复告警 var existing model.AgentAlert e.db.Where("rule_name = ? AND hostname = ? AND resolved = ?", rule.Name, report.Hostname, false).First(&existing) if existing.ID > 0 { continue } // 检查静默规则 if e.isSilenced(rule.Name, report.Hostname, rule.ID) { continue } alert := model.AgentAlert{ RuleName: rule.Name, Hostname: report.Hostname, Metric: "port_down", Severity: "critical", Message: fmt.Sprintf("主机 %s 端口 %d 服务疑似下线", report.Hostname, rule.Port), Value: float64(rule.Port), Threshold: float64(rule.Port), } e.db.Create(&alert) // 发送端口Down告警通知 e.notifyAlertRule(rule.Name, report.Hostname, "port_down", "critical", alert.Message, float64(rule.Port), float64(rule.Port)) now := time.Now() e.db.Model(&rule).Update("last_triggered", now) } } // 自动恢复告警(指标恢复正常) func (e *AlertEngine) autoResolveAlerts() { var alerts []model.AgentAlert e.db.Where("resolved = ?", false).Find(&alerts) for _, alert := range alerts { resolved := false switch alert.Metric { case "offline": // 主机重新上报了 var report model.AgentReport e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report) if report.ID > 0 && time.Since(report.ReportTime) < 3*time.Minute { resolved = true } case "cpu", "mem", "disk", "tcp_time_wait", "tcp_close_wait", "fd": // 获取对应规则,检查是否恢复正常 var rule model.AgentRule e.db.Where("name = ?", alert.RuleName).First(&rule) if rule.ID == 0 { continue } var report model.AgentReport if rule.Hostname != "" { e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report) } else { e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report) } if report.ID == 0 { continue } var value float64 switch alert.Metric { case "cpu": value = report.CPUUsage case "mem": value = report.MemUsage case "disk": var disks []model.DiskMetric json.Unmarshal([]byte(report.DiskInfo), &disks) for _, d := range disks { if d.UsagePercent > value { value = d.UsagePercent } } case "tcp_time_wait": var netStat model.NetworkStatModel json.Unmarshal([]byte(report.NetInfo), &netStat) value = float64(netStat.TCPTimeWait) case "tcp_close_wait": var netStat model.NetworkStatModel json.Unmarshal([]byte(report.NetInfo), &netStat) value = float64(netStat.TCPCloseWait) case "fd": var fdStat model.FDMetric json.Unmarshal([]byte(report.FDInfo), &fdStat) value = fdStat.UsagePct } if !e.compareOperator(rule.Operator, value, rule.Threshold) { resolved = true } case "port_down": var rule model.AgentRule e.db.Where("name = ?", alert.RuleName).First(&rule) if rule.ID > 0 { var report model.AgentReport e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report) var ports []model.PortMetric json.Unmarshal([]byte(report.PortInfo), &ports) for _, p := range ports { if p.Port == rule.Port { resolved = true break } } } } if resolved { now := time.Now() e.db.Model(&alert).Updates(map[string]interface{}{ "resolved": true, "resolved_at": now, }) // 发送恢复通知 e.notifyRecovery(alert.RuleName, alert.Hostname, alert.Metric, alert.Message) } } } func (e *AlertEngine) compareOperator(op string, val, threshold float64) bool { switch op { case ">": return val > threshold case ">=": return val >= threshold case "<": return val < threshold case "<=": return val <= threshold case "==": return val == threshold default: return val > threshold } } // notifyAlert 发送告警通知(通过 AgentRule 对应的 notify_channels) func (e *AlertEngine) notifyAlert(rule *model.AgentRule, hostname string, alert model.AgentAlert) { e.notifyAlertRule(rule.Name, hostname, alert.Metric, alert.Severity, alert.Message, alert.Value, alert.Threshold) } // notifyAlertRule 按规则发送告警通知到所有启用的飞书渠道 func (e *AlertEngine) notifyAlertRule(ruleName, hostname, metric, severity, message string, value, threshold float64) { var channels []model.NotifyChannel e.db.Where("enabled = ? AND type = ?", true, "feishu").Find(&channels) for _, ch := range channels { title := fmt.Sprintf("[%s] %s — %s (%s)", severity, ruleName, hostname, metric) content := fmt.Sprintf( "**规则名称:** %s\\n**目标主机:** %s\\n**指标类型:** %s\\n**告警级别:** %s\\n**当前值:** %.2f\\n**阈值:** %.2f\\n**详细信息:** %s\\n**触发时间:** %s", ruleName, hostname, metric, severity, value, threshold, message, time.Now().Format("2006-01-02 15:04:05"), ) if err := sendFeishuAlert(ch.Config, title, content, severity); err != nil { fmt.Printf("[AlertEngine] 飞书通知失败 (渠道:%s): %v\n", ch.Name, err) } } } // notifyRecovery 发送恢复通知到所有启用的飞书渠道 func (e *AlertEngine) notifyRecovery(ruleName, hostname, metric, message string) { var channels []model.NotifyChannel e.db.Where("enabled = ? AND type = ?", true, "feishu").Find(&channels) for _, ch := range channels { title := fmt.Sprintf("%s — 已恢复正常", ruleName) content := fmt.Sprintf("**告警已恢复**\\\\n\\\\n主机:%s\\\\n指标:%s\\\\n详情:%s\\\\n恢复时间:%s", hostname, metric, message, time.Now().Format("2006-01-02 15:04:05")) if err := sendFeishuRecovery(ch.Config, title, content); err != nil { fmt.Printf("[AlertEngine] 飞书恢复通知失败 (渠道:%s): %v\n", ch.Name, err) } } } // isSilenced 检查给定告警是否被静默规则匹配 // 匹配逻辑:status=1 且 当前时间在静默窗口内 // 匹配维度:rule_name(精确匹配)、rule_id(ID 匹配)、hostname(通过 target_ids 中的 server IP 间接匹配) // silence_type:onetime=一次性窗口,daily=每天指定时段,weekly=每周指定星期+时段 func (e *AlertEngine) isSilenced(ruleName, hostname string, ruleID uint) bool { var silences []model.AlertSilence e.db.Where("status = ?", 1).Find(&silences) now := time.Now() for _, s := range silences { // 检查规则名匹配 — 通过 rule_ids JSON 数组 if s.RuleIDs != "" && s.RuleIDs != "null" { var ruleIDs []uint json.Unmarshal([]byte(s.RuleIDs), &ruleIDs) matched := false for _, rid := range ruleIDs { if rid == ruleID { matched = true break } } if !matched { continue // 该静默规则不匹配此 rule_id } } // 检查目标主机匹配 — 通过 target_ids(server IDs)间接匹配 if s.TargetIDs != "" && s.TargetIDs != "null" { var targetIDs []uint json.Unmarshal([]byte(s.TargetIDs), &targetIDs) if len(targetIDs) > 0 { matched := false for _, tid := range targetIDs { var server model.Server e.db.Select("hostname").First(&server, tid) if server.Hostname == hostname { matched = true break } } if !matched { continue } } } // 检查时间窗口 switch s.SilenceType { case model.AlertSilenceOneTime: // 一次性:start_time <= now <= end_time if now.After(s.StartTime) && now.Before(s.EndTime) { return true } case model.AlertSilenceDaily: // 每天:检查当前时分是否在窗口内 nowMinutes := now.Hour()*60 + now.Minute() startMinutes := s.StartTime.Hour()*60 + s.StartTime.Minute() endMinutes := s.EndTime.Hour()*60 + s.EndTime.Minute() if startMinutes <= endMinutes { if nowMinutes >= startMinutes && nowMinutes <= endMinutes { return true } } else { // 跨天(如 23:00-06:00) if nowMinutes >= startMinutes || nowMinutes <= endMinutes { return true } } case model.AlertSilenceWeekly: // 每周:检查当前星期是否在 weekdays 列表中,且时分在窗口内 if s.Weekdays != "" && s.Weekdays != "null" { var weekdays []int json.Unmarshal([]byte(s.Weekdays), &weekdays) weekdayMatched := false todayWeekday := int(now.Weekday()) for _, wd := range weekdays { if wd == todayWeekday { weekdayMatched = true break } } if !weekdayMatched { continue } } nowMinutes := now.Hour()*60 + now.Minute() startMinutes := s.StartTime.Hour()*60 + s.StartTime.Minute() endMinutes := s.EndTime.Hour()*60 + s.EndTime.Minute() if startMinutes <= endMinutes { if nowMinutes >= startMinutes && nowMinutes <= endMinutes { return true } } else { if nowMinutes >= startMinutes || nowMinutes <= endMinutes { return true } } default: // 默认按 onetime 处理 if now.After(s.StartTime) && now.Before(s.EndTime) { return true } } } return false }