LokiStack/backend/services/alert.go

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/pkg/metrics"
	"lokistack/repository"
)

// 告警风暴抑制参数：最近 1 小时内全平台最多成功发送多少条。
//
// 为什么需要：单规则 cooldown 只能防"同一条重复"，挡不住"多条规则同时炸"。
// 实测过一次存储抖动让 8 条内置规则同时命中，10 秒内群里刷了 8 条，
// 之后大家开始无视这个群 —— 告警疲劳比漏报更难挽回。
const (
	alertStormWindow     = time.Hour
	alertStormMaxPerHour = 12
)

// overStormLimit 判断是否已经触发全局抑制。
// 返回 (是否抑制, 最近窗口内已发送条数)。
func overStormLimit() (bool, int) {
	db := repository.GetDB()
	if db == nil {
		return false, 0
	}
	cutoff := time.Now().Add(-alertStormWindow)
	db.RLock()
	defer db.RUnlock()
	sent := 0
	for _, delivery := range db.AlertDeliveries {
		if delivery.Status == "sent" && delivery.CreatedAt.After(cutoff) {
			sent++
		}
	}
	return sent >= alertStormMaxPerHour, sent
}

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 {
			start := time.Now()
			// 告警引擎是后台 goroutine：卡死或 panic 时页面完全看不出来，
			// 只能靠这两个指标发现（配合 Grafana 告警）
			ok := true
			func() {
				defer func() {
					if r := recover(); r != nil {
						ok = false
					}
				}()
				RunAlertCheck(context.Background())
			}()
			metrics.ObserveAlertEvaluation(time.Since(start).Seconds(), ok)
		}
	}()
}

// 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"))
	}

	// ── 告警风暴抑制：全平台每小时最多发 N 条 ──────────────────────
	//
	// 单条规则有 cooldown 只能说"同一条别重复"，挡不住"20 条规则同时炸"。
	// 实测过：一次存储抖动让 8 条内置规则同时命中，10 秒内往群里刷了 8 条，
	// 之后大家开始无视这个群。所以再加一道**全局闸门**：
	// 最近 1 小时内成功发送数达到上限后，后续命中只记"被抑制"，不再发送。
	// 被抑制的记录仍写进发送历史（可见、可追溯），并写明原因。
	if suppressed, sentCount := overStormLimit(); suppressed {
		recordDelivery(rule.ID, rule.Name, count, "cooldown",
			fmt.Sprintf("告警风暴抑制：最近 1 小时已发送 %d 条（上限 %d），本条未发送", sentCount, alertStormMaxPerHour))
		return
	}

	// 扇出到全部关联通道;部分失败记录失败原因(成功任一即更新冷却)
	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
}