package handlers import ( "encoding/json" "fmt" "time" "ezops/internal/models" "gorm.io/gorm" ) // AlertEngine 告警引擎 — 定时扫描 Agent 上报数据并触发告警 // Logan.Li type AlertEngine struct { db *gorm.DB } func NewAlertEngine(db *gorm.DB) *AlertEngine { return &AlertEngine{db: db} } // Check 每 30 秒调用一次 func (e *AlertEngine) Check() { e.checkResourceAlerts() e.checkOfflineAlerts() e.checkPortAlerts() e.autoResolveAlerts() } // 检查 CPU/内存/磁盘 阈值 func (e *AlertEngine) checkResourceAlerts() { var rules []models.AgentRule e.db.Where("enabled = ? AND metric IN ?", true, []string{"cpu", "mem", "disk"}).Find(&rules) for _, rule := range rules { // 获取目标主机最新上报 var report models.AgentReport query := e.db.Model(&models.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 []models.DiskMetric json.Unmarshal([]byte(report.DiskInfo), &disks) for _, d := range disks { if d.UsagePercent > value { value = d.UsagePercent } } } // 检查是否超阈值 if !e.compareOperator(rule.Operator, value, rule.Threshold) { continue } // 避免重复告警:如果最近 10 分钟已有同规则未恢复告警,不重复 var existing models.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 } // 创建告警 alert := models.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) // 更新规则最后触发时间 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 []models.Server e.db.Select("ip, hostname").Group("ip").Find(&servers) // 收集所有 agent_reports 中活跃上报的主机名(最近 3 分钟内有上报) threeMinAgo := time.Now().Add(-offlineThreshold) var activeHostnames []string e.db.Model(&models.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 models.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 models.AgentAlert e.db.Where("hostname = ? AND metric = ? AND resolved = ?", s.Hostname, "offline", false).First(&existing) if existing.ID == 0 { // 获取最后一次上报时间计算离线时长 var lastReport models.AgentReport e.db.Order("id DESC").First(&lastReport) offlineMinutes := 3.0 if lastReport.ID > 0 { offlineMinutes = time.Since(lastReport.ReportTime).Minutes() } alert := models.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) } } } // 检查端口消失(之前有的监听端口现在没了) func (e *AlertEngine) checkPortAlerts() { var rules []models.AgentRule e.db.Where("enabled = ? AND metric = ?", true, "port_down").Find(&rules) for _, rule := range rules { if rule.Port == 0 { continue } // 获取目标主机最新上报 var report models.AgentReport query := e.db.Model(&models.AgentReport{}) if rule.Hostname != "" { query = query.Where("hostname = ?", rule.Hostname) } query.Order("id DESC").First(&report) if report.ID == 0 { continue } // 检查端口是否在最新上报中 var ports []models.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 models.AgentAlert e.db.Where("rule_name = ? AND hostname = ? AND resolved = ?", rule.Name, report.Hostname, false).First(&existing) if existing.ID > 0 { continue } alert := models.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) now := time.Now() e.db.Model(&rule).Update("last_triggered", now) } } // 自动恢复告警(指标恢复正常) func (e *AlertEngine) autoResolveAlerts() { var alerts []models.AgentAlert e.db.Where("resolved = ?", false).Find(&alerts) for _, alert := range alerts { resolved := false switch alert.Metric { case "offline": // 主机重新上报了 var report models.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": // 获取对应规则,检查是否恢复正常 var rule models.AgentRule e.db.Where("name = ?", alert.RuleName).First(&rule) if rule.ID == 0 { continue } var report models.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 []models.DiskMetric json.Unmarshal([]byte(report.DiskInfo), &disks) for _, d := range disks { if d.UsagePercent > value { value = d.UsagePercent } } } if !e.compareOperator(rule.Operator, value, rule.Threshold) { resolved = true } case "port_down": var rule models.AgentRule e.db.Where("name = ?", alert.RuleName).First(&rule) if rule.ID > 0 { var report models.AgentReport e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report) var ports []models.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, }) } } } 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 } }