From b7de337c9e256094dca1c310e6d098e96e69d661 Mon Sep 17 00:00:00 2001 From: opsx Date: Fri, 30 Oct 2026 16:52:55 +0000 Subject: [PATCH] =?UTF-8?q?perf(daily-volume):=20=E6=AF=8F=E6=97=A5?= =?UTF-8?q?=E9=87=8F=E6=8A=93=E5=8F=96=E6=8B=86=E6=88=90=E3=80=8C=E4=BD=93?= =?UTF-8?q?=E7=A7=AF=E6=AF=8F=2030=20=E5=88=86=E9=92=9F=20/=20=E6=9D=A1?= =?UTF-8?q?=E6=95=B0=E6=AF=8F=E5=A4=A9=E6=9C=80=E5=A4=9A=203=20=E6=AC=A1?= =?UTF-8?q?=E3=80=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit === 现象 === 用户问：每日日志量不是有定时任务记录吗？为什么还要每次查询？ 盘了一遍发现：定时任务确实有（每 30 分钟），接口也确实只读库（实测 0.1ms）， 但每轮抓取都顺手跑一次**整天 index/stats** 拿条数（实测 6ms ~ 20s，随负载波动）， 2 天 × 多个数据源 × 每 30 分钟 = 一天几十次重查询，纯浪费单实例 Loki 的算力， 还会和用户页面上的查询抢 CPU。 第二条更糟：仓库里所有快照的 partial 全是 true。 根因：partial 里混了两种语义 ——「今天还没过完」和「起点超出 24h」。 后者让昨天/前天的柱子也画成虚线空心，看起来「每天都在进行中」。 === 修法 === 1. 抓取按成本拆成两条路径（体积走 volume_range，条数走 index/stats）： - 体积：今天 + 昨天，每 30 分钟刷，54ms~4s，无感 - 条数：只对**已结束**且 entries==0 的天补，每天最多 3 次、间隔≥1h； 尝试次数**先写库再查**，进程中途被杀也不会退化成每 30 分钟重试 2. partial 只表示「今天还没过完」，昨天起是实心柱 3. 0 与「未留存」分开：起点超出 24h 且查不到数据 → **跳过落库**保持 Missing。 否则「数据被 retention 删掉」会被画成「那天没有日志」，比灰色柱误导得多 4. 每次抓取打日志：`[daily-volume] 2026-10-30 体积 291.37 GB 已落库（volume_range 4.27s，34 个桶）` —— 以后不用猜「到底抓没抓、走的哪条路径」 === 实测取证（本集群）=== Loki 直接查（port-forward :13100）： 10-30(16.8h) volume_range 291.37GB/34桶 index/stats entries=2.21B 10-29(24h) volume_range 121.99GB/19桶 index/stats entries=927.9M 10-28/10-27 volume_range 0 桶、index/stats 0 —— 已被 retention 删掉 同范围 count_over_time 两次 60s 超时（永远别用在每日量上） 平台库里的快照与上述数字一致（288.32 / 121.75 GB），说明落库值可信； 10-27/10-28 的 34.87/25.65 GB 是当时抓的 —— 这正是「Loki 只留 1~3 天， 历史必须自己落库」的实证。 go vet ./services/ ./models/ 干净；go build 通过；go test ./services/ 通过。 --- backend/models/daily_volume.go | 13 +- backend/services/daily_volume.go | 227 ++++++++++++++++++++++++------- 2 files changed, 187 insertions(+), 53 deletions(-) diff --git a/backend/models/daily_volume.go b/backend/models/daily_volume.go index fbfc1f9..04c4ece 100644 --- a/backend/models/daily_volume.go +++ b/backend/models/daily_volume.go @@ -25,6 +25,15 @@ type DailyVolumeSnapshot struct { Entries int64 `json:"entries"` // 日志条数 // Partial 该日期是否为"不完整的一天"（当天抓取时还没过完）。 // 前端用它区分"真实低峰"和"只统计到一半"。 - Partial bool `json:"partial"` - CapturedAt time.Time `json:"capturedAt"` + // + // ⚠️ 只表示"今天还没过完"。早期实现把"起点超出 24 小时"也算成 partial， + // 导致昨天、前天的柱子也画成虚线空心，看起来像"每天都在进行中"（已修）。 + Partial bool `json:"partial"` + // EntriesAttempts 条数（index/stats）已尝试次数。 + // 整天跨度统计很贵（实测 6ms ~ 20s，取决于实例负载）， + // 所以只对"已经过完的天"补，且最多重试 entriesMaxAttempts 次， + // 避免每 30 分钟的定时任务无脑重跑一遍。 + EntriesAttempts int `json:"entriesAttempts"` + LastEntriesAttempt time.Time `json:"lastEntriesAttempt"` + CapturedAt time.Time `json:"capturedAt"` } diff --git a/backend/services/daily_volume.go b/backend/services/daily_volume.go index aa50f01..76a2793 100644 --- a/backend/services/daily_volume.go +++ b/backend/services/daily_volume.go @@ -2,6 +2,7 @@ package services import ( "context" + "fmt" "log" "time" @@ -31,6 +32,13 @@ const ( dailyVolumeBackfillDays = 2 // dailyVolumeRetentionDays 平台自己留多久的历史。一年足够看年度趋势。 dailyVolumeRetentionDays = 400 + + // entriesMaxAttempts 条数（index/stats）每天最多尝试次数。 + // 整天跨度要去重统计全部 chunk：实测空闲时 6ms、有并发查询时 20s 以上， + // 属于"附加指标"，不值得无限重试。 + entriesMaxAttempts = 3 + // entriesRetryInterval 两次条数尝试之间的最小间隔。 + entriesRetryInterval = time.Hour ) // dateKey 返回本地时区的 YYYY-MM-DD。 @@ -45,10 +53,22 @@ func dayRange(day time.Time) (int64, int64) { return start.UnixNano(), start.AddDate(0, 0, 1).UnixNano() } -// CaptureDailyVolume 抓取指定日期的日志量并落库（按天幂等：同一天重复抓会覆盖）。 +// formatGB 用于日志输出，运维看日志时比裸字节直观。 +func formatGB(bytes int64) string { + return fmt.Sprintf("%.2f GB", float64(bytes)/1e9) +} + +// CaptureDailyVolume 抓取指定日期的**日志体积**并落库（按天幂等：同一天重复抓会覆盖）。 // -// 数据来源是 Loki 的 index/stats（只读索引元数据），实测毫秒级， -// 所以就算是补几十天历史也不会拖慢任何东西。 +// 只用 index/volume_range：它只读 TSDB 索引里记录的 chunk 字节数，不解码日志内容。 +// 同一 selector 24 小时实测（本集群 ~370GB/天）： +// +// index/volume_range 54ms ~ 4s ← 本函数走的路径 +// index/stats 6ms ~ 20s ← 条数才需要，见 captureDailyEntries +// count_over_time 60s 超时 ← 永远不要用在"每日量"上 +// +// 条数不在这里抓：它只能走 index/stats 且整天跨度很贵，独立成 captureDailyEntries， +// 每天最多尝试 3 次，不跟着 30 分钟的定时任务反复跑。 func CaptureDailyVolume(source *models.LogSource, day time.Time, partial bool) error { if source == nil { return nil @@ -62,19 +82,22 @@ func CaptureDailyVolume(source *models.LogSource, day time.Time, partial bool) e if start >= end { return nil } - // 回补历史时，如果那天的起点已经超出 Loki 保留期（默认 24h）， - // Loki 里根本没那么多数据 —— 必须标成"不完整"， - // 否则图上会把"数据被删了"误读成"那天日志量真的很少"。 - if start < time.Now().Add(-24*time.Hour).UnixNano() { - partial = true - } - // ── 体积：index/volume_range（只读 TSDB chunk 元数据，不解码日志内容）── - // 实测同一条件 24 小时：index/stats 要 20s（遇超时），volume_range 111ms。 - // 用户关心的"每日日志量"就是体积，所以主路径走它。 + // 起点已超出常见保留期（24h）时，Loki 里可能已经没有那天的数据了。 + // 这种情况下查出来的 0 字节**绝不能落库**： + // 落库就会把"数据被 retention 删掉了"画成"那天没有日志"， + // 比灰色柱（未留存）误导得多。所以按"跳过落库 = 保持 Missing"处理。 + stale := start < time.Now().Add(-24*time.Hour).UnixNano() + + begin := time.Now() ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() buckets, volErr := LokiHistogram(ctx, source, `{namespace=~".+"}`, start, end) + elapsed := time.Since(begin) + if volErr != nil { + return fmt.Errorf("volume_range: %w", volErr) + } + var totalBytes int64 for _, b := range buckets { switch v := b["count"].(type) { @@ -85,28 +108,23 @@ func CaptureDailyVolume(source *models.LogSource, day time.Time, partial bool) e } } - // ── 条数：只能 index/stats，慢（整天约 20s）── - // 降级为"尽力而为"：拿不到就把条数留 0，绝不让它拖住体积入库。 - // 前端图只画体积，条数缺失不影响可读性。 - var totalEntries int64 - entryCtx, entryCancel := context.WithTimeout(context.Background(), 25*time.Second) - if stats, err := LokiStatsFor(entryCtx, source, `{namespace=~".+"}`, start, end); err == nil { - totalEntries = stats.Entries - if totalBytes == 0 { - totalBytes = stats.Bytes - } - } else { - log.Printf("[daily-volume] %s 条数统计超时/失败（不影响体积）: %v", dateKey(day), err) - } - entryCancel() - if volErr != nil && totalBytes == 0 { - return volErr - } - if totalBytes == 0 { + if totalBytes == 0 && stale { + log.Printf("[daily-volume] %s 超出保留期且查不到数据，跳过落库（图上显示「未留存」）", dateKey(day)) return nil } date := dateKey(day) + if err := upsertVolumeSnapshot(source.ID, date, totalBytes, partial); err != nil { + return err + } + log.Printf("[daily-volume] %s 体积 %s 已落库（volume_range %s，%d 个桶）", + date, formatGB(totalBytes), elapsed.Round(time.Millisecond), len(buckets)) + return nil +} + +// upsertVolumeSnapshot 写入/更新某天的体积快照（幂等）。 +// 只改体积与 partial，**不动已经抓到的条数**（条数贵，抓到就留着）。 +func upsertVolumeSnapshot(sourceID uint, date string, bytes int64, partial bool) error { db := repository.GetDB() if db == nil { return nil @@ -123,13 +141,8 @@ func CaptureDailyVolume(source *models.LogSource, day time.Time, partial bool) e if snap.ID > nextID { nextID = snap.ID } - // 同一天同源：覆盖（幂等） - if snap.Date == date && snap.SourceID == source.ID && snap.Namespace == "" { - snap.Bytes = totalBytes - // 条数拿不到时保留上一次的值，避免"一次超时把历史条数清零" - if totalEntries > 0 { - snap.Entries = totalEntries - } + if snap.Date == date && snap.SourceID == sourceID && snap.Namespace == "" { + snap.Bytes = bytes snap.Partial = partial snap.CapturedAt = time.Now() db.RequestSave() @@ -140,10 +153,9 @@ func CaptureDailyVolume(source *models.LogSource, day time.Time, partial bool) e db.DailyVolumeSnapshots = append(db.DailyVolumeSnapshots, models.DailyVolumeSnapshot{ ID: nextID, Date: date, - SourceID: source.ID, + SourceID: sourceID, Namespace: "", - Bytes: totalBytes, - Entries: totalEntries, + Bytes: bytes, Partial: partial, CapturedAt: time.Now(), }) @@ -151,7 +163,116 @@ func CaptureDailyVolume(source *models.LogSource, day time.Time, partial bool) e return nil } +// captureDailyEntries 给"已经过完、但还没有条数"的天补一次条数（尽力而为）。 +// +// 为什么不等同于体积每 30 分钟抓一次： +// 条数只能走 index/stats，要跨该天全部 chunk 做 streams/entries 去重统计， +// 实测 6ms（空闲）~ 20s（有并发查询时）。每 30 分钟 × 每一天都重跑一遍， +// 既浪费 Loki 的 CPU，又会和用户页面上的查询抢同一台单实例的算力。所以： +// +// • 只处理 Date < 今天 的天（今天没结束，条数没有意义） +// • 每天最多 entriesMaxAttempts 次，两次间隔 ≥ entriesRetryInterval +// • 尝试次数**先写库再查**：进程中途被杀也不会退化成"每 30 分钟重试" +// • 拿不到就留 0，前端图只画体积，不影响可读性 +func captureDailyEntries(source *models.LogSource) { + if source == nil { + return + } + now := time.Now() + today := dateKey(now) + + var dates []string + db := repository.GetDB() + if db == nil { + return + } + db.RLock() + for _, snap := range db.DailyVolumeSnapshots { + if snap.SourceID != source.ID || snap.Namespace != "" { + continue + } + if snap.Date >= today || snap.Entries > 0 { + continue + } + if snap.EntriesAttempts >= entriesMaxAttempts { + continue + } + if !snap.LastEntriesAttempt.IsZero() && now.Sub(snap.LastEntriesAttempt) < entriesRetryInterval { + continue + } + dates = append(dates, snap.Date) + } + db.RUnlock() + + for _, date := range dates { + // 先把"尝试"记下来再查：查的过程中进程重启，也不会变成无限重试 + markEntriesAttempt(source.ID, date) + + day, err := time.ParseInLocation("2006-01-02", date, time.Local) + if err != nil { + continue + } + start, end := dayRange(day) + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second) + begin := time.Now() + stats, statErr := LokiStatsFor(ctx, source, `{namespace=~".+"}`, start, end) + cancel() + if statErr != nil { + log.Printf("[daily-volume] %s 条数统计失败（体积已落库，不影响图表）: %v", date, statErr) + continue + } + if stats.Entries <= 0 { + continue + } + setSnapshotEntries(source.ID, date, stats.Entries) + log.Printf("[daily-volume] %s 条数 %d（index/stats %s）", + date, stats.Entries, time.Since(begin).Round(time.Millisecond)) + } +} + +// markEntriesAttempt 记录一次条数尝试（先写库再查询，失败也计数）。 +func markEntriesAttempt(sourceID uint, date string) { + db := repository.GetDB() + if db == nil { + return + } + db.Lock() + defer db.Unlock() + for i := range db.DailyVolumeSnapshots { + snap := &db.DailyVolumeSnapshots[i] + if snap.Date == date && snap.SourceID == sourceID && snap.Namespace == "" { + snap.EntriesAttempts++ + snap.LastEntriesAttempt = time.Now() + db.RequestSave() + return + } + } +} + +// setSnapshotEntries 写入条数。 +func setSnapshotEntries(sourceID uint, date string, entries int64) { + db := repository.GetDB() + if db == nil { + return + } + db.Lock() + defer db.Unlock() + for i := range db.DailyVolumeSnapshots { + snap := &db.DailyVolumeSnapshots[i] + if snap.Date == date && snap.SourceID == sourceID && snap.Namespace == "" { + snap.Entries = entries + db.RequestSave() + return + } + } +} + // CaptureRecentDailyVolumes 回补最近 N 天（含今天），供启动时与定时任务调用。 +// +// 两条路径分开： +// +// 体积（volume_range，毫秒级）→ 今天 + 昨天，每 30 分钟刷新，当天柱子保持新鲜 +// 条数（index/stats，可能 20s）→ 只给已结束且还没抓到条数的天补，每天最多 3 次 func CaptureRecentDailyVolumes() { db := repository.GetDB() if db == nil { @@ -169,12 +290,13 @@ func CaptureRecentDailyVolumes() { src := src for offset := dailyVolumeBackfillDays - 1; offset >= 0; offset-- { day := now.AddDate(0, 0, -offset) - // 今天是不完整的一天 + // 只有今天是不完整的一天 partial := offset == 0 if err := CaptureDailyVolume(&src, day, partial); err != nil { log.Printf("[daily-volume] 抓取 %s %s 失败: %v", src.Name, dateKey(day), err) } } + captureDailyEntries(&src) } pruneDailyVolumeSnapshots() } @@ -203,7 +325,12 @@ func pruneDailyVolumeSnapshots() { // GetDailyVolumeHistory 从平台留存的历史里取最近 days 天，按日期升序返回。 // -// 今天如果还没抓过，会实时补一次（这样用户切到页面时当天数字是最新的）。 +// 这个接口**只读库，绝不触发任何 Loki 查询** —— 图表刷新不应该产生任何 Loki 负载。 +// 当天那根柱子由后台定时任务（每 30 分钟，volume_range 毫秒级）补，慢几分钟可接受。 +// +// 事故复盘：这里原来同步 await 一次"整天 index/stats"，那次查询超时 20 秒， +// 于是 /logs/daily-volume 耗时 20s → 前端超时 → 页面显示"每日日志量没数据"； +// 而且这条重查询还和页面的 histogram/stats 抢 Loki，导致那两个接口同时 502。 func GetDailyVolumeHistory(sourceID uint, days int) []DailyVolumePoint { if days <= 0 { days = 7 @@ -212,13 +339,6 @@ func GetDailyVolumeHistory(sourceID uint, days int) []DailyVolumePoint { days = dailyVolumeRetentionDays } - // 这个接口**只读库**，绝不触发任何 Loki 查询。 - // - // 事故复盘：这里原来同步 await 一次"整天 index/stats"，那次查询超时 20 秒， - // 于是 /logs/daily-volume 耗时 20s → 前端超时 → 页面显示"每日日志量没数据"； - // 而且这条重查询还和页面的 histogram/stats 抢 Loki，导致那两个接口同时 502。 - // 当天那根柱子由后台定时任务（每 30 分钟）补，慢几分钟完全可接受。 - // 始终返回 days 个点（缺失的那天标 Missing）。 // 之前只返回"已抓到的快照"，于是 7/30/90/365 全都返回同样 4 个点， // 用户切范围看不到任何变化，以为"切换没反应"。 @@ -256,12 +376,17 @@ func GetDailyVolumeHistory(sourceID uint, days int) []DailyVolumePoint { // StartDailyVolumeCapture 后台每日抓取。 // -// 每 30 分钟检查一次：只要发现"昨天"还没落库就立刻补， -// 而不是死等 00:10 —— 平台可能整晚都没运行，定时器会错过那个点。 +// 每 30 分钟跑一轮（体积是毫秒级的 volume_range，跑得勤一点让"今天"这根柱子接近实时）： +// - 今天 + 昨天的体积：每轮都刷 +// - 条数：只在"已结束且还没有条数"的天触发，每天最多 3 次（见 captureDailyEntries） +// +// 用"检查式回补"而不是死等 00:10 的原因：平台可能整晚都没运行，定时器会错过那个点； +// 每轮都检查"昨天有没有落库"，漏了就立刻补，天然幂等。 func StartDailyVolumeCapture() { go func() { // 启动先补最近两天（覆盖重启场景） time.Sleep(20 * time.Second) + log.Printf("[daily-volume] 启动回补最近 %d 天", dailyVolumeBackfillDays) CaptureRecentDailyVolumes() ticker := time.NewTicker(30 * time.Minute)