816 lines
25 KiB
Go
816 lines
25 KiB
Go
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.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() {
|
||
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)
|
||
}
|
||
}
|
||
|
||
// 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(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(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(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 发送告警通知（通过 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
|
||
}
|