package controller import ( "bytes" "context" "crypto/rand" "encoding/hex" "encoding/json" "fmt" "io" "log" "net/http" "net/url" "strings" "time" "ezops/model" ezredis "ezops/pkg/redis" "gorm.io/gorm" ) // AlertEngine 告警引擎 — 定时扫描 Agent 上报数据并触发告警 // 支持分布式锁:多副本部署时仅 Leader 执行检查,避免重复告警 type AlertEngine struct { db *gorm.DB ds *LogDSHandler // 日志数据源 handler,用于执行日志告警规则 // Logan.Li cache ezredis.Cache // 可选 Redis 缓存(分布式锁) instanceID string // 当前实例唯一标识 } const ( alertEngineLockKey = "aiops:alert_engine:lock" alertEngineLockTTL = 60 * time.Second // 锁 TTL:2x tick 间隔,保证处理时间内不释放 ) // NewAlertEngine 创建告警引擎实例 // cache 参数用于分布式锁(多副本部署)。nil 时为单实例模式,不竞争锁。 // ds 用于日志告警规则查询数据源。// Logan.Li func NewAlertEngine(db *gorm.DB, cache ezredis.Cache, ds *LogDSHandler) *AlertEngine { // 生成唯一实例 ID(随机 hex,防止多副本 PID/hostname 碰撞) idBytes := make([]byte, 8) _, _ = rand.Read(idBytes) instanceID := fmt.Sprintf("ae-%s", hex.EncodeToString(idBytes)) return &AlertEngine{ db: db, ds: ds, 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.checkMiddlewareAlerts() e.checkSNMPAlerts() e.checkSiteAlerts() e.autoResolveSiteAlerts() e.checkLogAlertRules() e.autoResolveAlerts() e.checkLogAlertRecovery() } // 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() { hostMetrics := []string{ "cpu", "cpu_iowait", "mem", "mem_available", "swap", "load1", "disk", "disk_inode", "disk_io_util", "tcp_time_wait", "tcp_close_wait", "tcp_established", "net_rx_bps", "net_tx_bps", "net_rx_errors", "fd", "service_process", } var rules []model.AgentRule e.db.Where("enabled = ? AND metric IN ?", true, hostMetrics).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 if rule.Metric == "service_process" { 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 } else { var ok bool value, ok = extractHostMetricValue(report, rule.Metric) if !ok { continue } } // 检查是否超阈值 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 分钟内有上报) // Agent 报告统一使用 UTC,避免 SQLite 时区文本比较导致误报离线。 // Logan.Li threeMinAgo := time.Now().UTC().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("host", "主机离线", 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("host", 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) } } // checkLogAlertRules 执行 type == "log" 的告警规则 // 通过数据源查询 Loki/ES,结果条数 > 0 即触发。// Logan.Li func (e *AlertEngine) checkLogAlertRules() { if e.ds == nil { return } var rules []model.AlertRule e.db.Where("enabled = ? AND type = ?", true, "log").Find(&rules) for _, rule := range rules { ds, err := loadLogDS(e.ds, rule.DatasourceID) if err != nil || ds.Type != rule.Source { log.Printf("[AlertEngine] 日志告警规则 %s 数据源无效: id=%d source=%s", rule.Name, rule.DatasourceID, rule.Source) continue } count, err := e.queryLogDataSource(ds, rule.Expression) if err != nil { log.Printf("[AlertEngine] 日志告警规则 %s 查询失败: %v", rule.Name, err) continue } // 简单策略:查询结果数量 > 0 即触发(后续可扩展为 count_over_time 阈值)// Logan.Li if count == 0 { continue } // 避免重复告警:最近 10 分钟已有同规则未恢复事件,不重复 var existing model.AlertEvent tenMinAgo := time.Now().Add(-10 * time.Minute) e.db.Where("rule_id = ? AND status = ? AND started_at > ?", rule.ID, model.AlertEventFiring, tenMinAgo).First(&existing) if existing.ID > 0 { continue } if e.isSilenced(rule.Name, "", rule.ID) { continue } now := time.Now() alert := model.AlertEvent{ RuleID: rule.ID, RuleName: rule.Name, Severity: rule.Severity, Status: string(model.AlertEventFiring), Target: fmt.Sprintf("%s/%s", ds.Type, ds.Name), Message: fmt.Sprintf("日志告警触发: %s 在 %s 中匹配到 %d 条日志", rule.Name, ds.Name, count), Value: float64(count), Threshold: 0, StartedAt: now, LastEvalAt: now, } e.db.Create(&alert) e.notifyAlertRule("host", rule.Name, alert.Target, "log", rule.Severity, alert.Message, float64(count), 0) e.db.Model(&rule).Update("last_triggered", now) } } // queryLogDataSource 根据数据源类型查询日志条数 // Logan.Li func (e *AlertEngine) queryLogDataSource(ds *model.LogDataSource, expression string) (int, error) { end := time.Now() start := end.Add(-5 * time.Minute) client := newLogHTTPClient(ds.SkipTLSVerify, 30*time.Second) switch strings.ToLower(ds.Type) { case "loki": params := url.Values{} params.Set("query", expression) params.Set("limit", "1") params.Set("direction", "backward") params.Set("start", fmt.Sprintf("%d", start.UnixNano())) params.Set("end", fmt.Sprintf("%d", end.UnixNano())) target := ds.URL + "/loki/api/v1/query_range?" + params.Encode() req, _ := http.NewRequest(http.MethodGet, target, nil) applyLogDSAuthV2(req, ds) resp, err := client.Do(req) if err != nil { return 0, err } defer resp.Body.Close() body, _ := io.ReadAll(resp.Body) if resp.StatusCode >= 400 { return 0, fmt.Errorf("HTTP %s: %s", resp.Status, string(body)) } var raw map[string]interface{} _ = json.Unmarshal(body, &raw) entries := flattenLokiStreams(raw) return len(entries), nil case "elasticsearch": searchBody := map[string]interface{}{ "size": 0, "query": map[string]interface{}{ "bool": map[string]interface{}{ "must": []interface{}{ map[string]interface{}{"query_string": map[string]interface{}{"query": defaultESQuery(expression)}}, }, }, }, } rangeCond := map[string]interface{}{ "bool": map[string]interface{}{ "should": []interface{}{ map[string]interface{}{"range": map[string]interface{}{"@timestamp": map[string]string{"gte": start.Format(time.RFC3339Nano), "lte": end.Format(time.RFC3339Nano)}}}, map[string]interface{}{"range": map[string]interface{}{"timestamp": map[string]string{"gte": start.Format(time.RFC3339Nano), "lte": end.Format(time.RFC3339Nano)}}}, }, "minimum_should_match": 1, }, } searchBody["query"].(map[string]interface{})["bool"].(map[string]interface{})["must"] = append( searchBody["query"].(map[string]interface{})["bool"].(map[string]interface{})["must"].([]interface{}), rangeCond, ) payload, _ := json.Marshal(searchBody) target := fmt.Sprintf("%s/%s/_search", ds.URL, "*") req, _ := http.NewRequest(http.MethodPost, target, bytes.NewReader(payload)) req.Header.Set("Content-Type", "application/json") applyLogDSAuthV2(req, ds) resp, err := client.Do(req) if err != nil { return 0, err } defer resp.Body.Close() body, _ := io.ReadAll(resp.Body) if resp.StatusCode >= 400 { return 0, fmt.Errorf("HTTP %s: %s", resp.Status, string(body)) } var raw map[string]interface{} _ = json.Unmarshal(body, &raw) total := esHitsTotal(raw) return int(total), nil default: return 0, fmt.Errorf("未知日志数据源类型: %s", ds.Type) } } // 自动恢复告警(指标恢复正常) 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("host", alert.RuleName, alert.Hostname, alert.Metric, alert.Message) } } } // checkLogAlertRecovery 恢复已解决的日志告警事件 // Logan.Li func (e *AlertEngine) checkLogAlertRecovery() { if e.ds == nil { return } var events []model.AlertEvent e.db.Where("status = ?", model.AlertEventFiring).Find(&events) for _, evt := range events { if evt.RuleID == 0 { continue } var rule model.AlertRule if err := e.db.First(&rule, evt.RuleID).Error; err != nil || rule.Type != "log" { continue } ds, err := loadLogDS(e.ds, rule.DatasourceID) if err != nil { continue } count, err := e.queryLogDataSource(ds, rule.Expression) if err != nil { continue } // 连续两个周期无日志匹配则恢复(简化策略)// Logan.Li if count > 0 { continue } now := time.Now() e.db.Model(&evt).Updates(map[string]interface{}{ "status": model.AlertEventResolved, "resolved_at": now, }) e.notifyRecovery("host", rule.Name, evt.Target, "log", evt.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 发送主机监控告警通知 // Logan.Li func (e *AlertEngine) notifyAlert(rule *model.AgentRule, hostname string, alert model.AgentAlert) { e.notifyAlertRule("host", rule.Name, hostname, alert.Metric, alert.Severity, alert.Message, alert.Value, alert.Threshold) } // notifyAlertRule 按告警集合策略发送通知(无策略时回退全局渠道)// Logan.Li func (e *AlertEngine) notifyAlertRule(category, ruleName, hostname, metric, severity, message string, value, threshold float64) { dispatchCategoryAlert(e.db, category, ruleName, hostname, metric, severity, message, value, threshold) } // notifyRecovery 按告警集合策略发送恢复通知 // Logan.Li func (e *AlertEngine) notifyRecovery(category, ruleName, hostname, metric, message string) { dispatchCategoryRecovery(e.db, category, ruleName, hostname, metric, message) } // 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 }