package services import ( "context" "fmt" "log" "time" "lokistack/models" "lokistack/repository" ) // DailyVolumePoint 返回给前端的单日数据点 type DailyVolumePoint struct { Date string `json:"date"` Bytes int64 `json:"bytes"` Entries int64 `json:"entries"` // Partial 当天还没过完(或抓取时数据不全),前端用虚线/浅色区分 Partial bool `json:"partial"` // FromHistory true=来自平台留存历史;false=刚从 Loki 实时查的 FromHistory bool `json:"fromHistory"` // Missing 该天平台没有抓到快照(超过留存期/平台当天没运行)。 // 必须和"当天日志量真的是 0"区分开:前者是"不知道",后者是"确实没日志"。 // 否则 30/90/365 天的图上会出现大片 0,被误读成"日志量暴跌"。 Missing bool `json:"missing"` } const ( // dailyVolumeBackfillDays 启动时回补天数。 // 取 2 天:昨天(完整)+ 今天(进行中),足够覆盖"重启导致漏抓"的常见情况。 // 不设更大是因为 Loki retention 通常只有 24h,更早的已经查不到了。 dailyVolumeBackfillDays = 2 // dailyVolumeRetentionDays 平台自己留多久的历史。一年足够看年度趋势。 dailyVolumeRetentionDays = 400 // entriesMaxAttempts 条数(index/stats)每天最多尝试次数。 // 整天跨度要去重统计全部 chunk:实测空闲时 6ms、有并发查询时 20s 以上, // 属于"附加指标",不值得无限重试。 entriesMaxAttempts = 3 // entriesRetryInterval 两次条数尝试之间的最小间隔。 entriesRetryInterval = time.Hour ) // dateKey 返回本地时区的 YYYY-MM-DD。 func dateKey(t time.Time) string { return t.In(time.Local).Format("2006-01-02") } // dayRange 返回某天本地时区的 [00:00, 次日 00:00) 纳秒区间。 func dayRange(day time.Time) (int64, int64) { local := day.In(time.Local) start := time.Date(local.Year(), local.Month(), local.Day(), 0, 0, 0, 0, time.Local) return start.UnixNano(), start.AddDate(0, 0, 1).UnixNano() } // formatGB 用于日志输出,运维看日志时比裸字节直观。 func formatGB(bytes int64) string { return fmt.Sprintf("%.2f GB", float64(bytes)/1e9) } // CaptureDailyVolume 抓取指定日期的**日志体积**并落库(按天幂等:同一天重复抓会覆盖)。 // // 只用 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 } start, end := dayRange(day) // 当天未结束时截到"现在",否则会把未来时间也查进去 now := time.Now().UnixNano() if end > now { end = now } if start >= end { return nil } // 起点已超出常见保留期(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) { case int64: totalBytes += v case float64: totalBytes += int64(v) } } 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 } db.Lock() defer db.Unlock() nextID := uint(0) // 用下标取地址:`for _, snap := range` 拿到的是**副本**, // 改 snap 的字段不会写回切片(第一版就是这里写错了, // 导致同一天永远保持首次抓取的值、partial 不会更新)。 for i := range db.DailyVolumeSnapshots { snap := &db.DailyVolumeSnapshots[i] if snap.ID > nextID { nextID = snap.ID } if snap.Date == date && snap.SourceID == sourceID && snap.Namespace == "" { snap.Bytes = bytes snap.Partial = partial snap.CapturedAt = time.Now() db.RequestSave() return nil } } nextID++ db.DailyVolumeSnapshots = append(db.DailyVolumeSnapshots, models.DailyVolumeSnapshot{ ID: nextID, Date: date, SourceID: sourceID, Namespace: "", Bytes: bytes, Partial: partial, CapturedAt: time.Now(), }) db.RequestSave() 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 { return } db.RLock() sources := make([]models.LogSource, 0, len(db.LogSources)) for _, src := range db.LogSources { sources = append(sources, src) } db.RUnlock() now := time.Now() for _, src := range sources { 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() } // pruneDailyVolumeSnapshots 丢弃超过保留期的历史,并修正历史遗留的 partial 标记。 func pruneDailyVolumeSnapshots() { db := repository.GetDB() if db == nil { return } cutoff := dateKey(time.Now().AddDate(0, 0, -dailyVolumeRetentionDays)) today := dateKey(time.Now()) db.Lock() defer db.Unlock() changed := false kept := make([]models.DailyVolumeSnapshot, 0, len(db.DailyVolumeSnapshots)) for _, snap := range db.DailyVolumeSnapshots { if snap.Date < cutoff { continue } kept = append(kept, snap) } if len(kept) != len(db.DailyVolumeSnapshots) { log.Printf("[daily-volume] 清理过期快照 %d 条", len(db.DailyVolumeSnapshots)-len(kept)) db.DailyVolumeSnapshots = kept changed = true } // 历史遗留修正:**过去的天不可能"还在进行中"**。 // 早期版本把「起点超出 24h」也当成 partial,于是昨天/前天的柱子一直是虚线空心, // 看起来像"每天都在采样中"。只有当快照是在那天**结束之后**才抓的 // (说明抓的是完整一天)才改成完整;当天中途抓到的继续保持 partial —— 那时确实只有半天。 for i := range db.DailyVolumeSnapshots { snap := &db.DailyVolumeSnapshots[i] if !snap.Partial || snap.Date >= today { continue } day, err := time.ParseInLocation("2006-01-02", snap.Date, time.Local) if err != nil { continue } if _, dayEnd := dayRange(day); snap.CapturedAt.UnixNano() >= dayEnd { snap.Partial = false changed = true } } if changed { db.RequestSave() } } // 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 } if days > dailyVolumeRetentionDays { days = dailyVolumeRetentionDays } // 始终返回 days 个点(缺失的那天标 Missing)。 // 之前只返回"已抓到的快照",于是 7/30/90/365 全都返回同样 4 个点, // 用户切范围看不到任何变化,以为"切换没反应"。 now := time.Now() byDate := map[string]models.DailyVolumeSnapshot{} db := repository.GetDB() if db != nil { db.RLock() for _, snap := range db.DailyVolumeSnapshots { if snap.SourceID != sourceID || snap.Namespace != "" { continue } byDate[snap.Date] = snap } db.RUnlock() } points := make([]DailyVolumePoint, 0, days) for i := days - 1; i >= 0; i-- { date := dateKey(now.AddDate(0, 0, -i)) if snap, ok := byDate[date]; ok { points = append(points, DailyVolumePoint{ Date: snap.Date, Bytes: snap.Bytes, Entries: snap.Entries, Partial: snap.Partial, FromHistory: true, }) continue } points = append(points, DailyVolumePoint{Date: date, Missing: true, FromHistory: true}) } return points } // StartDailyVolumeCapture 后台每日抓取。 // // 每 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) defer ticker.Stop() for range ticker.C { CaptureRecentDailyVolumes() } }() }