logan
2e97c04f84
- NextID 改用独立 seqMu，不再依赖数据写锁，杜绝 Lock 内调用 NextID 的死锁 - CaptureAppVolumeForHour 先取 ID 再 Lock - 新增写锁看门狗：长时间占用时转储 goroutine 栈 - 新增 TestNextIDWhileWriteLocked 回归测试
368 lines
11 KiB
Go
368 lines
11 KiB
Go
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)
|
||
}
|