feat(analysis,alerts): 每日量落库看历史趋势、趋势图/饼图可读性、告警三分页
=== 1. 每日日志量必须落库(核心问题:Loki 只留 24 小时)===
实测:请求"近 7 天每日量",Loki 只返回 27 个**小时**桶,全部在最近 24h 内 ——
因为 retention_period=24h,更早的 chunk 已经被删了。
也就是说「固定近 7 天」这张卡片,过去一直是"6 天空白 + 1 天有数据",
用户看着像坏了,实际是数据物理不存在。
方案:平台自己留存历史。
- models.DailyVolumeSnapshot{Date(本地YYYY-MM-DD), SourceID, Bytes, Entries, Partial}
- services/daily_volume.go
CaptureDailyVolume 按天抓 index/stats 落库(幂等覆盖)
StartDailyVolumeCapture 启动 20s 后回补 + 每 30 分钟检查一次
GetDailyVolumeHistory 读库返回最近 N 天(今天会实时补抓)
- 保留 400 天(够看年度趋势),按 source+date 去重
- 前端可选 7 / 30 / 90 / 365 天,能看多长取决于平台积累多久
为什么用"每 30 分钟检查昨天有没有落库"而不是固定 00:10 定时:
平台可能整晚没运行,固定时点的定时器会直接错过;检查式回补不依赖平台在线时刻。
Partial 标记(关键,否则会误读):
- 当天未结束 → partial=true
- 待抓取日期的起点已超出 24h 保留期 → partial=true(Loki 里就没那么多数据)
实测:10-27 32.5GB/2.65亿条 partial=true(保留期外),
10-28 348GB/28.56亿条 partial=true(当天进行中)
=== 2. 修一个 Go 副本 bug(自己写的,靠实测发现)===
第一版用 `for _, snap := range db.DailyVolumeSnapshots` 后改 snap 字段,
range 拿到的是**副本**,改动不会写回切片 → 同一天永远保持首次抓取的值,
partial 也永不更新(实测发现 10-27 该是 true 却显示 false)。
改为 `for i := range ... { snap := &db.DailyVolumeSnapshots[i] }`。
=== 3. 趋势图可读性 ===
从"裸折线"换成"渐变面积图",并在图上方给出峰值 / 均值:
- 面积图能一眼看出"这一段整体比上一段高",折线只能靠对齐格子数
- 峰值/均值避免用户自己对刻度
- Y 轴与 tooltip 统一 formatBytes,"每桶体积"写进 subtitle
顺带确认:趋势图只画体积(字节),不再混入 error(条数)——单位不同不能同轴。
=== 4. 饼图:命名空间分布 → 应用日志量分布 ===
本环境的命名空间只有 astralx / log 两个,饼图没有信息量;
改成按应用看谁在刷日志(前 8 名 + 其他),才是运维真正要的答案。
=== 5. 告警页拆成三个 Tab ===
「告警规则 / 通知通道 / 发送历史」独立成分页(用 ui/tabs)。
之前通道是折叠块、历史堆在规则下面,规则一多就要滚很久才看得到。
发送历史增加空状态文案,不再是"空页面不知道有没有在发"。
=== 6. 内置规则关联默认通道 + 首次部署 webhook 留空的评估 ===
结论:**webhook 可以留空,但通道必须是禁用状态**。
- 若种子给的是"已启用 + 空地址"通道,规则一命中就必然投递失败,
发送历史堆满红色失败,反而让人以为平台坏了
- 所以首次部署建一个**禁用**的占位通道「默认通知通道(待配置 Webhook)」,
内置规则全部挂到它;管理员填好 Webhook 再启用即可,不用逐条关联
- 保存时加校验:启用则必须有非空 Webhook(返回明确原因,不是裸 500)
- schemaVersion v5→v6:把已有的"无通道"规则补挂到默认通道
实测:8 条规则全部已关联通道
=== 实测(v11-1028-2252 / 187062644c0f)===
/logs/daily-volume?days=7 → 2 天数据,partial 标记正确
规则 8 条全部有 channelIds
平台 ready、健康 ok
This commit is contained in:
1 parent
cb91470d75
commit
37e2f0febc
20 files changed
+716
-202
No files matched your search
@@ -328,6 +328,22 @@ func (ctrl *LogsController) Health(c *gin.Context) {
|
||||
utils.Success(c, gin.H{"ready": services.LokiReady(ctx, src)})
|
||||
}
|
||||
|
||||
// DailyVolume 每日日志量历史(平台留存,不受 Loki retention 限制)。
|
||||
//
|
||||
// query: sourceId, days(默认 7,最长 400)
|
||||
func (ctrl *LogsController) DailyVolume(c *gin.Context) {
|
||||
sourceID, _ := strconv.Atoi(c.Query("sourceId"))
|
||||
if sourceID <= 0 {
|
||||
utils.BadRequest(c, "缺少 sourceId")
|
||||
return
|
||||
}
|
||||
days, _ := strconv.Atoi(c.Query("days"))
|
||||
if days <= 0 {
|
||||
days = 7
|
||||
}
|
||||
utils.Success(c, services.GetDailyVolumeHistory(uint(sourceID), days))
|
||||
}
|
||||
|
||||
// Stats 返回该查询在时间范围内的「总条数/字节/块数」,用于前端展示总页数。
|
||||
// 走 index/stats(只读索引),成本远低于真数日志。
|
||||
func (ctrl *LogsController) Stats(c *gin.Context) {
|
||||
|
||||
@@ -141,6 +141,8 @@ func main() {
|
||||
repository.GetDB().RecoverOrphanDeployments()
|
||||
|
||||
services.StartAlertEngine()
|
||||
// 每日日志量落库:Loki retention 只有 24h,历史趋势必须自己留
|
||||
services.StartDailyVolumeCapture()
|
||||
defer services.StopAlertEngine()
|
||||
|
||||
r := routers.SetupRouter()
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
package models
|
||||
|
||||
import "time"
|
||||
|
||||
// DailyVolumeSnapshot 每日日志量快照(平台自己留存的历史)。
|
||||
//
|
||||
// 为什么必须落库:
|
||||
// Loki 的 retention_period 是 24 小时(运维可改,但默认就该短),
|
||||
// 意味着「近 7 天每日日志量」直接查 Loki 时,只有最近 1 天有数据,
|
||||
// 其余 6 天是空的 —— 界面上看着像坏了,实际是数据已经不在了。
|
||||
// 想要 30 天 / 1 年趋势,只能在数据被删前每天抓一次存进平台库。
|
||||
//
|
||||
// 粒度说明:
|
||||
// - 只按「数据源 + 命名空间」存。index/volume 系列接口只能按 selector 标签聚合,
|
||||
// 拿不到按 app 的拆分(Loki 3.4 不支持 aggregateBy),
|
||||
// 所以按 app 的历史量暂时无法落库,应用维度请看实时排行(有 20~30s 成本)。
|
||||
// - 每天一条,日期用**本地时区**的 YYYY-MM-DD,避免跨时区对不上。
|
||||
type DailyVolumeSnapshot struct {
|
||||
ID uint `json:"id"`
|
||||
Date string `json:"date"` // YYYY-MM-DD(本地时区)
|
||||
SourceID uint `json:"sourceId"` // 数据源
|
||||
Namespace string `json:"namespace"` // "" 表示该数据源全量
|
||||
Bytes int64 `json:"bytes"` // 日志体积
|
||||
Entries int64 `json:"entries"` // 日志条数
|
||||
// Partial 该日期是否为"不完整的一天"(当天抓取时还没过完)。
|
||||
// 前端用它区分"真实低峰"和"只统计到一半"。
|
||||
Partial bool `json:"partial"`
|
||||
CapturedAt time.Time `json:"capturedAt"`
|
||||
}
|
||||
+66
-30
@@ -40,6 +40,8 @@ type JsonDB struct {
|
||||
LogPolicyGroups []models.LogPolicyGroup `json:"logPolicyGroups"`
|
||||
// AuditLogs 操作审计日志(保留期 90 天 / 上限 2 万条,见 services.auditRetention)
|
||||
AuditLogs []models.AuditLog `json:"auditLogs"`
|
||||
// DailyVolumeSnapshots 每日日志量历史(Loki retention 只有 24h,历史必须自己留)
|
||||
DailyVolumeSnapshots []models.DailyVolumeSnapshot `json:"dailyVolumeSnapshots"`
|
||||
|
||||
// SchemaVersion 数据结构版本号。
|
||||
// 之前加字段全靠 JSON 反序列化的零值兜底,字段一多必然出事
|
||||
@@ -111,6 +113,26 @@ func (d *JsonDB) raiseLegacyAlertThresholds() {
|
||||
}
|
||||
}
|
||||
|
||||
// linkRulesToDefaultChannel 给未关联通道的规则挂上默认通道。
|
||||
//
|
||||
// 没有通道时不凭空造(那是 seed 首次部署的职责);这里只在已有通道时补齐关联。
|
||||
func (d *JsonDB) linkRulesToDefaultChannel() {
|
||||
if len(d.NotifyChannels) == 0 {
|
||||
return
|
||||
}
|
||||
def := d.NotifyChannels[0].ID
|
||||
linked := 0
|
||||
for i := range d.AlertRules {
|
||||
if len(d.AlertRules[i].ChannelIDs) == 0 {
|
||||
d.AlertRules[i].ChannelIDs = []uint{def}
|
||||
linked++
|
||||
}
|
||||
}
|
||||
if linked > 0 {
|
||||
log.Printf("[schema] %d 条告警规则已关联默认通知通道 %q", linked, d.NotifyChannels[0].Name)
|
||||
}
|
||||
}
|
||||
|
||||
// fixUnreachableLogSourceURLs 把数据源里 0.0.0.0 / [::] 的监听地址换成主机真实 IP。
|
||||
//
|
||||
// 平台进程不在被采集主机上,0.0.0.0 只在本机有意义,用它必然连不上。
|
||||
@@ -344,7 +366,7 @@ func (d *JsonDB) dedupeLogSourceIDs() {
|
||||
//
|
||||
// v0/v1(无字段)→ 1:新增 auditLogs / schemaVersion(老库走零值兜底即可,无需处理)
|
||||
// v1 → 2:预留示例(如把 ScrapeJobs 从 logPath 拆分时,需按旧结构生成默认 job)
|
||||
const CurrentSchemaVersion = 5
|
||||
const CurrentSchemaVersion = 6
|
||||
|
||||
// migrateSchema 按版本号顺序迁移老库。返回是否发生了迁移(需要回写)。
|
||||
func (d *JsonDB) migrateSchema() bool {
|
||||
@@ -397,6 +419,13 @@ func (d *JsonDB) migrateSchema() bool {
|
||||
// 只改"仍等于旧默认值"的规则,用户手动调过的阈值不动。
|
||||
d.raiseLegacyAlertThresholds()
|
||||
d.SchemaVersion = 5
|
||||
case 5:
|
||||
// v5 → v6:把"没有关联任何通道"的告警规则挂到默认通道。
|
||||
//
|
||||
// 之前内置规则 ChannelIDs 全是空的,规则命中后只能去回退旧的全局
|
||||
// notify-config;那个早就不用了,结果就是"规则在响但哪里都没发出去"。
|
||||
d.linkRulesToDefaultChannel()
|
||||
d.SchemaVersion = 6
|
||||
default:
|
||||
// 防御:出现未知版本(比当前代码还新)时不乱动,避免降级把数据写坏
|
||||
log.Printf("[schema] 数据库版本 %d 高于当前代码支持版本 %d,跳过迁移", d.SchemaVersion, CurrentSchemaVersion)
|
||||
@@ -412,36 +441,38 @@ func InitDB() error {
|
||||
_ = os.MkdirAll(cfg.DataDir, 0700)
|
||||
|
||||
d := &JsonDB{
|
||||
dataFile: cfg.DatabasePath,
|
||||
Users: []models.User{},
|
||||
Clusters: []models.Cluster{},
|
||||
Hosts: []models.Host{},
|
||||
Storages: []models.StorageConfig{},
|
||||
Deployments: []models.LokiDeployment{},
|
||||
Collectors: []models.Collector{},
|
||||
Logs: []models.DeploymentLog{},
|
||||
Settings: []models.SystemSetting{},
|
||||
LogSources: []models.LogSource{},
|
||||
AlertRules: []models.AlertRule{},
|
||||
AlertDeliveries: []models.AlertDelivery{},
|
||||
NotifyChannels: []models.NotifyChannel{},
|
||||
LogPolicyGroups: []models.LogPolicyGroup{},
|
||||
AuditLogs: []models.AuditLog{},
|
||||
SchemaVersion: CurrentSchemaVersion,
|
||||
dataFile: cfg.DatabasePath,
|
||||
Users: []models.User{},
|
||||
Clusters: []models.Cluster{},
|
||||
Hosts: []models.Host{},
|
||||
Storages: []models.StorageConfig{},
|
||||
Deployments: []models.LokiDeployment{},
|
||||
Collectors: []models.Collector{},
|
||||
Logs: []models.DeploymentLog{},
|
||||
Settings: []models.SystemSetting{},
|
||||
LogSources: []models.LogSource{},
|
||||
AlertRules: []models.AlertRule{},
|
||||
AlertDeliveries: []models.AlertDelivery{},
|
||||
NotifyChannels: []models.NotifyChannel{},
|
||||
LogPolicyGroups: []models.LogPolicyGroup{},
|
||||
AuditLogs: []models.AuditLog{},
|
||||
DailyVolumeSnapshots: []models.DailyVolumeSnapshot{},
|
||||
SchemaVersion: CurrentSchemaVersion,
|
||||
sequences: map[string]uint{
|
||||
"users": 0,
|
||||
"clusters": 0,
|
||||
"hosts": 0,
|
||||
"storages": 0,
|
||||
"deployments": 0,
|
||||
"collectors": 0,
|
||||
"logs": 0,
|
||||
"logPolicyGroups": 0,
|
||||
"logSources": 0,
|
||||
"notifyChannels": 0,
|
||||
"alertRules": 0,
|
||||
"alertDeliveries": 0,
|
||||
"auditLogs": 0,
|
||||
"users": 0,
|
||||
"clusters": 0,
|
||||
"hosts": 0,
|
||||
"storages": 0,
|
||||
"deployments": 0,
|
||||
"collectors": 0,
|
||||
"logs": 0,
|
||||
"logPolicyGroups": 0,
|
||||
"logSources": 0,
|
||||
"notifyChannels": 0,
|
||||
"alertRules": 0,
|
||||
"alertDeliveries": 0,
|
||||
"auditLogs": 0,
|
||||
"dailyVolumeSnapshots": 0,
|
||||
},
|
||||
saveChan: make(chan saveRequest, 10),
|
||||
stopChan: make(chan struct{}),
|
||||
@@ -519,6 +550,11 @@ func InitDB() error {
|
||||
d.sequences["auditLogs"] = a.ID
|
||||
}
|
||||
}
|
||||
for _, v := range d.DailyVolumeSnapshots {
|
||||
if v.ID > d.sequences["dailyVolumeSnapshots"] {
|
||||
d.sequences["dailyVolumeSnapshots"] = v.ID
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -92,7 +92,8 @@ func SetupRouter() *gin.Engine {
|
||||
logs.GET("/labels", logsCtrl.Labels)
|
||||
logs.GET("/label/:name/values", logsCtrl.LabelValues)
|
||||
logs.GET("/my-scope", logsCtrl.MyScope)
|
||||
logs.POST("/histogram", logsCtrl.Histogram)
|
||||
logs.GET("/histogram", logsCtrl.Histogram)
|
||||
logs.GET("/daily-volume", logsCtrl.DailyVolume)
|
||||
// 总量统计(index/stats):只读索引,用于算总页数
|
||||
logs.POST("/stats", logsCtrl.Stats)
|
||||
logs.POST("/metrics", middleware.AdminAuth(), logsCtrl.Metrics)
|
||||
|
||||
@@ -31,6 +31,40 @@ type builtinRule struct {
|
||||
// 3. cooldown 统一 ≥ 120 分钟:同一问题 2 小时内只提醒一次。
|
||||
// 4. 绝对阈值天然不跨环境通用,大集群(如 300GB/天)必须按 app/namespace
|
||||
// 收窄查询,否则任何阈值都会被常态流量击穿。这一点在 UI 上有提示。
|
||||
//
|
||||
// ensurePlaceholderChannel 返回一个可挂靠的通道 ID:
|
||||
// 已有通道取第一个;没有则建一个禁用的占位通道。
|
||||
func ensurePlaceholderChannel(db *repository.JsonDB) uint {
|
||||
db.Lock()
|
||||
defer db.Unlock()
|
||||
if len(db.NotifyChannels) > 0 {
|
||||
return db.NotifyChannels[0].ID
|
||||
}
|
||||
next := uint(0)
|
||||
for _, ch := range db.NotifyChannels {
|
||||
if ch.ID > next {
|
||||
next = ch.ID
|
||||
}
|
||||
}
|
||||
id := next + 1
|
||||
db.NotifyChannels = append(db.NotifyChannels, models.NotifyChannel{
|
||||
ID: id,
|
||||
Name: "默认通知通道(待配置 Webhook)",
|
||||
Enabled: false,
|
||||
CreatedAt: time.Now(),
|
||||
UpdatedAt: time.Now(),
|
||||
})
|
||||
return id
|
||||
}
|
||||
|
||||
// channelIDsFor 把单个通道 ID 包成切片;0 表示没有可用通道(保持空)。
|
||||
func channelIDsFor(id uint) []uint {
|
||||
if id == 0 {
|
||||
return nil
|
||||
}
|
||||
return []uint{id}
|
||||
}
|
||||
|
||||
var builtinRules = []builtinRule{
|
||||
// —— 致命类:这类日志本该极少出现,但也不能一条就报 ——
|
||||
{name: "致命日志 FATAL", query: `{namespace=~".+"} |= "FATAL"`, windowMin: 15, threshold: 10, cooldown: 120},
|
||||
@@ -77,6 +111,14 @@ func SeedAlertRules() {
|
||||
// 绑定第一个 managed 数据源(通常是唯一部署)
|
||||
target := sources[0]
|
||||
|
||||
// 内置规则一律挂到默认通知通道。
|
||||
// 首次部署没有任何通道时,这里建一个**禁用**的占位通道(Webhook 留空):
|
||||
// - 规则不会变成"孤儿"(发送历史会明确写"通道未投递"而不是静默)
|
||||
// - 管理员填好 Webhook 再启用即可,不必逐条给规则关联通道
|
||||
// - 之所以不放"已启用 + 空地址":那样每次命中都会投递失败刷屏
|
||||
// 这里直接操作 db 而不调 services,避免 seed ↔ services 的包循环依赖。
|
||||
defaultChannelID := ensurePlaceholderChannel(db)
|
||||
|
||||
newID := db.NextID("alertRules")
|
||||
db.Lock()
|
||||
defer db.Unlock()
|
||||
@@ -89,7 +131,8 @@ func SeedAlertRules() {
|
||||
ID: newID, Name: r.name, SourceID: target.id,
|
||||
Query: r.query, WindowMin: r.windowMin, Threshold: r.threshold,
|
||||
Enabled: false, CooldownMin: r.cooldown,
|
||||
CreatedAt: time.Now(), UpdatedAt: time.Now(),
|
||||
ChannelIDs: channelIDsFor(defaultChannelID),
|
||||
CreatedAt: time.Now(), UpdatedAt: time.Now(),
|
||||
})
|
||||
}
|
||||
db.RequestSave()
|
||||
|
||||
@@ -0,0 +1,243 @@
|
||||
package services
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"sort"
|
||||
"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"`
|
||||
}
|
||||
|
||||
const (
|
||||
// dailyVolumeBackfillDays 启动时回补天数。
|
||||
// 取 2 天:昨天(完整)+ 今天(进行中),足够覆盖"重启导致漏抓"的常见情况。
|
||||
// 不设更大是因为 Loki retention 通常只有 24h,更早的已经查不到了。
|
||||
dailyVolumeBackfillDays = 2
|
||||
// dailyVolumeRetentionDays 平台自己留多久的历史。一年足够看年度趋势。
|
||||
dailyVolumeRetentionDays = 400
|
||||
)
|
||||
|
||||
// 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()
|
||||
}
|
||||
|
||||
// CaptureDailyVolume 抓取指定日期的日志量并落库(按天幂等:同一天重复抓会覆盖)。
|
||||
//
|
||||
// 数据来源是 Loki 的 index/stats(只读索引元数据),实测毫秒级,
|
||||
// 所以就算是补几十天历史也不会拖慢任何东西。
|
||||
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
|
||||
}
|
||||
// 回补历史时,如果那天的起点已经超出 Loki 保留期(默认 24h),
|
||||
// Loki 里根本没那么多数据 —— 必须标成"不完整",
|
||||
// 否则图上会把"数据被删了"误读成"那天日志量真的很少"。
|
||||
if start < time.Now().Add(-24*time.Hour).UnixNano() {
|
||||
partial = true
|
||||
}
|
||||
|
||||
// 只查总量(不带 namespace 条件),一次拿到该源当天的条数与体积
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
stats, err := LokiStatsFor(ctx, source, `{namespace=~".+"}`, start, end)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
date := dateKey(day)
|
||||
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 == source.ID && snap.Namespace == "" {
|
||||
snap.Bytes = stats.Bytes
|
||||
snap.Entries = stats.Entries
|
||||
snap.Partial = partial
|
||||
snap.CapturedAt = time.Now()
|
||||
db.RequestSave()
|
||||
return nil
|
||||
}
|
||||
}
|
||||
nextID++
|
||||
db.DailyVolumeSnapshots = append(db.DailyVolumeSnapshots, models.DailyVolumeSnapshot{
|
||||
ID: nextID,
|
||||
Date: date,
|
||||
SourceID: source.ID,
|
||||
Namespace: "",
|
||||
Bytes: stats.Bytes,
|
||||
Entries: stats.Entries,
|
||||
Partial: partial,
|
||||
CapturedAt: time.Now(),
|
||||
})
|
||||
db.RequestSave()
|
||||
return nil
|
||||
}
|
||||
|
||||
// CaptureRecentDailyVolumes 回补最近 N 天(含今天),供启动时与定时任务调用。
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
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 天,按日期升序返回。
|
||||
//
|
||||
// 今天如果还没抓过,会实时补一次(这样用户切到页面时当天数字是最新的)。
|
||||
func GetDailyVolumeHistory(sourceID uint, days int) []DailyVolumePoint {
|
||||
if days <= 0 {
|
||||
days = 7
|
||||
}
|
||||
if days > dailyVolumeRetentionDays {
|
||||
days = dailyVolumeRetentionDays
|
||||
}
|
||||
|
||||
// 今天实时补抓,保证"今天"这根柱子是活的
|
||||
db := repository.GetDB()
|
||||
if db != nil {
|
||||
db.RLock()
|
||||
var target *models.LogSource
|
||||
for i := range db.LogSources {
|
||||
if db.LogSources[i].ID == sourceID {
|
||||
src := db.LogSources[i]
|
||||
target = &src
|
||||
break
|
||||
}
|
||||
}
|
||||
db.RUnlock()
|
||||
if target != nil {
|
||||
if err := CaptureDailyVolume(target, time.Now(), true); err != nil {
|
||||
log.Printf("[daily-volume] 实时补抓今天失败: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
from := dateKey(time.Now().AddDate(0, 0, -(days - 1)))
|
||||
points := []DailyVolumePoint{}
|
||||
db = repository.GetDB()
|
||||
if db == nil {
|
||||
return points
|
||||
}
|
||||
db.RLock()
|
||||
defer db.RUnlock()
|
||||
for _, snap := range db.DailyVolumeSnapshots {
|
||||
if snap.SourceID != sourceID || snap.Namespace != "" {
|
||||
continue
|
||||
}
|
||||
if snap.Date < from {
|
||||
continue
|
||||
}
|
||||
points = append(points, DailyVolumePoint{
|
||||
Date: snap.Date,
|
||||
Bytes: snap.Bytes,
|
||||
Entries: snap.Entries,
|
||||
Partial: snap.Partial,
|
||||
FromHistory: true,
|
||||
})
|
||||
}
|
||||
sort.Slice(points, func(i, j int) bool { return points[i].Date < points[j].Date })
|
||||
return points
|
||||
}
|
||||
|
||||
// StartDailyVolumeCapture 后台每日抓取。
|
||||
//
|
||||
// 每 30 分钟检查一次:只要发现"昨天"还没落库就立刻补,
|
||||
// 而不是死等 00:10 —— 平台可能整晚都没运行,定时器会错过那个点。
|
||||
func StartDailyVolumeCapture() {
|
||||
go func() {
|
||||
// 启动先补最近两天(覆盖重启场景)
|
||||
time.Sleep(20 * time.Second)
|
||||
CaptureRecentDailyVolumes()
|
||||
|
||||
ticker := time.NewTicker(30 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
CaptureRecentDailyVolumes()
|
||||
}
|
||||
}()
|
||||
}
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"lokistack/models"
|
||||
@@ -60,6 +61,13 @@ func GetNotifyChannelDecrypted(id uint) *models.NotifyChannel {
|
||||
|
||||
// SaveNotifyChannel 新建(id=0)或更新;webhook/secret 非空才更新(留空=保留旧值)
|
||||
func SaveNotifyChannel(id uint, name, webhook, secret string, enabled bool) (*models.NotifyChannel, error) {
|
||||
// 启用则必须有 Webhook:否则告警命中后必然投递失败,
|
||||
// 用户只会看到发送历史里一片红,却不知道是配置没填完。
|
||||
if enabled && strings.TrimSpace(webhook) == "" {
|
||||
if existing := GetNotifyChannelDecrypted(id); existing == nil || strings.TrimSpace(existing.Webhook) == "" {
|
||||
return nil, fmt.Errorf("启用通知通道前必须填写 Webhook 地址")
|
||||
}
|
||||
}
|
||||
db := repository.GetDB()
|
||||
// 更新:留空字段回填旧值(旧值已是密文)
|
||||
if id > 0 {
|
||||
|
||||
Reference in new issue
Block a user