package services import ( "context" "encoding/json" "fmt" "log" "net/url" "sort" "strconv" "strings" "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 日志条数(预聚合快照累加) Entries int64 `json:"entries"` // Bytes 日志体积(字节,预聚合快照累加)。 // 快照里没有体积的早期小时会让这个值偏小 —— 响应里的 bytesHours 用来提示覆盖率。 Bytes int64 `json:"bytes"` // 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 bytes 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 } // ── 体积:逐个 app 走 index/volume(索引接口,实测 ~80ms/个)── // // 为什么不用一条 `sum by (app) (bytes_over_time(...))` 拿全:那条要重扫整小时 // 的日志内容(本集群约 29GB/小时,15 分钟就要 13s,一小时约 50s),而 // index/volume 只读 chunk 元数据。代价是必须一个 app 一次调用(index/volume // 的返回值只带 selector 里出现过的 label),50 个 app ≈ 4~10s。 // // 体积失败**不影响条数落库**:拿不到就留 0,前端把 0 显示成"体积未知", // 排行仍可按条数排(GetAppRanking 会按可用指标退让)。 vctx, vcancel := context.WithTimeout(context.Background(), 2*time.Minute) for i := range rows { if rows[i].app == "" { continue } bytes, err := LokiVolumeBytesFor(vctx, src, appVolumeSelector(rows[i].ns, rows[i].app), start.UnixNano(), end.UnixNano()) if err != nil { log.Printf("[app-volume] %s %s/%s 体积获取失败(条数已留存): %v", key, rows[i].ns, rows[i].app, err) continue } rows[i].bytes = bytes } vcancel() db := repository.GetDB() if db == nil { return fmt.Errorf("数据库未初始化") } // NextID 内部会自己加锁;必须在 db.Lock() 之前调用。 // JsonDB 的 mutex 不可重入,先 Lock 再 NextID 会永久死锁并占住写锁, // 导致所有读库接口(登录、列表)全部卡死。 nextID := db.NextID("appVolumeSnapshots") db.Lock() defer db.Unlock() 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.Bytes = r.bytes 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, Bytes: r.bytes, 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) } }() } // appVolumeSelector 生成 index/volume 用的 stream selector。 // 值用 strconv.Quote 转义:app 名理论上可能出现引号/反斜杠,拼错就是 Loki 400。 func appVolumeSelector(namespace, app string) string { parts := make([]string, 0, 2) if namespace != "" { parts = append(parts, "namespace="+strconv.Quote(namespace)) } if app != "" { parts = append(parts, "app="+strconv.Quote(app)) } if len(parts) == 0 { return `{namespace=~".+"}` } return "{" + strings.Join(parts, ", ") + "}" } // GetAppRanking 从**预聚合快照**里取指定时间范围的应用排行(毫秒级)。 // // sortBy:bytes(默认,按体积)| entries(按条数)。 // 体积优先是因为用户要的是"哪个应用把磁盘/带宽吃了";条数受日志长短影响, // 一个高频 INFO 应用可以条数第一但体积很小。 // 排序永远带兜底:体积相同(或都是 0)时按条数,再按应用名,保证顺序稳定。 func GetAppRanking(sourceID uint, from, to time.Time, limit int, sortBy string) []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 bytes 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 item.bytes += snap.Bytes } rows := make([]AppRankRow, 0, len(sum)) for _, item := range sum { rows = append(rows, AppRankRow{ App: item.app, Namespace: item.ns, Entries: item.entries, Bytes: item.bytes, FromHistory: true, }) } byEntries := sortBy == "entries" sort.Slice(rows, func(i, j int) bool { a, b := rows[i], rows[j] if byEntries { if a.Entries != b.Entries { return a.Entries > b.Entries } if a.Bytes != b.Bytes { return a.Bytes > b.Bytes } } else { if a.Bytes != b.Bytes { return a.Bytes > b.Bytes } if a.Entries != b.Entries { return a.Entries > b.Entries } } return a.App < b.App }) if len(rows) > limit { rows = rows[:limit] } return rows } // AppVolumeCoverage 返回快照覆盖的小时数,前端用来提示"历史只有 N 小时数据"。 func AppVolumeCoverage(sourceID uint) int { hours, _ := AppVolumeCoverageInfo(sourceID) return hours } // AppVolumeCoverageInfo 返回(快照覆盖小时数, 其中带体积的小时数)。 // // 第二个值是给"按体积排行"用的诚实性提示:体积列是本轮新加的, // 之前抓的小时只有条数没有体积,24h 排行里这些小时的体积会偏小。 // 两个数不一致时前端要明确告诉用户覆盖率,而不是让用户误读成"这个应用体积很小"。 func AppVolumeCoverageInfo(sourceID uint) (hours, bytesHours int) { db := repository.GetDB() if db == nil { return 0, 0 } db.RLock() defer db.RUnlock() all := map[string]bool{} withBytes := map[string]bool{} for _, snap := range db.AppVolumeSnapshots { if snap.SourceID != sourceID { continue } all[snap.Hour] = true if snap.Bytes > 0 { withBytes[snap.Hour] = true } } return len(all), len(withBytes) }