package services import ( "context" "log" "sort" "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"` } const ( // dailyVolumeBackfillDays 启动时回补天数。 // 取 2 天:昨天(完整)+ 今天(进行中),足够覆盖"重启导致漏抓"的常见情况。 // 不设更大是因为 Loki retention 通常只有 24h,更早的已经查不到了。 dailyVolumeBackfillDays = 2 // dailyVolumeRetentionDays 平台自己留多久的历史。一年足够看年度趋势。 dailyVolumeRetentionDays = 400 ) // 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() } // CaptureDailyVolume 抓取指定日期的日志量并落库(按天幂等:同一天重复抓会覆盖)。 // // 数据来源是 Loki 的 index/stats(只读索引元数据),实测毫秒级, // 所以就算是补几十天历史也不会拖慢任何东西。 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 } // 回补历史时,如果那天的起点已经超出 Loki 保留期(默认 24h), // Loki 里根本没那么多数据 —— 必须标成"不完整", // 否则图上会把"数据被删了"误读成"那天日志量真的很少"。 if start < time.Now().Add(-24*time.Hour).UnixNano() { partial = true } // 只查总量(不带 namespace 条件),一次拿到该源当天的条数与体积 // 整天(24h)的 index/stats:大集群上比 15 分钟的查询慢得多, // 实测 20s 会超时(这也是之前每日量接口卡 20s 的原因)。给 90s。 ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) defer cancel() stats, err := LokiStatsFor(ctx, source, `{namespace=~".+"}`, start, end) if err != nil { return err } date := dateKey(day) 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 == source.ID && snap.Namespace == "" { snap.Bytes = stats.Bytes snap.Entries = stats.Entries snap.Partial = partial snap.CapturedAt = time.Now() db.RequestSave() return nil } } nextID++ db.DailyVolumeSnapshots = append(db.DailyVolumeSnapshots, models.DailyVolumeSnapshot{ ID: nextID, Date: date, SourceID: source.ID, Namespace: "", Bytes: stats.Bytes, Entries: stats.Entries, Partial: partial, CapturedAt: time.Now(), }) db.RequestSave() return nil } // CaptureRecentDailyVolumes 回补最近 N 天(含今天),供启动时与定时任务调用。 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) } } } pruneDailyVolumeSnapshots() } // pruneDailyVolumeSnapshots 丢弃超过保留期的历史。 func pruneDailyVolumeSnapshots() { db := repository.GetDB() if db == nil { return } cutoff := dateKey(time.Now().AddDate(0, 0, -dailyVolumeRetentionDays)) db.Lock() defer db.Unlock() kept := make([]models.DailyVolumeSnapshot, 0, len(db.DailyVolumeSnapshots)) for _, snap := range db.DailyVolumeSnapshots { if snap.Date >= cutoff { kept = append(kept, snap) } } if len(kept) != len(db.DailyVolumeSnapshots) { log.Printf("[daily-volume] 清理过期快照 %d 条", len(db.DailyVolumeSnapshots)-len(kept)) db.DailyVolumeSnapshots = kept db.RequestSave() } } // GetDailyVolumeHistory 从平台留存的历史里取最近 days 天,按日期升序返回。 // // 今天如果还没抓过,会实时补一次(这样用户切到页面时当天数字是最新的)。 func GetDailyVolumeHistory(sourceID uint, days int) []DailyVolumePoint { if days <= 0 { days = 7 } if days > dailyVolumeRetentionDays { days = dailyVolumeRetentionDays } // 这个接口**只读库**,绝不触发任何 Loki 查询。 // // 事故复盘:这里原来同步 await 一次"整天 index/stats",那次查询超时 20 秒, // 于是 /logs/daily-volume 耗时 20s → 前端超时 → 页面显示"每日日志量没数据"; // 而且这条重查询还和页面的 histogram/stats 抢 Loki,导致那两个接口同时 502。 // 当天那根柱子由后台定时任务(每 30 分钟)补,慢几分钟完全可接受。 from := dateKey(time.Now().AddDate(0, 0, -(days - 1))) points := []DailyVolumePoint{} db := repository.GetDB() if db == nil { return points } db.RLock() defer db.RUnlock() for _, snap := range db.DailyVolumeSnapshots { if snap.SourceID != sourceID || snap.Namespace != "" { continue } if snap.Date < from { continue } points = append(points, DailyVolumePoint{ Date: snap.Date, Bytes: snap.Bytes, Entries: snap.Entries, Partial: snap.Partial, FromHistory: true, }) } sort.Slice(points, func(i, j int) bool { return points[i].Date < points[j].Date }) return points } // StartDailyVolumeCapture 后台每日抓取。 // // 每 30 分钟检查一次:只要发现"昨天"还没落库就立刻补, // 而不是死等 00:10 —— 平台可能整晚都没运行,定时器会错过那个点。 func StartDailyVolumeCapture() { go func() { // 启动先补最近两天(覆盖重启场景) time.Sleep(20 * time.Second) CaptureRecentDailyVolumes() ticker := time.NewTicker(30 * time.Minute) defer ticker.Stop() for range ticker.C { CaptureRecentDailyVolumes() } }() }