opsx
5e59347c40
=== 0. 【严重】修复上一版引入的死锁（平台写路径全卡死）===
上一版把保留期读取写成 `GetRetentionSettings()`（内部 db.RLock()），
却在 `db.Lock()` 已持有的情况下调用（recordDelivery / RecordAudit 的裁剪路径）。
Go 的 sync.RWMutex **不可重入** → 第一次写投递历史/审计日志就永久死锁。
现象：登录 28 秒返回空（登录也要写审计）、所有写接口挂住。
go build / go vet / 之前的单测（不触发写路径）全都查不出来。
修法：
- 拆出 retentionFromDBLocked(db) / alertDeliveryRetentionLocked(db) /
auditLogRetentionLocked(db)，调用方已持锁时用无锁版本
- upsertSettingLocked 里**不能**用 db.NextID()（内部 d.mu.Lock()，同样死锁），
改为锁内自算 max(id)+1
- 新增 retention_test.go：持写锁调用裁剪函数，2 秒超时未返回即判死锁
实测修复后：登录 103ms（原 28000ms）、panic/deadlock 命中 0。
=== 1. 平台数据保留期可配（系统设置 → 通用配置 → 日志保留时间）===
新增 GET/PUT /api/v1/settings/retention：
alertDeliveryDays 告警发送历史保留天数，默认 7（原写死"最近 50 条"）
auditLogDays 审计日志保留天数，默认 90（原写死常量）
保存后立即按新策略裁剪一次，不必等下次触发。
校验：发送历史 1~365 天、审计 1~3650 天，超出返回明确中文原因（不是 500）。
注意与「Loki 日志保留期」区分：那个是部署级配置（组件部署 → 保留策略）。
=== 2. 发送历史：50 条硬上限 → 保留 7 天 ===
后端 Deliveries 接口原来写死 `len(out) < 50`，现在按保留期取（响应上限 500 可调），
标题改为「发送历史（保留 N 天）」，N 从设置里读。
=== 3. 菜单「日志服务」→「组件部署」===
=== 4. 分析页：去掉「全部命名空间」+ 改手动刷新 ===
- 移除「全部命名空间」选项：{namespace=~".+"} 要扫全集群（约 370GB/天），
一次分析几十秒，得不偿失；历史值 __all__ 也会被忽略并回退到策略/第一个 ns
- 筛选改为「选择值 / 已生效值」分离：切换命名空间或时间范围**不再自动查询**，
只把刷新按钮点亮并提示「筛选条件已改，点刷新生效」；
点一次才应用并取数。
原因：一次分析要跑多条 count_over_time，切一下下拉就重查一轮，
快速连点会把 Loki 打满（并发护栏 429），还拖慢别人。
=== 实测（9739c7c092e7）===
登录 103ms / token 273 字符
GET retention → {alertDeliveryDays:7, auditLogDays:90}
PUT 0 天 → 400「告警发送历史保留天数需在 1~365 之间」
平台日志 panic/deadlock 命中 0
go test ./... 全绿（含 2 个新死锁回归用例）
415 lines
13 KiB
Go
415 lines
13 KiB
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
|
||
}
|
||
|
||
// pruneDeliveriesLocked 按保留期 + 条数上限裁剪投递历史（调用方必须已持锁）。
|
||
func pruneDeliveriesLocked(db *repository.JsonDB) {
|
||
cut := 0
|
||
// 注意：调用方已持写锁，必须用无锁读取版本，否则 RWMutex 会死锁
|
||
deadline := time.Now().Add(-alertDeliveryRetentionLocked(db))
|
||
for cut < len(db.AlertDeliveries) && db.AlertDeliveries[cut].CreatedAt.Before(deadline) {
|
||
cut++
|
||
}
|
||
if remain := len(db.AlertDeliveries) - cut; remain > maxAlertDeliveries {
|
||
cut += remain - maxAlertDeliveries
|
||
}
|
||
if cut > 0 {
|
||
db.AlertDeliveries = append([]models.AlertDelivery{}, db.AlertDeliveries[cut:]...)
|
||
}
|
||
}
|
||
|
||
// PruneAlertDeliveries 按当前保留期裁剪一次（供保存设置后立即生效）。
|
||
func PruneAlertDeliveries() {
|
||
db := repository.GetDB()
|
||
if db == nil {
|
||
return
|
||
}
|
||
db.Lock()
|
||
defer db.Unlock()
|
||
before := len(db.AlertDeliveries)
|
||
pruneDeliveriesLocked(db)
|
||
if len(db.AlertDeliveries) != before {
|
||
db.RequestSave()
|
||
}
|
||
}
|
||
|
||
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,
|
||
})
|
||
// 保留策略：按 告警发送历史保留天数（默认 7 天，系统设置可改）+ 条数硬上限兜底。
|
||
// 原来写死 只留最近 50 条，真实发送记录一小时内就被冷却记录挤光了。
|
||
pruneDeliveriesLocked(db)
|
||
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
|
||
}
|