LokiStack/backend/services/daily_volume.go

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 丢弃超过保留期的历史。
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 天，按日期升序返回。
//
// 这个接口**只读库，绝不触发任何 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()
		}
	}()
}