refactor(analysis): 去掉排行 10/30/50 选择器,改为一次取 50 + 前端分页;零点结算昨天
CI / Backend (go vet + test) (push) Canceled after 0s
CI / Frontend (lint + build + test) (push) Canceled after 0s

=== 1. 为什么去掉「显示条数」选择器 ===
topk 的 N 不影响查询成本(扫描的 chunk 集合完全相同),实测:
  topk(10)  17833ms / 11 条
  topk(30)  18019ms / 34 条
  topk(50)   7394ms / 54 条
  topk(100) 17553ms / 74 条
波动来自集群负载,与 N 无关 —— 它只决定最后返回多少序列。
所以"选 30"既不更快也不更慢,本质是个翻页开关;
而选它还要重新等一次十几秒的查询,纯亏。
改为:一次 topk(50),前端 10 条/页 → 5 页只需 1 次查询。
附带收益:占比的分母从"前 10 合计"变成"前 50 合计",更接近真实占比。
副标题明确写清"一次取前 50 个应用(占比按这 50 个合计计算)"。

=== 2. 零点结算昨天(daily volume)===
原来只有 30 分钟 ticker 回补,且启动时间不对齐整点 →
昨天最早的一段可能已经超出 Loki 24h 保留期被删,
结算结果系统性偏低,图上却标"完整一天",属静默错数。
新增 nextMidnightFinalize():每天 00:01 触发 FinalizeYesterdayVolume()
  - 昨天定稿 partial=false
  - 补一次条数(index/stats,一天一次可接受)
  - 顺带刷新今天 partial=true 的柱子
ticker 仍保留,负责当天数据的 30 分钟刷新。

=== 3. 顺带 ===
- 去掉每日量"其中 N 天平台没有留存记录"的提示文案(按用户要求)
- 修 QuickQueryTemplates.tsx 的未使用变量导致 tsc -b 失败
- 工作区里上一轮中断的"排行体积列"改动一并纳入(app-ranking 返回 bytes,
  TopAppsTable 增加体积列,快照记录 per-app 字节数)

Pod = Hub latest = a16978d2f88b
This commit is contained in:
opsx committed 2026-10-30 17:13:18 +00:00
1 parent 63dd491c5b
commit d4422f1541
22 files changed
+715 -117

No files matched your search

+54 -2
View File
@@ -349,8 +349,60 @@ func (ctrl *LogsController) AppRanking(c *gin.Context) {
limit = v
}
}
rows := services.GetAppRanking(uint(sourceID), time.Unix(0, from), time.Unix(0, to), limit)
utils.Success(c, gin.H{"list": rows, "hours": services.AppVolumeCoverage(uint(sourceID))})
rows := services.GetAppRanking(uint(sourceID), time.Unix(0, from), time.Unix(0, to), limit, c.DefaultQuery("sort", "bytes"))
hours, bytesHours := services.AppVolumeCoverageInfo(uint(sourceID))
utils.Success(c, gin.H{"list": rows, "hours": hours, "bytesHours": bytesHours})
}
// LabelValuesBatch 一次取多个 label 的候选值:GET /logs/label-values?names=app,pod&query={...}
//
// 快捷查询的下拉是级联的(命名空间 → 应用 → Pod),而"应用"本身要走
// app → service_name → container 三级回退。逐个请求时前端一次要发 3 个往返,
// 每个都要经 K8s API 代理到 Loki 索引;合并成一个接口后只发一次。
// 返回 {"app": [...], "service_name": [...], ...},单个 label 失败返回空数组不报错。
func (ctrl *LogsController) LabelValuesBatch(c *gin.Context) {
id, _ := strconv.ParseUint(c.Query("sourceId"), 10, 64)
src := services.GetLogSource(uint(id))
if src == nil {
utils.BadRequest(c, "数据源不存在")
return
}
raw := strings.TrimSpace(c.Query("names"))
if raw == "" {
utils.BadRequest(c, "缺少 names(逗号分隔的 label 名,如 app,pod)")
return
}
names := make([]string, 0, 4)
for _, name := range strings.Split(raw, ",") {
if name = strings.TrimSpace(name); name != "" {
names = append(names, name)
}
}
if len(names) == 0 {
utils.BadRequest(c, "names 里没有有效的 label 名")
return
}
if len(names) > 10 {
utils.BadRequest(c, "一次最多取 10 个 label 的值")
return
}
requestedQuery := strings.TrimSpace(c.Query("query"))
if requestedQuery == "" {
requestedQuery = "{}"
}
query, err := services.AuthorizedLogQuery(src, c.GetUint("userId"), c.GetString("role") == "admin", requestedQuery)
if err != nil {
utils.Forbidden(c, err.Error())
return
}
start, _ := strconv.ParseInt(c.Query("start"), 10, 64)
end, _ := strconv.ParseInt(c.Query("end"), 10, 64)
values, err := services.LokiLabelValuesBatch(c.Request.Context(), src, names, query, start, end)
if err != nil {
utils.Error(c, 502, err.Error())
return
}
utils.Success(c, values)
}
// DailyVolume 每日日志量历史(平台留存,不受 Loki retention 限制)。
+9 -1
View File
@@ -26,7 +26,15 @@ type AppVolumeSnapshot struct {
SourceID uint `json:"sourceId"`
Namespace string `json:"namespace"`
App string `json:"app"`
Entries int64 `json:"entries"`
// Entries 该小时的日志条数(count_over_time,要读日志内容,成本高但准确)
Entries int64 `json:"entries"`
// Bytes 该小时的日志体积(字节,来自 index/volume 索引接口)
//
// 为什么不跟 Entries 用同一条查询:按 app 取体积在 Loki 里没有"一次扫全集群"的
// 索引聚合(index/volume 的返回只带 selector 里出现的 label),所以只能一个 app
// 调一次;换来的是 80ms/次的索引查询,而不是重扫整小时日志内容(约 50s)。
// 老快照(本字段上线前抓的)Bytes=0,前端据此把"体积未知"和"体积真的是 0"分开。
Bytes int64 `json:"bytes"`
// Partial 该小时数据是否不完整(如平台停了一段时间后才回补,超出 Loki 保留期)
Partial bool `json:"partial"`
CapturedAt time.Time `json:"capturedAt"`
+3
View File
@@ -96,6 +96,9 @@ func SetupRouter() *gin.Engine {
logs.POST("/context", logsCtrl.Context)
logs.GET("/labels", logsCtrl.Labels)
logs.GET("/label/:name/values", logsCtrl.LabelValues)
// 批量取 label 值:一次请求拿多个 label(app+service_name+container 三级回退),
// 前端下拉联动从 3 个往返降到 1 个
logs.GET("/label-values", logsCtrl.LabelValuesBatch)
logs.GET("/my-scope", logsCtrl.MyScope)
// ⚠️ 必须是 POST:前端 logsApi.histogram 用 POST 传 body(sourceId/query/start/end)。
// 曾被误改成 GET,导致 POST /logs/histogram 全部 404,
+2
View File
@@ -34,6 +34,8 @@ func TestCriticalRoutes(t *testing.T) {
{"GET", "/api/v1/logs/sources"},
{"GET", "/api/v1/logs/labels"},
{"GET", "/api/v1/logs/label/:name/values"},
// 批量 label 值(下拉联动一次请求)—— 路由写错就是前端全部下拉空
{"GET", "/api/v1/logs/label-values"},
{"GET", "/api/v1/logs/my-scope"},
{"GET", "/api/v1/logs/daily-volume"},
{"GET", "/api/v1/logs/health"},
+97 -12
View File
@@ -8,6 +8,7 @@ import (
"net/url"
"sort"
"strconv"
"strings"
"time"
"lokistack/models"
@@ -36,7 +37,11 @@ const (
type AppRankRow struct {
App string `json:"app"`
Namespace string `json:"namespace"`
Entries int64 `json:"entries"`
// Entries 日志条数(预聚合快照累加)
Entries int64 `json:"entries"`
// Bytes 日志体积(字节,预聚合快照累加)。
// 快照里没有体积的早期小时会让这个值偏小 —— 响应里的 bytesHours 用来提示覆盖率。
Bytes int64 `json:"bytes"`
// FromHistory 该行来自平台预聚合(而不是实时扫日志)
FromHistory bool `json:"fromHistory"`
}
@@ -86,6 +91,7 @@ func CaptureAppVolumeForHour(src *models.LogSource, hour time.Time) error {
type row struct {
ns, app string
entries int64
bytes int64
}
rows := make([]row, 0, len(resp.Data.Result))
for _, item := range resp.Data.Result {
@@ -114,6 +120,29 @@ func CaptureAppVolumeForHour(src *models.LogSource, hour time.Time) error {
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("数据库未初始化")
@@ -128,6 +157,7 @@ func CaptureAppVolumeForHour(src *models.LogSource, hour time.Time) error {
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
@@ -138,7 +168,7 @@ func CaptureAppVolumeForHour(src *models.LogSource, hour time.Time) error {
}
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(),
Entries: r.entries, Bytes: r.bytes, CapturedAt: time.Now(),
})
nextID++
}
@@ -213,8 +243,29 @@ func StartAppVolumeCapture() {
}()
}
// 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 从**预聚合快照**里取指定时间范围的应用排行(毫秒级)。
func GetAppRanking(sourceID uint, from, to time.Time, limit int) []AppRankRow {
//
// sortBy:bytes(默认,按体积)| entries(按条数)。
// 体积优先是因为用户要的是"哪个应用把磁盘/带宽吃了";条数受日志长短影响,
// 一个高频 INFO 应用可以条数第一但体积很小。
// 排序永远带兜底:体积相同(或都是 0)时按条数,再按应用名,保证顺序稳定。
func GetAppRanking(sourceID uint, from, to time.Time, limit int, sortBy string) []AppRankRow {
if limit <= 0 {
limit = 10
}
@@ -231,6 +282,7 @@ func GetAppRanking(sourceID uint, from, to time.Time, limit int) []AppRankRow {
type agg struct {
ns, app string
entries int64
bytes int64
}
sum := map[string]*agg{}
for _, snap := range db.AppVolumeSnapshots {
@@ -244,16 +296,34 @@ func GetAppRanking(sourceID uint, from, to time.Time, limit int) []AppRankRow {
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, FromHistory: true})
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 {
if rows[i].Entries == rows[j].Entries {
return rows[i].App < rows[j].App
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 rows[i].Entries > rows[j].Entries
return a.App < b.App
})
if len(rows) > limit {
rows = rows[:limit]
@@ -263,17 +333,32 @@ func GetAppRanking(sourceID uint, from, to time.Time, limit int) []AppRankRow {
// 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
return 0, 0
}
db.RLock()
defer db.RUnlock()
hours := map[string]bool{}
all := map[string]bool{}
withBytes := map[string]bool{}
for _, snap := range db.AppVolumeSnapshots {
if snap.SourceID == sourceID {
hours[snap.Hour] = true
if snap.SourceID != sourceID {
continue
}
all[snap.Hour] = true
if snap.Bytes > 0 {
withBytes[snap.Hour] = true
}
}
return len(hours)
return len(all), len(withBytes)
}
+78 -7
View File
@@ -170,10 +170,10 @@ func upsertVolumeSnapshot(sourceID uint, date string, bytes int64, partial bool)
// 实测 6ms(空闲)~ 20s(有并发查询时)。每 30 分钟 × 每一天都重跑一遍,
// 既浪费 Loki 的 CPU,又会和用户页面上的查询抢同一台单实例的算力。所以:
//
// • 只处理 Date < 今天 的天(今天没结束,条数没有意义)
// • 每天最多 entriesMaxAttempts 次,两次间隔 ≥ entriesRetryInterval
// • 尝试次数**先写库再查**:进程中途被杀也不会退化成"每 30 分钟重试"
// • 拿不到就留 0,前端图只画体积,不影响可读性
// - 只处理 Date < 今天 的天(今天没结束,条数没有意义)
// - 每天最多 entriesMaxAttempts 次,两次间隔 ≥ entriesRetryInterval
// - 尝试次数**先写库再查**:进程中途被杀也不会退化成"每 30 分钟重试"
// - 拿不到就留 0,前端图只画体积,不影响可读性
func captureDailyEntries(source *models.LogSource) {
if source == nil {
return
@@ -311,16 +311,31 @@ func pruneDailyVolumeSnapshots() {
today := dateKey(time.Now())
db.Lock()
defer db.Unlock()
known := make(map[uint]bool, len(db.LogSources))
for _, src := range db.LogSources {
known[src.ID] = true
}
changed := false
kept := make([]models.DailyVolumeSnapshot, 0, len(db.DailyVolumeSnapshots))
droppedOrphan := 0
for _, snap := range db.DailyVolumeSnapshots {
// 超过留存期
if snap.Date < cutoff {
continue
}
// 数据源已被删除:它的快照没有任何入口能查到,留着只会混淆取证
// (实测遇到过库里留着 id=2 的历史,查"到底抓没抓"时白查一轮)
if !known[snap.SourceID] {
droppedOrphan++
continue
}
kept = append(kept, snap)
}
if droppedOrphan > 0 {
log.Printf("[daily-volume] 清理已删除数据源的快照 %d 条", droppedOrphan)
}
if len(kept) != len(db.DailyVolumeSnapshots) {
log.Printf("[daily-volume] 清理过期快照 %d 条", len(db.DailyVolumeSnapshots)-len(kept))
log.Printf("[daily-volume] 清理过期快照 %d 条", len(db.DailyVolumeSnapshots)-len(kept)-droppedOrphan)
db.DailyVolumeSnapshots = kept
changed = true
}
@@ -414,10 +429,66 @@ func StartDailyVolumeCapture() {
log.Printf("[daily-volume] 启动回补最近 %d 天", dailyVolumeBackfillDays)
CaptureRecentDailyVolumes()
// 两条节奏:
// 30 分钟 ticker → 刷新"今天"的柱子(当天数据一直在涨)
// 零点后 1 分钟 → **结算昨天**(把它定稿成完整一天)
//
// 为什么零点结算必须对齐而不能靠 30 分钟 ticker:
// Loki 保留期默认 24h,昨天 00:00 的数据在今天 00:00 之后就开始被删。
// 如果等到 00:30 那次 ticker 才抓,昨天最早的 30 分钟已经没了,
// 结算结果会系统性偏低,而图上又标着"完整一天",属于静默错数。
// 00:01 结算时最多只丢 1 分钟,误差可忽略。
ticker := time.NewTicker(30 * time.Minute)
defer ticker.Stop()
for range ticker.C {
CaptureRecentDailyVolumes()
midnight := time.NewTimer(time.Until(nextMidnightFinalize()))
defer midnight.Stop()
for {
select {
case <-ticker.C:
CaptureRecentDailyVolumes()
case <-midnight.C:
FinalizeYesterdayVolume()
midnight.Reset(time.Until(nextMidnightFinalize()))
}
}
}()
}
// nextMidnightFinalize 返回下一个"零点后 1 分钟"的时间点。
func nextMidnightFinalize() time.Time {
now := time.Now()
next := time.Date(now.Year(), now.Month(), now.Day()+1, 0, 1, 0, 0, time.Local)
return next
}
// FinalizeYesterdayVolume 零点后结算昨天:把昨天定稿(partial=false),
// 同时刷新今天的 partial=true 柱子。多数据源逐个处理,单个失败不影响其它。
func FinalizeYesterdayVolume() {
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()
yesterday := now.AddDate(0, 0, -1)
for _, src := range sources {
src := src
if err := CaptureDailyVolume(&src, yesterday, false); err != nil {
log.Printf("[daily-volume] 零点结算 %s %s 失败: %v", src.Name, dateKey(yesterday), err)
continue
}
log.Printf("[daily-volume] 零点结算完成 %s %s", src.Name, dateKey(yesterday))
// 昨天定稿后,条数也可以补一次(走 index/stats,慢但一天一次可接受)
captureDailyEntries(&src)
if err := CaptureDailyVolume(&src, now, true); err != nil {
log.Printf("[daily-volume] 刷新今天 %s 失败: %v", src.Name, err)
}
}
}
@@ -0,0 +1,102 @@
package services
import (
"fmt"
"net/http"
"net/http/httptest"
"sync/atomic"
"testing"
"time"
"lokistack/models"
"lokistack/repository"
)
// 条数(index/stats)只给"已结束且还没有条数"的天补,并且每天最多 3 次。
// 这条护栏的直接目的:别让 30 分钟一次的定时任务把整天 index/stats 反复重跑。
func TestCaptureDailyEntriesOnlyForPastDaysAndWrites(t *testing.T) {
setupDailyVolumeDB(t)
var calls int32
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
atomic.AddInt32(&calls, 1)
if r.URL.Path != "/loki/api/v1/index/stats" {
t.Errorf("意外请求路径: %s", r.URL.Path)
}
w.Header().Set("Content-Type", "application/json")
fmt.Fprint(w, `{"streams":3,"chunks":9,"bytes":123456,"entries":456}`)
}))
defer srv.Close()
src := &models.LogSource{ID: 1, Name: "test", TargetType: "url", URL: srv.URL}
yesterday := dateKey(time.Now().AddDate(0, 0, -1))
today := dateKey(time.Now())
db := repository.GetDB()
db.Lock()
db.DailyVolumeSnapshots = []models.DailyVolumeSnapshot{
{ID: 1, Date: yesterday, SourceID: 1, Bytes: 100},
{ID: 2, Date: today, SourceID: 1, Bytes: 10, Partial: true},
}
db.Unlock()
captureDailyEntries(src)
got := snapshotsByDate(t)
if snap := got[yesterday]; snap.Entries != 456 {
t.Fatalf("昨天的条数应写入 456,得到 %d", snap.Entries)
}
if snap := got[yesterday]; snap.EntriesAttempts != 1 {
t.Errorf("昨天应记录 1 次尝试,得到 %d", snap.EntriesAttempts)
}
if snap := got[today]; snap.EntriesAttempts != 0 || snap.Entries != 0 {
t.Errorf("今天没结束,不应抓条数(attempts=%d entries=%d)", snap.EntriesAttempts, snap.Entries)
}
if n := atomic.LoadInt32(&calls); n != 1 {
t.Errorf("应只打 1 次 index/stats,实际 %d 次", n)
}
// 1 小时内重复触发不应再打 Loki
captureDailyEntries(src)
if n := atomic.LoadInt32(&calls); n != 1 {
t.Errorf("重试间隔内不应再查询,实际 %d 次", n)
}
}
// 失败(Loki 报错)时必须吃得下,并且最多尝试 entriesMaxAttempts 次就放弃。
func TestCaptureDailyEntriesRetryBudget(t *testing.T) {
setupDailyVolumeDB(t)
var calls int32
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
atomic.AddInt32(&calls, 1)
http.Error(w, "boom", http.StatusInternalServerError)
}))
defer srv.Close()
src := &models.LogSource{ID: 1, Name: "test", TargetType: "url", URL: srv.URL}
yesterday := dateKey(time.Now().AddDate(0, 0, -1))
db := repository.GetDB()
db.Lock()
db.DailyVolumeSnapshots = []models.DailyVolumeSnapshot{{ID: 1, Date: yesterday, SourceID: 1, Bytes: 100}}
db.Unlock()
// 模拟"每轮定时任务都到点":把上次尝试时间清空,绕过 1 小时间隔
for i := 0; i < entriesMaxAttempts+3; i++ {
captureDailyEntries(src)
db.Lock()
for j := range db.DailyVolumeSnapshots {
db.DailyVolumeSnapshots[j].LastEntriesAttempt = time.Time{}
}
db.Unlock()
}
if n := atomic.LoadInt32(&calls); n != entriesMaxAttempts {
t.Errorf("应恰好尝试 %d 次后放弃,实际 %d 次", entriesMaxAttempts, n)
}
if snap := snapshotsByDate(t)[yesterday]; snap.Entries != 0 || snap.EntriesAttempts != entriesMaxAttempts {
t.Errorf("失败后应留 entries=0 / attempts=%d,得到 entries=%d attempts=%d",
entriesMaxAttempts, snap.Entries, snap.EntriesAttempts)
}
}
+20
View File
@@ -17,6 +17,15 @@ func setupDailyVolumeDB(t *testing.T) {
t.Fatal(err)
}
t.Cleanup(func() { _ = repository.Shutdown() })
// 每天的快照都挂着数据源,prune 会清掉"数据源已删除"的孤儿快照,
// 所以测试里先登记好 id=1 的数据源。
db := repository.GetDB()
db.Lock()
if len(db.LogSources) == 0 {
db.LogSources = append(db.LogSources, models.LogSource{ID: 1, Name: "test-loki", TargetType: "url"})
}
db.Unlock()
}
func snapshotsByDate(t *testing.T) map[string]models.DailyVolumeSnapshot {
@@ -56,6 +65,8 @@ func TestPruneDailyVolumeNormalizesPartialAndDropsExpired(t *testing.T) {
{ID: 3, Date: dateKey(now), SourceID: 1, Bytes: 10, Partial: true},
// 超过留存期 → 必须被清理
{ID: 4, Date: dateKey(expired), SourceID: 1, Bytes: 999},
// 数据源已删除的孤儿快照 → 必须被清理
{ID: 5, Date: dateKey(yesterday), SourceID: 99, Bytes: 888},
}
db.Unlock()
@@ -65,6 +76,15 @@ func TestPruneDailyVolumeNormalizesPartialAndDropsExpired(t *testing.T) {
if _, ok := got[dateKey(expired)]; ok {
t.Error("超过留存期的快照没有被清理")
}
// 孤儿快照(SourceID=99 的数据源已不存在)必须被清掉
db = repository.GetDB()
db.RLock()
for _, snap := range db.DailyVolumeSnapshots {
if snap.SourceID == 99 {
t.Error("数据源已删除的孤儿快照没有被清理")
}
}
db.RUnlock()
if snap := got[dateKey(yesterday)]; snap.Partial {
t.Error("昨天结束之后抓的快照应标记为完整(partial=false)")
}
+107 -1
View File
@@ -409,6 +409,109 @@ func LokiLabelValues(ctx context.Context, src *models.LogSource, name, selector
return resp.Data, nil
}
// LokiLabelValuesBatch 一次取多个 label 的候选值。
//
// 前端快捷查询要对 app / service_name / container 做逐级回退(哪个有值用哪个),
// 原来最多并发 3 个请求、每个请求各走一次 Loki 索引查询 + 一次 kube-apiserver
// service proxy;合并成本接口后前端只发一次,后端内部并发(上限 4)。
//
// 语义:单个 label 失败不影响其它 label(返回空数组),**全部失败**才报错 ——
// 否则某个 label 在特定 selector 下索引异常,整个下拉框都会变成"加载失败"。
func LokiLabelValuesBatch(ctx context.Context, src *models.LogSource, names []string, selector string, start, end int64) (map[string][]string, error) {
out := make(map[string][]string, len(names))
var (
mu sync.Mutex
wg sync.WaitGroup
firstErr error
okCount int
seen = make(map[string]bool, len(names))
)
sem := make(chan struct{}, 4)
for _, raw := range names {
name := strings.TrimSpace(raw)
if name == "" || seen[name] {
continue
}
seen[name] = true
out[name] = []string{}
wg.Add(1)
go func(labelName string) {
defer wg.Done()
sem <- struct{}{}
defer func() { <-sem }()
values, err := LokiLabelValues(ctx, src, labelName, selector, start, end)
mu.Lock()
defer mu.Unlock()
if err != nil {
if firstErr == nil {
firstErr = err
}
return
}
okCount++
out[labelName] = values
}(name)
}
wg.Wait()
if okCount == 0 && firstErr != nil {
return out, firstErr
}
return out, nil
}
// LokiVolumeBytesFor 用 index/volume 读某个 selector 的日志体积(字节)。
//
// 这是 Loki 的**索引接口**:只读 TSDB 里记录的 chunk 字节数,不解码日志内容。
// 实测单个 app、一小时窗口约 80ms(对比 bytes_over_time 同窗口 13s/15 分钟、
// 约 50s/小时 —— 因为它要把整小时的日志解压出来数行字节)。
//
// 代价:index/volume 返回的 metric 只包含 selector 里出现过的 label,
// 所以"按 app 取体积"没法一条查询拿全,必须一个 app 调一次(见 app_volume.go)。
// aggregateBy=series 让返回值带上 selector 里的全部 label,便于调用方核对。
func LokiVolumeBytesFor(ctx context.Context, src *models.LogSource, selector string, start, end int64) (int64, error) {
if end <= 0 {
end = time.Now().UnixNano()
}
if start <= 0 {
start = end - int64(time.Hour)
}
ctx, cancel := context.WithTimeout(ctx, 20*time.Second)
defer cancel()
q := url.Values{}
q.Set("query", selector)
q.Set("start", strconv.FormatInt(start, 10))
q.Set("end", strconv.FormatInt(end, 10))
q.Set("aggregateBy", "series")
body, _, err := lokiProxyGET(ctx, src, "/loki/api/v1/index/volume", q)
if err != nil {
return 0, 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 0, fmt.Errorf("解析 volume 响应失败: %w", err)
}
var total int64
for _, item := range resp.Data.Result {
switch v := item.Value[1].(type) {
case float64:
total += int64(v)
case string:
if n, convErr := strconv.ParseFloat(v, 64); convErr == nil {
total += int64(n)
}
}
}
return total, nil
}
// extractStreamSelector 从 LogQL 中提取 {...} 部分(labels API 仅接受 stream selector)
func extractStreamSelector(query string) string {
i := strings.Index(query, "{")
@@ -826,9 +929,12 @@ func pickHistogramStep(spanNano int64) string {
return "1m"
case span <= 6*time.Hour:
return "5m"
case span <= 24*time.Hour:
case span <= 12*time.Hour:
return "15m"
case span <= 7*24*time.Hour:
// 12 小时以上统一 1 小时一桶。
// 24 小时原来是 15m(96 根柱子):横轴挤成一片、且每根柱子只有十来分钟的数据,
// 看不出趋势;1h → 24 根,和"按小时看日志量"的心智一致。
return "1h"
default:
return "6h"