package services // 日志告警引擎:周期执行 AlertRule 的 LogQL 计数查询,超阈值发送飞书卡片。 // 设计参照运维 app-alert.py:即时查询 sum(count_over_time(q[窗口])) 精确计数 + // 按 app/pod 分组定位 + 日志样例 + 告警冷却(同规则冷却期内不重复发)。 import ( "context" "encoding/json" "fmt" "net/url" "strings" "time" "lokistack/models" "lokistack/repository" ) const alertCheckInterval = time.Minute var alertTicker *time.Ticker // StartAlertEngine 启动周期检查(main 调用) func StartAlertEngine() { if alertTicker != nil { return } alertTicker = time.NewTicker(alertCheckInterval) go func() { for range alertTicker.C { RunAlertCheck(context.Background()) } }() } // StopAlertEngine 优雅关停 func StopAlertEngine() { if alertTicker != nil { alertTicker.Stop() alertTicker = nil } } // lokiInstantCount 即时计数:sum(count_over_time(q[window])) func lokiInstantCount(ctx context.Context, src *models.LogSource, query string, windowMin int) (int, error) { wrapped := fmt.Sprintf("sum(count_over_time(%s[%dm]))", query, windowMin) q := url.Values{} q.Set("query", wrapped) q.Set("time", fmt.Sprintf("%d", time.Now().UnixNano())) body, _, err := lokiProxyGET(ctx, src, "/loki/api/v1/query", q) if err != nil { return 0, err } var resp struct { Data struct { Result []struct { Value [2]interface{} `json:"value"` } `json:"result"` } `json:"data"` } if err := json.Unmarshal(body, &resp); err != nil { return 0, err } if len(resp.Data.Result) == 0 { return 0, nil } count := 0 switch v := resp.Data.Result[0].Value[1].(type) { case string: fmt.Sscanf(v, "%d", &count) case float64: count = int(v) } return count, nil } // lokiGroupTop 按 app/pod 分组定位(取前 10) func lokiGroupTop(ctx context.Context, src *models.LogSource, query string, windowMin int) []string { wrapped := fmt.Sprintf("sum by (app, pod) (count_over_time(%s[%dm]))", query, windowMin) q := url.Values{} q.Set("query", wrapped) q.Set("time", fmt.Sprintf("%d", time.Now().UnixNano())) body, _, err := lokiProxyGET(ctx, src, "/loki/api/v1/query", q) if err != nil { return nil } var resp struct { Data struct { Result []struct { Metric map[string]string `json:"metric"` Value [2]interface{} `json:"value"` } `json:"result"` } `json:"data"` } if err := json.Unmarshal(body, &resp); err != nil { return nil } lines := []string{} for _, r := range resp.Data.Result { count := 0 if v, ok := r.Value[1].(string); ok { fmt.Sscanf(v, "%d", &count) } else if v, ok := r.Value[1].(float64); ok { count = int(v) } app := r.Metric["app"] if app == "" { app = "?" } pod := r.Metric["pod"] if pod == "" { pod = "?" } lines = append(lines, fmt.Sprintf("- %s / %s(%d 条)", app, pod, count)) if len(lines) >= 10 { break } } return lines } // lokiSampleLines 日志样例(最多 3 条,每条截 220) func lokiSampleLines(ctx context.Context, src *models.LogSource, query string, windowMin int) []string { res, err := QueryLokiLogs(ctx, src, query, 0, 0, 3, "backward", "") if err != nil { return nil } out := []string{} for _, e := range res.Entries { line := e.Message if len(line) > 220 { line = line[:220] } out = append(out, "> "+line) if len(out) >= 3 { break } } return out } func recordDelivery(ruleID uint, ruleName string, count int, status, reason string) { db := repository.GetDB() newID := db.NextID("alertDeliveries") db.Lock() db.AlertDeliveries = append(db.AlertDeliveries, models.AlertDelivery{ ID: newID, RuleID: ruleID, RuleName: ruleName, Count: count, Status: status, Reason: reason, }) // 只保留最近 50 条 if len(db.AlertDeliveries) > 50 { db.AlertDeliveries = db.AlertDeliveries[len(db.AlertDeliveries)-50:] } db.RequestSave() db.Unlock() } // RunAlertCheck 执行一次全部启用规则的检查(定时器与手动触发共用) func RunAlertCheck(ctx context.Context) { db := repository.GetDB() db.RLock() rules := make([]models.AlertRule, 0, len(db.AlertRules)) for i := range db.AlertRules { if db.AlertRules[i].Enabled { rules = append(rules, db.AlertRules[i]) } } db.RUnlock() for i := range rules { checkRule(ctx, &rules[i]) } } func checkRule(ctx context.Context, rule *models.AlertRule) { src := GetLogSource(rule.SourceID) if src == nil { return } count, err := lokiInstantCount(ctx, src, rule.Query, rule.WindowMin) if err != nil { recordDelivery(rule.ID, rule.Name, 0, "failed", err.Error()) return } if count <= rule.Threshold { return } // 冷却检查(同规则) db := repository.GetDB() db.RLock() lastFired := rule.LastFiredAt cooldown := rule.CooldownMin if cooldown <= 0 { cooldown = 60 } db.RUnlock() if !lastFired.IsZero() && time.Since(lastFired) < time.Duration(cooldown)*time.Minute { recordDelivery(rule.ID, rule.Name, count, "cooldown", "冷却中,未重复发送") return } // 组装卡片 groups := lokiGroupTop(ctx, src, rule.Query, rule.WindowMin) samples := lokiSampleLines(ctx, src, rule.Query, rule.WindowMin) ts := time.Now().Format("2006-01-02 15:04:05") lines := []string{ fmt.Sprintf("**告警时间**: %s", ts), fmt.Sprintf("**规则**: %s", rule.Name), fmt.Sprintf("**命中数量**: %d 条(%d 分钟内)", count, rule.WindowMin), fmt.Sprintf("**触发阈值**: > %d 条", rule.Threshold), fmt.Sprintf("**查询**: `%s`", rule.Query), } if len(groups) > 0 { lines = append(lines, "**来源 app/pod**:\n"+strings.Join(groups, "\n")) } if len(samples) > 0 { lines = append(lines, "**日志样例**:\n"+strings.Join(samples, "\n")) } // 扇出到全部关联通道;部分失败记录失败原因(成功任一即更新冷却) title := "🔴 日志告警 " + rule.Name var failMsgs []string if len(rule.ChannelIDs) > 0 { for _, chID := range rule.ChannelIDs { if err := SendLarkCardToChannel(chID, title, "red", lines); err != nil { chName := fmt.Sprintf("通道%d", chID) if ch := GetNotifyChannel(chID); ch != nil { chName = ch.Name } failMsgs = append(failMsgs, fmt.Sprintf("%s: %v", chName, err)) } } } else if cfg := GetNotifyConfig(); cfg.Enabled && cfg.Webhook != "" { // 兼容:未关联通道回退旧全局配置 if err := SendLarkCard(title, "red", lines); err != nil { failMsgs = append(failMsgs, err.Error()) } } else { failMsgs = append(failMsgs, "规则未关联通知通道") } if len(failMsgs) > 0 { recordDelivery(rule.ID, rule.Name, count, "failed", strings.Join(failMsgs, "; ")) return } recordDelivery(rule.ID, rule.Name, count, "sent", fmt.Sprintf("已扇出到 %d 个通道", maxInt(len(rule.ChannelIDs), 1))) // 更新 LastFiredAt db.Lock() for i := range db.AlertRules { if db.AlertRules[i].ID == rule.ID { db.AlertRules[i].LastFiredAt = time.Now() break } } db.RequestSave() db.Unlock() recordDelivery(rule.ID, rule.Name, count, "sent", "") } func maxInt(a, b int) int { if a > b { return a } return b }