388 lines
11 KiB
Go
388 lines
11 KiB
Go
package controller
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"time"
|
||
|
||
"ezops/model"
|
||
"ezops/pkg/siteprobe"
|
||
)
|
||
|
||
// checkMiddlewareAlerts 中间件指标告警 // Logan.Li
|
||
func (e *AlertEngine) checkMiddlewareAlerts() {
|
||
var rules []model.MonitorAlertRule
|
||
e.db.Where("enabled = ? AND category = ?", true, model.MonitorAlertMiddleware).Find(&rules)
|
||
if len(rules) == 0 {
|
||
return
|
||
}
|
||
|
||
var targets []model.MonitorTarget
|
||
e.db.Where("type = ?", model.TargetTypeMiddleware).Find(&targets)
|
||
targetMap := make(map[uint]model.MonitorTarget)
|
||
for _, t := range targets {
|
||
targetMap[t.ID] = t
|
||
}
|
||
|
||
for _, rule := range rules {
|
||
checkTargets := targets
|
||
if rule.TargetID > 0 {
|
||
if t, ok := targetMap[rule.TargetID]; ok {
|
||
checkTargets = []model.MonitorTarget{t}
|
||
} else {
|
||
continue
|
||
}
|
||
}
|
||
for _, target := range checkTargets {
|
||
e.evalMiddlewareRule(rule, target)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (e *AlertEngine) evalMiddlewareRule(rule model.MonitorAlertRule, target model.MonitorTarget) {
|
||
if rule.Metric == "middleware_offline" {
|
||
if target.Status == model.TargetStatusOffline || target.Status == model.TargetStatusUnknown {
|
||
e.fireMiddlewareAlert(rule, target, 1, 1, "中间件不可达或采集失败")
|
||
}
|
||
return
|
||
}
|
||
|
||
value, ok := e.getLatestMetricValue(target.ID, rule.Metric)
|
||
if !ok {
|
||
return
|
||
}
|
||
if !e.compareOperator(rule.Operator, value, rule.Threshold) {
|
||
return
|
||
}
|
||
msg := fmt.Sprintf("%s %s %.2f %s %.2f", target.Name, rule.Metric, value, rule.Operator, rule.Threshold)
|
||
e.fireMiddlewareAlert(rule, target, value, rule.Threshold, msg)
|
||
}
|
||
|
||
func (e *AlertEngine) getLatestMetricValue(targetID uint, metric string) (float64, bool) {
|
||
var hist model.MetricHistory
|
||
e.db.Where("target_id = ? AND metric_name = ?", targetID, metric).
|
||
Order("timestamp DESC").First(&hist)
|
||
if hist.ID == 0 {
|
||
return 0, false
|
||
}
|
||
return hist.Value, true
|
||
}
|
||
|
||
func (e *AlertEngine) fireMiddlewareAlert(rule model.MonitorAlertRule, target model.MonitorTarget, value, threshold float64, msg string) {
|
||
tenMinAgo := time.Now().Add(-10 * time.Minute)
|
||
var existing model.AgentAlert
|
||
e.db.Where("rule_name = ? AND hostname = ? AND resolved = ? AND created_at > ?",
|
||
rule.Name, target.Name, false, tenMinAgo).First(&existing)
|
||
if existing.ID > 0 {
|
||
return
|
||
}
|
||
if e.isSilenced(rule.Name, target.Name, rule.ID) {
|
||
return
|
||
}
|
||
sev := rule.Severity
|
||
if sev == "" {
|
||
sev = "warning"
|
||
}
|
||
alert := model.AgentAlert{
|
||
RuleName: rule.Name, Hostname: target.Name, Metric: rule.Metric,
|
||
Severity: sev, Message: msg, Value: value, Threshold: threshold,
|
||
}
|
||
e.db.Create(&alert)
|
||
e.notifyAlertRule("middleware", rule.Name, target.Name, rule.Metric, sev, msg, value, threshold)
|
||
now := time.Now()
|
||
e.db.Model(&rule).Update("last_triggered", now)
|
||
}
|
||
|
||
// checkSiteAlerts 站点拨测告警 // Logan.Li
|
||
func (e *AlertEngine) checkSiteAlerts() {
|
||
var rules []model.MonitorAlertRule
|
||
e.db.Where("enabled = ? AND category = ?", true, model.MonitorAlertSite).Find(&rules)
|
||
if len(rules) == 0 {
|
||
return
|
||
}
|
||
|
||
var tasks []model.SiteTask
|
||
e.db.Find(&tasks)
|
||
taskMap := make(map[uint]model.SiteTask)
|
||
for _, t := range tasks {
|
||
taskMap[t.ID] = t
|
||
}
|
||
|
||
for _, rule := range rules {
|
||
checkTasks := tasks
|
||
if rule.TargetID > 0 {
|
||
if t, ok := taskMap[rule.TargetID]; ok {
|
||
checkTasks = []model.SiteTask{t}
|
||
} else {
|
||
continue
|
||
}
|
||
}
|
||
for _, task := range checkTasks {
|
||
if task.Status != model.SiteTaskStatusRunning {
|
||
continue
|
||
}
|
||
e.evalSiteRule(rule, task)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (e *AlertEngine) evalSiteRule(rule model.MonitorAlertRule, task model.SiteTask) {
|
||
var value float64
|
||
var triggered bool
|
||
var msg string
|
||
|
||
switch rule.Metric {
|
||
case "probe_fail":
|
||
triggered = task.LastResult == "fail" || task.LastResult == "timeout"
|
||
value = 1
|
||
if triggered {
|
||
msg = fmt.Sprintf("拨测任务 %s 失败: %s", task.Name, task.LastResult)
|
||
}
|
||
case "probe_timeout":
|
||
triggered = task.LastResult == "timeout"
|
||
value = 1
|
||
if triggered {
|
||
msg = fmt.Sprintf("拨测任务 %s 超时", task.Name)
|
||
}
|
||
case "response_time":
|
||
value = task.AvgResponse
|
||
triggered = e.compareOperator(rule.Operator, value, rule.Threshold)
|
||
if triggered {
|
||
msg = fmt.Sprintf("拨测 %s 延迟 %.0fms 超过阈值 %.0fms", task.Name, value, rule.Threshold)
|
||
}
|
||
case "availability":
|
||
value = task.SuccessRate
|
||
triggered = e.compareOperator(rule.Operator, value, rule.Threshold)
|
||
if triggered {
|
||
msg = fmt.Sprintf("拨测 %s 可用率 %.1f%% 低于阈值 %.1f%%", task.Name, value, rule.Threshold)
|
||
}
|
||
case "http_status_error":
|
||
triggered = task.LastResult == "fail"
|
||
value = 1
|
||
if triggered {
|
||
msg = fmt.Sprintf("拨测 %s HTTP 状态码异常", task.Name)
|
||
}
|
||
case "ssl_expiry_days":
|
||
if !siteprobe.IsSSLProbeTask(task) {
|
||
return
|
||
}
|
||
var result model.SiteResult
|
||
e.db.Where("task_id = ? AND ssl_cert_expiry IS NOT NULL", task.ID).
|
||
Order("checked_at DESC").First(&result)
|
||
if result.ID == 0 {
|
||
return
|
||
}
|
||
value = float64(result.SSLDaysRemaining)
|
||
triggered = e.compareOperator(rule.Operator, value, rule.Threshold)
|
||
if triggered {
|
||
expiryStr := ""
|
||
if result.SSLCertExpiry != nil {
|
||
expiryStr = result.SSLCertExpiry.Format("2006-01-02")
|
||
}
|
||
if value < 0 {
|
||
msg = fmt.Sprintf("SSL 拨测 %s 证书已过期", task.Name)
|
||
} else if expiryStr != "" {
|
||
msg = fmt.Sprintf("SSL 拨测 %s 证书剩余 %.0f 天（到期 %s）", task.Name, value, expiryStr)
|
||
} else {
|
||
msg = fmt.Sprintf("SSL 拨测 %s 证书剩余 %.0f 天", task.Name, value)
|
||
}
|
||
}
|
||
default:
|
||
return
|
||
}
|
||
|
||
if !triggered {
|
||
e.resolveActiveSiteAlerts(task.ID, rule.Metric)
|
||
return
|
||
}
|
||
|
||
tenMinAgo := time.Now().Add(-10 * time.Minute)
|
||
var existing model.SiteAlert
|
||
e.db.Where("task_id = ? AND title = ? AND status = ? AND started_at > ?",
|
||
task.ID, rule.Name, model.SiteAlertStatusActive, tenMinAgo).First(&existing)
|
||
if existing.ID > 0 {
|
||
return
|
||
}
|
||
|
||
sev := model.SiteAlertWarning
|
||
if rule.Severity == "critical" || (rule.Metric == "ssl_expiry_days" && value <= 7) {
|
||
sev = model.SiteAlertCritical
|
||
}
|
||
now := time.Now()
|
||
siteAlert := model.SiteAlert{
|
||
TaskID: task.ID, Severity: sev, Status: model.SiteAlertStatusActive,
|
||
Title: rule.Name, Message: msg, TriggerType: rule.Metric,
|
||
Threshold: rule.Threshold, ResponseTime: task.AvgResponse,
|
||
StartedAt: now,
|
||
}
|
||
e.db.Create(&siteAlert)
|
||
e.notifyAlertRule("site", rule.Name, task.Name, rule.Metric, string(sev), msg, value, rule.Threshold)
|
||
nt := time.Now()
|
||
e.db.Model(&rule).Update("last_triggered", nt)
|
||
}
|
||
|
||
// extractHostMetricValue 从 Agent 上报提取扩展指标值 // Logan.Li
|
||
func extractHostMetricValue(report model.AgentReport, metric string) (float64, bool) {
|
||
switch metric {
|
||
case "cpu":
|
||
return report.CPUUsage, true
|
||
case "cpu_iowait":
|
||
return report.CPUIOwait, true
|
||
case "mem":
|
||
return report.MemUsage, true
|
||
case "mem_available":
|
||
return report.MemAvailable, true
|
||
case "load1":
|
||
return report.Load1, true
|
||
case "disk":
|
||
var disks []model.DiskMetric
|
||
json.Unmarshal([]byte(report.DiskInfo), &disks)
|
||
max := 0.0
|
||
for _, d := range disks {
|
||
if d.UsagePercent > max {
|
||
max = d.UsagePercent
|
||
}
|
||
}
|
||
return max, true
|
||
case "disk_inode":
|
||
var disks []model.DiskMetric
|
||
json.Unmarshal([]byte(report.DiskInfo), &disks)
|
||
max := 0.0
|
||
for _, d := range disks {
|
||
if d.InodeUsagePct > max {
|
||
max = d.InodeUsagePct
|
||
}
|
||
}
|
||
return max, true
|
||
case "disk_io_util":
|
||
var dio model.DiskIOStatModel
|
||
json.Unmarshal([]byte(report.DiskIOInfo), &dio)
|
||
return dio.Util, true
|
||
case "swap":
|
||
var swap model.SwapMetric
|
||
json.Unmarshal([]byte(report.SwapInfo), &swap)
|
||
return swap.UsagePercent, true
|
||
case "tcp_time_wait":
|
||
var netStat model.NetworkStatModel
|
||
json.Unmarshal([]byte(report.NetInfo), &netStat)
|
||
return float64(netStat.TCPTimeWait), true
|
||
case "tcp_close_wait":
|
||
var netStat model.NetworkStatModel
|
||
json.Unmarshal([]byte(report.NetInfo), &netStat)
|
||
return float64(netStat.TCPCloseWait), true
|
||
case "tcp_established":
|
||
var netStat model.NetworkStatModel
|
||
json.Unmarshal([]byte(report.NetInfo), &netStat)
|
||
return float64(netStat.TCPEstablished), true
|
||
case "net_rx_bps":
|
||
var netStat model.NetworkStatModel
|
||
json.Unmarshal([]byte(report.NetInfo), &netStat)
|
||
return netStat.NetRxBPS, true
|
||
case "net_tx_bps":
|
||
var netStat model.NetworkStatModel
|
||
json.Unmarshal([]byte(report.NetInfo), &netStat)
|
||
return netStat.NetTxBPS, true
|
||
case "net_rx_errors":
|
||
var netStat model.NetworkStatModel
|
||
json.Unmarshal([]byte(report.NetInfo), &netStat)
|
||
return float64(netStat.NetRxErrors), true
|
||
case "fd":
|
||
var fdStat model.FDMetric
|
||
json.Unmarshal([]byte(report.FDInfo), &fdStat)
|
||
return fdStat.UsagePct, true
|
||
default:
|
||
return 0, false
|
||
}
|
||
}
|
||
|
||
// resolveActiveSiteAlerts 指标恢复正常时关闭对应活跃告警 // Logan.Li
|
||
func (e *AlertEngine) resolveActiveSiteAlerts(taskID uint, triggerType string) {
|
||
now := time.Now()
|
||
e.db.Model(&model.SiteAlert{}).
|
||
Where("task_id = ? AND status = ? AND trigger_type = ?", taskID, model.SiteAlertStatusActive, triggerType).
|
||
Updates(map[string]interface{}{
|
||
"status": model.SiteAlertStatusResolved, "resolved_at": now,
|
||
})
|
||
}
|
||
|
||
// siteRuleStillTriggered 判断任务是否仍满足告警规则 // Logan.Li
|
||
func (e *AlertEngine) siteRuleStillTriggered(rule model.MonitorAlertRule, task model.SiteTask) bool {
|
||
switch rule.Metric {
|
||
case "probe_fail":
|
||
return task.LastResult == "fail" || task.LastResult == "timeout"
|
||
case "probe_timeout":
|
||
return task.LastResult == "timeout"
|
||
case "http_status_error":
|
||
return task.LastResult == "fail"
|
||
case "response_time":
|
||
return e.compareOperator(rule.Operator, task.AvgResponse, rule.Threshold)
|
||
case "availability":
|
||
return e.compareOperator(rule.Operator, task.SuccessRate, rule.Threshold)
|
||
case "ssl_expiry_days":
|
||
if !siteprobe.IsSSLProbeTask(task) {
|
||
return false
|
||
}
|
||
var result model.SiteResult
|
||
e.db.Where("task_id = ? AND ssl_cert_expiry IS NOT NULL", task.ID).
|
||
Order("checked_at DESC").First(&result)
|
||
if result.ID == 0 {
|
||
return false
|
||
}
|
||
return e.compareOperator(rule.Operator, float64(result.SSLDaysRemaining), rule.Threshold)
|
||
default:
|
||
return true
|
||
}
|
||
}
|
||
|
||
// autoResolveSiteAlerts 扫描活跃拨测告警，指标已恢复则自动关闭（含 SSL）// Logan.Li
|
||
func (e *AlertEngine) autoResolveSiteAlerts() {
|
||
var alerts []model.SiteAlert
|
||
e.db.Where("status = ?", model.SiteAlertStatusActive).Find(&alerts)
|
||
if len(alerts) == 0 {
|
||
return
|
||
}
|
||
taskCache := map[uint]model.SiteTask{}
|
||
ruleCache := map[string]model.MonitorAlertRule{}
|
||
now := time.Now()
|
||
for _, alert := range alerts {
|
||
task, ok := taskCache[alert.TaskID]
|
||
if !ok {
|
||
var t model.SiteTask
|
||
if e.db.First(&t, alert.TaskID).Error != nil {
|
||
continue
|
||
}
|
||
task = t
|
||
taskCache[alert.TaskID] = task
|
||
}
|
||
rule, ok := ruleCache[alert.Title]
|
||
if !ok {
|
||
var r model.MonitorAlertRule
|
||
e.db.Where("category = ? AND name = ?", model.MonitorAlertSite, alert.Title).First(&r)
|
||
if r.ID == 0 {
|
||
r.Metric = alert.TriggerType
|
||
}
|
||
rule = r
|
||
ruleCache[alert.Title] = rule
|
||
}
|
||
if rule.Metric == "" {
|
||
rule.Metric = alert.TriggerType
|
||
}
|
||
if !e.siteRuleStillTriggered(rule, task) {
|
||
e.db.Model(&alert).Updates(map[string]interface{}{
|
||
"status": model.SiteAlertStatusResolved, "resolved_at": now,
|
||
})
|
||
}
|
||
}
|
||
}
|
||
|
||
// isMiddlewareMetric 判断 AgentAlert 是否属于中间件 // Logan.Li
|
||
func isMiddlewareMetric(metric string) bool {
|
||
prefixes := []string{"redis_", "mysql_", "middleware_"}
|
||
for _, p := range prefixes {
|
||
if len(metric) >= len(p) && metric[:len(p)] == p {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|