LokiStack/backend/services/app_volume.go

package services

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/url"
	"sort"
	"strconv"
	"time"

	"lokistack/models"
	"lokistack/repository"
)

// 按应用的日志量：**小时级预聚合**。
//
// 背景：按应用排行必须读日志内容（Loki 3.4 的 index/volume 不支持 app 维度聚合），
// 成本 ∝ 时间范围 —— 5 分钟 10 秒、15 分钟 20 秒、6 小时要 5~8 分钟、24 小时不可用。
//
// 方案：后台每小时把「上一个完整小时」的 per-app 条数算好落库；
// 页面查 6h/24h/7d 时直接读库（毫秒级），不再扫日志。
// 前台只在「≤1 小时」的短范围走实时查询（那种范围本来就快）。

const (
	// appVolumeRetentionDays 小时级 per-app 快照保留天数
	appVolumeRetentionDays = 90
	// appVolumeSampleTopN 每个小时只保留条数最多的 N 个应用，避免快照爆炸
	appVolumeSampleTopN = 50
	// appVolumeTimeout 单次抓取的超时（1 小时窗口 ≈ 15GB，给足时间）
	appVolumeTimeout = 5 * time.Minute
)

// AppRankRow 排行结果的一行
type AppRankRow struct {
	App       string `json:"app"`
	Namespace string `json:"namespace"`
	Entries   int64  `json:"entries"`
	// FromHistory 该行来自平台预聚合（而不是实时扫日志）
	FromHistory bool `json:"fromHistory"`
}

// hourKey 把时间对齐到整点，格式 YYYY-MM-DDTHH（本地时区，与快照一致）
func hourKey(t time.Time) string {
	return t.Local().Format("2006-01-02T15")
}

// CaptureAppVolumeForHour 抓取指定小时（整点到整点）的 per-app 条数。
func CaptureAppVolumeForHour(src *models.LogSource, hour time.Time) error {
	if src == nil {
		return fmt.Errorf("数据源为空")
	}
	start := hour.Truncate(time.Hour)
	end := start.Add(time.Hour)
	key := hourKey(start)

	ctx, cancel := context.WithTimeout(context.Background(), appVolumeTimeout)
	defer cancel()

	// 用**瞬时查询**而不是 query_range：
	//   - 只评估一个时间点（窗口覆盖整个小时），扫描量与 query_range 相同
	//     但省掉多步评估的开销
	//   - 直接走 lokiProxyGET + 自己 5 分钟超时，绕开 LokiMetrics 的 90 秒
	//     上限与重型查询信号量（后台任务不该和前台查询抢令牌）
	query := fmt.Sprintf(`sum by (app, namespace) (count_over_time({namespace=~".+", app!="loki"}[%dm]))`, int(time.Hour.Minutes()))
	q := url.Values{}
	q.Set("query", query)
	q.Set("time", fmt.Sprintf("%d", end.UnixNano()))
	body, _, err := lokiProxyGET(ctx, src, "/loki/api/v1/query", q)
	if err != nil {
		return err
	}
	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 fmt.Errorf("解析 Loki 响应失败: %w", err)
	}

	type row struct {
		ns, app string
		entries int64
	}
	rows := make([]row, 0, len(resp.Data.Result))
	for _, item := range resp.Data.Result {
		app := item.Metric["app"]
		if app == "" {
			app = item.Metric["service_name"]
		}
		if app == "" {
			app = item.Metric["container"]
		}
		if app == "" {
			continue
		}
		value, _ := strconv.ParseFloat(fmt.Sprintf("%v", item.Value[1]), 64)
		if value <= 0 {
			continue
		}
		rows = append(rows, row{ns: item.Metric["namespace"], app: app, entries: int64(value)})
	}
	// 只留 top N，控制快照体积
	sort.Slice(rows, func(i, j int) bool { return rows[i].entries > rows[j].entries })
	if len(rows) > appVolumeSampleTopN {
		rows = rows[:appVolumeSampleTopN]
	}
	if len(rows) == 0 {
		return nil
	}

	db := repository.GetDB()
	if db == nil {
		return fmt.Errorf("数据库未初始化")
	}
	db.Lock()
	defer db.Unlock()
	nextID := db.NextID("appVolumeSnapshots")
	for _, r := range rows {
		// 幂等：同 (hour, source, ns, app) 已存在则更新，不重复插入
		found := false
		for i := range db.AppVolumeSnapshots {
			snap := &db.AppVolumeSnapshots[i]
			if snap.Hour == key && snap.SourceID == src.ID && snap.Namespace == r.ns && snap.App == r.app {
				snap.Entries = r.entries
				snap.CapturedAt = time.Now()
				found = true
				break
			}
		}
		if found {
			continue
		}
		db.AppVolumeSnapshots = append(db.AppVolumeSnapshots, models.AppVolumeSnapshot{
			ID: nextID, Hour: key, SourceID: src.ID, Namespace: r.ns, App: r.app,
			Entries: r.entries, CapturedAt: time.Now(),
		})
		nextID++
	}
	pruneAppVolumeLocked(db)
	db.RequestSave()
	log.Printf("[app-volume] %s 已留存 %d 个应用的量", key, len(rows))
	return nil
}

// pruneAppVolumeLocked 按保留期裁剪（调用方必须已持写锁）。
func pruneAppVolumeLocked(db *repository.JsonDB) {
	cutoff := hourKey(time.Now().AddDate(0, 0, -appVolumeRetentionDays))
	keep := db.AppVolumeSnapshots[:0]
	for _, snap := range db.AppVolumeSnapshots {
		if snap.Hour >= cutoff {
			keep = append(keep, snap)
		}
	}
	db.AppVolumeSnapshots = keep
}

// CaptureMissingAppVolume 回补最近 N 个**完整小时**里还没抓过的那些。
func CaptureMissingAppVolume(hours int) {
	if hours <= 0 {
		hours = 2
	}
	db := repository.GetDB()
	if db == nil {
		return
	}
	db.RLock()
	sources := make([]models.LogSource, 0, len(db.LogSources))
	for _, s := range db.LogSources {
		sources = append(sources, s)
	}
	captured := map[string]bool{}
	for _, snap := range db.AppVolumeSnapshots {
		captured[fmt.Sprintf("%d|%s", snap.SourceID, snap.Hour)] = true
	}
	db.RUnlock()

	now := time.Now()
	for _, src := range sources {
		for i := 1; i <= hours; i++ {
			hour := now.Add(-time.Duration(i) * time.Hour).Truncate(time.Hour)
			if captured[fmt.Sprintf("%d|%s", src.ID, hourKey(hour))] {
				continue
			}
			// 超出 Loki 保留期的窗口不抓（抓也抓不到，白等几分钟）
			if time.Since(hour.Add(time.Hour)) > 26*time.Hour {
				continue
			}
			if err := CaptureAppVolumeForHour(&src, hour); err != nil {
				log.Printf("[app-volume] %s %s 抓取失败: %v", src.Name, hourKey(hour), err)
			}
		}
	}
}

// StartAppVolumeCapture 后台任务：每小时第 5 分钟抓上一个完整小时。
func StartAppVolumeCapture() {
	go func() {
		time.Sleep(45 * time.Second) // 避开启动时的其它任务
		CaptureMissingAppVolume(2)
		for {
			now := time.Now()
			// 下一个整点后 5 分钟
			next := now.Truncate(time.Hour).Add(time.Hour + 5*time.Minute)
			time.Sleep(time.Until(next))
			CaptureMissingAppVolume(3)
		}
	}()
}

// GetAppRanking 从**预聚合快照**里取指定时间范围的应用排行（毫秒级）。
func GetAppRanking(sourceID uint, from, to time.Time, limit int) []AppRankRow {
	if limit <= 0 {
		limit = 10
	}
	fromHour := hourKey(from)
	toHour := hourKey(to)

	db := repository.GetDB()
	if db == nil {
		return nil
	}
	db.RLock()
	defer db.RUnlock()

	type agg struct {
		ns, app string
		entries int64
	}
	sum := map[string]*agg{}
	for _, snap := range db.AppVolumeSnapshots {
		if snap.SourceID != sourceID || snap.Hour < fromHour || snap.Hour > toHour {
			continue
		}
		key := snap.Namespace + "\x00" + snap.App
		item, ok := sum[key]
		if !ok {
			item = &agg{ns: snap.Namespace, app: snap.App}
			sum[key] = item
		}
		item.entries += snap.Entries
	}
	rows := make([]AppRankRow, 0, len(sum))
	for _, item := range sum {
		rows = append(rows, AppRankRow{App: item.app, Namespace: item.ns, Entries: item.entries, FromHistory: true})
	}
	sort.Slice(rows, func(i, j int) bool {
		if rows[i].Entries == rows[j].Entries {
			return rows[i].App < rows[j].App
		}
		return rows[i].Entries > rows[j].Entries
	})
	if len(rows) > limit {
		rows = rows[:limit]
	}
	return rows
}

// AppVolumeCoverage 返回快照覆盖的小时数，前端用来提示"历史只有 N 小时数据"。
func AppVolumeCoverage(sourceID uint) int {
	db := repository.GetDB()
	if db == nil {
		return 0
	}
	db.RLock()
	defer db.RUnlock()
	hours := map[string]bool{}
	for _, snap := range db.AppVolumeSnapshots {
		if snap.SourceID == sourceID {
			hours[snap.Hour] = true
		}
	}
	return len(hours)
}