LokiStack/backend/services/alert.go

package services

// 日志告警引擎:周期执行 AlertRule 的 LogQL 计数查询,超阈值发送飞书卡片。
// 设计参照运维 app-alert.py:即时查询 sum(count_over_time(q[窗口])) 精确计数 +
// 按 app/pod 分组定位 + 日志样例 + 告警冷却(同规则冷却期内不重复发)。
import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/url"
	"strings"
	"sync"
	"sync/atomic"
	"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

const (
	// alertRoundTimeout 一轮检查的总时限。超过就中断本轮，
	// 否则重叠保护会让引擎一直"在运行"，后续轮次全部被跳过。
	alertRoundTimeout = 5 * time.Minute
	// alertRuleTimeout 单条规则的查询时限：Loki 单次查询不该超过它。
	alertRuleTimeout = 90 * time.Second
)

// alertCheckRunning 防止 ticker 重叠执行。
var alertCheckRunning atomic.Bool

var alertTicker *time.Ticker

// StartAlertEngine 启动周期检查(main 调用)
func StartAlertEngine() {
	if alertTicker != nil {
		return
	}
	alertTicker = time.NewTicker(alertCheckInterval)
	go func() {
		for range alertTicker.C {
			// ⚠️ 重叠保护（重要）：
			// ticker 是每分钟触发，而一轮规则检查（每条规则都要扫一遍日志内容，
			// 实测 20~30s/条 × 8 条 = 2~4 分钟）远超一分钟。
			// 原来没有这个保护，于是每过一分钟就再开一轮 → 查询越堆越多，
			// 既抢走页面查询的 Loki 资源，又让每轮越来越慢，最后看起来像"告警不响了"。
			// 现在：上一轮没跑完就直接跳过本轮，宁可少评估一次也不堆积。
			if !alertCheckRunning.CompareAndSwap(false, true) {
				log.Printf("[alert] 上一轮检查仍在运行，跳过本轮")
				continue
			}
			start := time.Now()
			ok := true
			func() {
				defer func() {
					// 整轮兜底：即使有未预期的 panic 也不能让引擎 goroutine 挂掉
					if r := recover(); r != nil {
						ok = false
						log.Printf("[alert] 本轮检查 panic: %v", r)
					}
					alertCheckRunning.Store(false)
				}()
				// 每轮总超时：防止个别慢查询把整轮拖成"永久运行"，
				// 那种情况下重叠保护会让引擎再也不启动新一轮。
				ctx, cancel := context.WithTimeout(context.Background(), alertRoundTimeout)
				defer cancel()
				RunAlertCheck(ctx)
			}()
			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
}

// cooldownRecordLast 记录每条规则上次写"冷却"记录的时间，用于节流。
var (
	cooldownRecordMu   sync.Mutex
	cooldownRecordLast = map[uint]time.Time{}
)

// recordCooldownThrottled 同一规则 10 分钟内只允许写一条冷却记录。
func recordCooldownThrottled(ruleID uint) bool {
	cooldownRecordMu.Lock()
	defer cooldownRecordMu.Unlock()
	now := time.Now()
	if last, ok := cooldownRecordLast[ruleID]; ok && now.Sub(last) < 10*time.Minute {
		return false
	}
	cooldownRecordLast[ruleID] = now
	return true
}

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 {
		// 单条规则独立超时 + 独立 panic 隔离：
		// 一条规则查询卡住或代码 panic，不能连带把后面所有规则都跳过。
		ruleCtx, cancel := context.WithTimeout(ctx, alertRuleTimeout)
		func() {
			defer func() {
				if r := recover(); r != nil {
					log.Printf("[alert] 规则 %q 检查 panic: %v", rules[i].Name, r)
					recordDelivery(rules[i].ID, rules[i].Name, 0, "failed", fmt.Sprintf("规则检查异常: %v", r))
				}
			}()
			checkRule(ruleCtx, &rules[i])
		}()
		cancel()
	}
}

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 {
		// 冷却记录节流：命中但处于冷却时，**每 10 分钟最多记一条**。
		// 原来每个 tick（1 分钟）都写，50 条上限下，真实的"已发送"记录
		// 一小时内就被冷却记录挤光了，发送历史基本没法看。
		if recordCooldownThrottled(rule.ID) {
			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
}