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 分钟无上报)— 内置规则,无需手动创建 func (e *AlertEngine) checkOfflineAlerts() { offlineThreshold := 3 * time.Minute // 查所有曾上报的主机 var hostnames []string e.db.Model(&models.AgentReport{}).Distinct("hostname").Pluck("hostname", &hostnames) for _, hostname := range hostnames { var report models.AgentReport e.db.Where("hostname = ?", hostname).Order("id DESC").First(&report) if report.ID == 0 { continue } isOffline := time.Since(report.ReportTime) > offlineThreshold // 检查是否有未恢复的离线告警 var existing models.AgentAlert e.db.Where("hostname = ? AND metric = ? AND resolved = ?", hostname, "offline", false).First(&existing) if isOffline && existing.ID == 0 { alert := models.AgentAlert{ RuleName: "主机离线", Hostname: hostname, Metric: "offline", Severity: "critical", Message: fmt.Sprintf("主机 %s 已超过 3 分钟无数据上报,可能已离线", hostname), Value: time.Since(report.ReportTime).Minutes(), 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 } }