=== 1. 主机模式数据源地址 0.0.0.0 → 主机真实 IP(查询页"永远加载中"的根因)===
现象:切换数据源后命名空间/App 下拉不动,一直显示"加载中…",控制台报
Get "http://0.0.0.0:3100/loki/api/v1/labels": dial tcp 0.0.0.0:3100: connect: connection refused
根因:主机模式部署的 endpoint 记的是**容器监听地址** 0.0.0.0:3100,
而平台进程不在被采集主机上,0.0.0.0 只在本机有意义。
EnsureLogSourceForDeployment 把它原样写成了数据源 URL。
修法:
- 新增 reachableEndpoint():把 0.0.0.0 / [::] 换成关联主机的真实 IP
- schemaVersion v3→v4 迁移 fixUnreachableLogSourceURLs(),修正已有脏数据
- 实测:http://0.0.0.0:3100 → http://172.19.16.8:3100,该源 label 查询由 502 变 200
=== 2. 前端:区分"加载中"与"加载失败" ===
原来用 `loading={!options.length}` 表示加载中 —— 数据源不可达时列表永远为空,
于是界面永远转圈,把真实错误完全藏起来了(这次的 0.0.0.0 就是这样被掩盖的)。
改为独立的 namespaceLoading / namespaceError 两个状态:
- 切换数据源先清空旧 options(否则显示上一个源的命名空间,看着像"没反应")
- 失败显示"加载失败"并在旁边给「重试」按钮,不用刷新整页
=== 3. 日志分析页:主图改走索引接口(原来 5 条 count_over_time 全超时)===
实测 astralx(约 370GB/天)15 分钟窗口:
count_over_time → exec 29s + 排队 95s,超过 30s 护栏 → 502
index/volume_range → 0s
改为两阶段渲染:
- 总量趋势改用 /logs/histogram(内部 index/volume_range),**0s 出图**
- 错误/警告/app 维度仍需读日志内容,串行执行,先渲染主图再更新
- heavyQueryTimeout 30s → 90s(实测 error 维度查询 14s 可完成)
=== 4. 告警规则按日志平台定位收敛 ===
- seed 里删掉「Kubernetes 节点 NotReady」「容器 OOMKilled」:
这类属于 kube-state-metrics / Prometheus 的职责,用日志字符串匹配不可靠
- 迁移里清掉线上已存在的同类规则(3 条)
- 修 AlertRule / NotifyChannel 重复 ID(同 logSources 一样漏在 sequences 重建外)
=== 5. 告警页 UI ===
- 规则列表由"每条一张卡"改为紧凑表格(一屏看全,阈值可对齐比较)
- 通知通道(飞书/Lark/Webhook)从「系统设置」搬到告警页折叠区:
规则和"发到哪"是一件事的两半,分两页配容易漏配导致告警发不出去
=== 6. 权限收敛 ===
- /alerts 加 adminOnly:普通用户按查询授权只看「日志查询」,不再看到告警菜单
- 系统设置从侧栏移到右上角齿轮(低频平台配置,不占侧栏一位)
=== 7. 迁移设计教训(写进代码注释)===
v2 时把"告警去重 + 清基础设施规则"塞进 case 1,但线上库已到 v2,
那段代码永远不执行。**结构变更只要逻辑变了、目标库已达该版本,就必须再抬一版**,
不能改历史迁移分支。
=== 实测(v4-1028-2210 / 979bdb4a48e8)===
v4 迁移:数据源 URL 修正 http://0.0.0.0:3100 → http://172.19.16.8:3100
source 1 label:200,0s(astralx / log)
source 2 label:200,0s(不再 connection refused)
分析页 histogram astralx 15m:0s
分析页 metrics error astralx 15m:14s
告警规则:10 条(已无基础设施类)、ID 唯一
另清理 172.19.16.8 上 13 个残留容器,回收 6.1GB
726 lines
21 KiB
Go
726 lines
21 KiB
Go
package repository
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"log"
|
||
"os"
|
||
"strings"
|
||
"path/filepath"
|
||
"sync"
|
||
"time"
|
||
|
||
"lokistack/config"
|
||
"lokistack/models"
|
||
"lokistack/pkg/metrics"
|
||
)
|
||
|
||
// saveRequest 表示一个保存请求,通过channel序列化
|
||
type saveRequest struct {
|
||
done chan error // 用于同步等待保存完成
|
||
}
|
||
|
||
// JsonDB 使用基于channel的序列化保存机制,防止并发数据丢失。
|
||
// 该包只负责持久化(落盘 + 并发安全),领域实体定义在 models 包。
|
||
type JsonDB struct {
|
||
mu sync.RWMutex
|
||
dataFile string
|
||
Users []models.User `json:"users"`
|
||
Clusters []models.Cluster `json:"clusters"`
|
||
Hosts []models.Host `json:"hosts"`
|
||
Storages []models.StorageConfig `json:"storages"`
|
||
Deployments []models.LokiDeployment `json:"deployments"`
|
||
Collectors []models.Collector `json:"collectors"`
|
||
Logs []models.DeploymentLog `json:"logs"`
|
||
Settings []models.SystemSetting `json:"settings"`
|
||
LogSources []models.LogSource `json:"logSources"`
|
||
AlertRules []models.AlertRule `json:"alertRules"`
|
||
AlertDeliveries []models.AlertDelivery `json:"alertDeliveries"`
|
||
NotifyChannels []models.NotifyChannel `json:"notifyChannels"`
|
||
LogPolicyGroups []models.LogPolicyGroup `json:"logPolicyGroups"`
|
||
// AuditLogs 操作审计日志(保留期 90 天 / 上限 2 万条,见 services.auditRetention)
|
||
AuditLogs []models.AuditLog `json:"auditLogs"`
|
||
|
||
// SchemaVersion 数据结构版本号。
|
||
// 之前加字段全靠 JSON 反序列化的零值兜底,字段一多必然出事
|
||
// (例如新增 ScrapeJobs 时老数据要手写兼容)。有了版本号,
|
||
// 以后的结构变更可以在 InitDB 里按版本号顺序迁移,可测、可回滚。
|
||
SchemaVersion int `json:"schemaVersion"`
|
||
sequences map[string]uint
|
||
|
||
// 防抖保存机制
|
||
saveTimer *time.Timer
|
||
saveMu sync.Mutex
|
||
saveChan chan saveRequest
|
||
stopChan chan struct{}
|
||
wg sync.WaitGroup
|
||
}
|
||
|
||
var db *JsonDB
|
||
|
||
func GetDB() *JsonDB {
|
||
return db
|
||
}
|
||
|
||
// fixUnreachableLogSourceURLs 把数据源里 0.0.0.0 / [::] 的监听地址换成主机真实 IP。
|
||
//
|
||
// 平台进程不在被采集主机上,0.0.0.0 只在本机有意义,用它必然连不上。
|
||
func (d *JsonDB) fixUnreachableLogSourceURLs() {
|
||
hostIPByID := map[uint]string{}
|
||
for _, h := range d.Hosts {
|
||
hostIPByID[h.ID] = strings.TrimSpace(h.HostIP)
|
||
}
|
||
// 数据源按名字对应部署名,用于推断该源属于哪台主机
|
||
hostByName := map[string]string{}
|
||
for _, dep := range d.Deployments {
|
||
if dep.HostID == nil {
|
||
continue
|
||
}
|
||
if ip, ok := hostIPByID[*dep.HostID]; ok && ip != "" {
|
||
hostByName[dep.Name] = ip
|
||
}
|
||
}
|
||
|
||
for i := range d.LogSources {
|
||
src := &d.LogSources[i]
|
||
if !strings.Contains(src.URL, "0.0.0.0") && !strings.Contains(src.URL, "[::]") {
|
||
continue
|
||
}
|
||
ip := hostByName[src.Name]
|
||
if ip == "" && len(d.Hosts) == 1 {
|
||
// 只有一台主机时可以直接推断(常见于单机试用)
|
||
ip = strings.TrimSpace(d.Hosts[0].HostIP)
|
||
}
|
||
if ip == "" {
|
||
log.Printf("[schema] 数据源 %q 的 URL 是不可达的监听地址 %s,但无法推断主机 IP,请手动修正", src.Name, src.URL)
|
||
continue
|
||
}
|
||
old := src.URL
|
||
src.URL = strings.Replace(src.URL, "0.0.0.0", ip, 1)
|
||
src.URL = strings.Replace(src.URL, "[::]", ip, 1)
|
||
log.Printf("[schema] 数据源 %q URL 修正:%s → %s", src.Name, old, src.URL)
|
||
}
|
||
}
|
||
|
||
// dedupeCollectionIDs 对所有带 ID 的集合做重复 ID 修复。
|
||
//
|
||
// 注意:实际迁移里按版本分步调用(见 migrateSchema),
|
||
// 不要直接用它替代分步逻辑 —— 那会让"已迁移库"漏掉后加的清理步骤。
|
||
//
|
||
// 背景:这些集合曾经都漏在 InitDB 的 sequences 重建之外,导致 NextID 每次
|
||
// 进程重启都从 1 开始,于是不同时期创建的对象会撞号。撞号后的表现各不相同:
|
||
// LogSource : 查询页数据源下拉选不动(两个 option 同 value)
|
||
// AlertRule : 规则列表出现两条同 id,编辑/删除操作命中错的那条
|
||
// NotifyChannel: 通知渠道覆盖错对象
|
||
func (d *JsonDB) dedupeCollectionIDs() {
|
||
d.dedupeLogSourceIDs()
|
||
d.dedupeAlertRuleIDs()
|
||
d.dedupeNotifyChannelIDs()
|
||
}
|
||
|
||
// dedupeAlertRuleIDs 重新编号重复的告警规则 ID。
|
||
// 同名规则(seed 重复执行导致)只保留第一条,其余删除。
|
||
func (d *JsonDB) dedupeAlertRuleIDs() {
|
||
if len(d.AlertRules) == 0 {
|
||
return
|
||
}
|
||
seenID := map[uint]bool{}
|
||
seenName := map[string]bool{}
|
||
next := uint(0)
|
||
for _, r := range d.AlertRules {
|
||
if r.ID > next {
|
||
next = r.ID
|
||
}
|
||
}
|
||
fixed := make([]models.AlertRule, 0, len(d.AlertRules))
|
||
for i := range d.AlertRules {
|
||
rule := d.AlertRules[i]
|
||
if rule.Name != "" && seenName[rule.Name] {
|
||
log.Printf("[schema] 删除重复的告警规则 %q(同名)", rule.Name)
|
||
continue
|
||
}
|
||
seenName[rule.Name] = true
|
||
if rule.ID == 0 || seenID[rule.ID] {
|
||
next++
|
||
log.Printf("[schema] 告警规则 %q 重复 ID %d → 重新分配为 %d", rule.Name, rule.ID, next)
|
||
rule.ID = next
|
||
}
|
||
seenID[rule.ID] = true
|
||
fixed = append(fixed, rule)
|
||
}
|
||
d.AlertRules = fixed
|
||
for _, r := range d.AlertRules {
|
||
if r.ID > d.sequences["alertRules"] {
|
||
d.sequences["alertRules"] = r.ID
|
||
}
|
||
}
|
||
}
|
||
|
||
// dedupeNotifyChannelIDs 重新编号重复的通知渠道 ID。
|
||
func (d *JsonDB) dedupeNotifyChannelIDs() {
|
||
if len(d.NotifyChannels) == 0 {
|
||
return
|
||
}
|
||
seen := map[uint]bool{}
|
||
next := uint(0)
|
||
for _, c := range d.NotifyChannels {
|
||
if c.ID > next {
|
||
next = c.ID
|
||
}
|
||
}
|
||
for i := range d.NotifyChannels {
|
||
ch := &d.NotifyChannels[i]
|
||
if ch.ID == 0 || seen[ch.ID] {
|
||
next++
|
||
log.Printf("[schema] 通知渠道 %q 重复 ID %d → 重新分配为 %d", ch.Name, ch.ID, next)
|
||
ch.ID = next
|
||
}
|
||
seen[ch.ID] = true
|
||
if ch.ID > d.sequences["notifyChannels"] {
|
||
d.sequences["notifyChannels"] = ch.ID
|
||
}
|
||
}
|
||
}
|
||
|
||
// removeInfraAlertRules 删除误配置的「基础设施类」告警规则。
|
||
//
|
||
// 判据:查询指向 kube-state-metrics / event-exporter,或在日志里匹配
|
||
// 节点 NotReady / 容器 OOMKilled。这些是 Prometheus 指标监控的职责:
|
||
// - 日志里有没有这行取决于采集器是否采到那些命名空间,不可靠
|
||
// - 指标侧(kube_node_status_condition / kube_pod_container_status_terminated_reason)
|
||
// 才是权威数据源
|
||
func (d *JsonDB) removeInfraAlertRules() {
|
||
infraMarkers := []string{"kube-state-metrics", "event-exporter"}
|
||
infraNames := []string{"Kubernetes 节点 NotReady", "K8s 节点 NotReady", "容器 OOMKilled"}
|
||
|
||
kept := make([]models.AlertRule, 0, len(d.AlertRules))
|
||
for _, r := range d.AlertRules {
|
||
isInfra := false
|
||
for _, marker := range infraMarkers {
|
||
if strings.Contains(r.Query, marker) {
|
||
isInfra = true
|
||
}
|
||
}
|
||
for _, name := range infraNames {
|
||
if strings.TrimSpace(r.Name) == name {
|
||
isInfra = true
|
||
}
|
||
}
|
||
if isInfra {
|
||
log.Printf("[schema] 移除基础设施类告警规则 %q(应交给 Prometheus)", r.Name)
|
||
continue
|
||
}
|
||
kept = append(kept, r)
|
||
}
|
||
d.AlertRules = kept
|
||
}
|
||
|
||
// dedupeLogSourceIDs 修复历史遗留的重复 ID,并同步更新引用它的策略组。
|
||
//
|
||
// 策略组通过 SourceID 关联数据源。重复 ID 时无法判断分组到底指向哪一个,
|
||
// 这里按"保留第一个、其余顺延"处理,并让原本指向该 ID 的策略组继续指向
|
||
// 编号最小的那个(即重建前第一次创建的数据源)。
|
||
func (d *JsonDB) dedupeLogSourceIDs() {
|
||
if len(d.LogSources) < 2 {
|
||
return
|
||
}
|
||
seen := map[uint]bool{}
|
||
remap := map[uint]uint{} // 旧ID → 新ID(仅记录需要改的)
|
||
next := uint(0)
|
||
for _, s := range d.LogSources {
|
||
if s.ID > next {
|
||
next = s.ID
|
||
}
|
||
}
|
||
|
||
fixed := make([]models.LogSource, 0, len(d.LogSources))
|
||
kept := map[uint]bool{} // 已被占用的 ID(首次出现时占位)
|
||
for i := range d.LogSources {
|
||
src := d.LogSources[i]
|
||
if src.ID == 0 || kept[src.ID] {
|
||
// 重复或非法 ID:分配新号
|
||
next++
|
||
old := src.ID
|
||
src.ID = next
|
||
if _, ok := remap[old]; !ok && old != 0 {
|
||
// 同一个旧 ID 只记录一次映射(指向保留的那个)
|
||
remap[old] = 0 // 0 表示"不需要改引用"
|
||
}
|
||
log.Printf("[schema] 数据源 %q 重复 ID %d → 重新分配为 %d", src.Name, old, src.ID)
|
||
} else {
|
||
kept[src.ID] = true
|
||
}
|
||
seen[src.ID] = true
|
||
fixed = append(fixed, src)
|
||
}
|
||
d.LogSources = fixed
|
||
|
||
// 同步策略组引用(只在旧 ID 已不存在时改写)
|
||
for i := range d.LogPolicyGroups {
|
||
sid := d.LogPolicyGroups[i].SourceID
|
||
if sid == 0 || seen[sid] {
|
||
continue
|
||
}
|
||
if nv, ok := remap[sid]; ok && nv != 0 {
|
||
d.LogPolicyGroups[i].SourceID = nv
|
||
}
|
||
}
|
||
|
||
// 迁移发生在 sequences 重建**之后**,重新分配可能抬高最大 ID,
|
||
// 必须同步提升计数器,否则紧接着创建的数据源会撞号。
|
||
for _, s := range d.LogSources {
|
||
if s.ID > d.sequences["logSources"] {
|
||
d.sequences["logSources"] = s.ID
|
||
}
|
||
}
|
||
|
||
// 保证只有一个默认数据源
|
||
defaultSeen := false
|
||
for i := range d.LogSources {
|
||
if d.LogSources[i].IsDefault {
|
||
if defaultSeen {
|
||
d.LogSources[i].IsDefault = false
|
||
}
|
||
defaultSeen = true
|
||
}
|
||
}
|
||
if !defaultSeen && len(d.LogSources) > 0 {
|
||
d.LogSources[0].IsDefault = true
|
||
}
|
||
}
|
||
|
||
// CurrentSchemaVersion 当前数据结构版本。
|
||
// 每次做**不兼容**的结构变更时 +1,并在 migrateSchema 里补一段迁移逻辑。
|
||
//
|
||
// v0/v1(无字段)→ 1:新增 auditLogs / schemaVersion(老库走零值兜底即可,无需处理)
|
||
// v1 → 2:预留示例(如把 ScrapeJobs 从 logPath 拆分时,需按旧结构生成默认 job)
|
||
const CurrentSchemaVersion = 4
|
||
|
||
// migrateSchema 按版本号顺序迁移老库。返回是否发生了迁移(需要回写)。
|
||
func (d *JsonDB) migrateSchema() bool {
|
||
if d.SchemaVersion >= CurrentSchemaVersion {
|
||
return false
|
||
}
|
||
from := d.SchemaVersion
|
||
|
||
// 逐版本递进,每段只处理"上一版 → 下一版"的差异,
|
||
// 不要写成 if version < X 的叠加判断,否则分支会指数增长。
|
||
for d.SchemaVersion < CurrentSchemaVersion {
|
||
switch d.SchemaVersion {
|
||
case 0:
|
||
// v0 → v1:补审计表;老库没有 auditLogs 字段,反序列化后为 nil,
|
||
// 这里显式初始化,避免后续 append 到 nil 之外还留着 nil 判断。
|
||
if d.AuditLogs == nil {
|
||
d.AuditLogs = []models.AuditLog{}
|
||
}
|
||
d.SchemaVersion = 1
|
||
case 1:
|
||
// v1 → v2:修复 LogSource 重复 ID(曾漏在 sequences 重建之外,
|
||
// NextID 每次重启从 1 开始,线上出现两个数据源都是 id=1)。
|
||
d.dedupeLogSourceIDs()
|
||
d.SchemaVersion = 2
|
||
case 2:
|
||
// v2 → v3:修复 AlertRule / NotifyChannel 重复 ID + 清掉基础设施类规则。
|
||
//
|
||
// ⚠️ 教训:v2 那次我把"告警去重 + 清基础设施规则"塞进了 case 1,
|
||
// 但当时线上库已迁移到 v2,于是这段代码永远不执行(DB 里 schemaVersion
|
||
// 已经等于当前版本)。结构变更只要改了逻辑、且目标库已到该版本,
|
||
// 就必须**再抬一版**,不能改历史迁移分支。
|
||
d.dedupeAlertRuleIDs()
|
||
d.dedupeNotifyChannelIDs()
|
||
d.removeInfraAlertRules()
|
||
d.SchemaVersion = 3
|
||
case 3:
|
||
// v3 → v4:修复主机模式数据源的不可达 URL。
|
||
//
|
||
// 主机模式部署的 endpoint 存的是容器监听地址 0.0.0.0:3100,
|
||
// 平台把它原样当数据源 URL → 查询页对该数据源的所有请求
|
||
// 都报 dial tcp 0.0.0.0:3100: connection refused,
|
||
// 前端表现为「命名空间/App 下拉永远加载中」。
|
||
d.fixUnreachableLogSourceURLs()
|
||
d.SchemaVersion = 4
|
||
default:
|
||
// 防御:出现未知版本(比当前代码还新)时不乱动,避免降级把数据写坏
|
||
log.Printf("[schema] 数据库版本 %d 高于当前代码支持版本 %d,跳过迁移", d.SchemaVersion, CurrentSchemaVersion)
|
||
return false
|
||
}
|
||
}
|
||
log.Printf("[schema] 数据结构已从 v%d 迁移到 v%d", from, d.SchemaVersion)
|
||
return true
|
||
}
|
||
|
||
func InitDB() error {
|
||
cfg := config.GetConfig()
|
||
_ = 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,
|
||
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,
|
||
},
|
||
saveChan: make(chan saveRequest, 10),
|
||
stopChan: make(chan struct{}),
|
||
}
|
||
|
||
if data, err := os.ReadFile(d.dataFile); err == nil {
|
||
if err := json.Unmarshal(data, d); err == nil {
|
||
d.sequences = map[string]uint{}
|
||
for _, u := range d.Users {
|
||
if u.ID > d.sequences["users"] {
|
||
d.sequences["users"] = u.ID
|
||
}
|
||
}
|
||
for _, c := range d.Clusters {
|
||
if c.ID > d.sequences["clusters"] {
|
||
d.sequences["clusters"] = c.ID
|
||
}
|
||
}
|
||
for _, h := range d.Hosts {
|
||
if h.ID > d.sequences["hosts"] {
|
||
d.sequences["hosts"] = h.ID
|
||
}
|
||
}
|
||
for _, s := range d.Storages {
|
||
if s.ID > d.sequences["storages"] {
|
||
d.sequences["storages"] = s.ID
|
||
}
|
||
}
|
||
for _, de := range d.Deployments {
|
||
if de.ID > d.sequences["deployments"] {
|
||
d.sequences["deployments"] = de.ID
|
||
}
|
||
}
|
||
for _, col := range d.Collectors {
|
||
if col.ID > d.sequences["collectors"] {
|
||
d.sequences["collectors"] = col.ID
|
||
}
|
||
}
|
||
for _, l := range d.Logs {
|
||
if l.ID > d.sequences["logs"] {
|
||
d.sequences["logs"] = l.ID
|
||
}
|
||
}
|
||
for _, group := range d.LogPolicyGroups {
|
||
if group.ID > d.sequences["logPolicyGroups"] {
|
||
d.sequences["logPolicyGroups"] = group.ID
|
||
}
|
||
}
|
||
// ⚠️ 以下集合曾经漏在重建之外,导致 NextID 每次重启都从 1 开始,
|
||
// 于是不同进程生命周期里创建的对象 ID 重复(线上实际出现过
|
||
// 两个 LogSource 都是 id=1 → 前端数据源下拉无法切换)。
|
||
// 以后新增任何带 ID 的集合,都必须在这里补一段。
|
||
for _, s := range d.LogSources {
|
||
if s.ID > d.sequences["logSources"] {
|
||
d.sequences["logSources"] = s.ID
|
||
}
|
||
}
|
||
for _, ch := range d.NotifyChannels {
|
||
if ch.ID > d.sequences["notifyChannels"] {
|
||
d.sequences["notifyChannels"] = ch.ID
|
||
}
|
||
}
|
||
for _, rule := range d.AlertRules {
|
||
if rule.ID > d.sequences["alertRules"] {
|
||
d.sequences["alertRules"] = rule.ID
|
||
}
|
||
}
|
||
for _, del := range d.AlertDeliveries {
|
||
if del.ID > d.sequences["alertDeliveries"] {
|
||
d.sequences["alertDeliveries"] = del.ID
|
||
}
|
||
}
|
||
for _, a := range d.AuditLogs {
|
||
if a.ID > d.sequences["auditLogs"] {
|
||
d.sequences["auditLogs"] = a.ID
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
migratedLogAccess := migrateLogAccessToGroups(d)
|
||
// 结构版本迁移(审计表、采集项等新字段的兼容处理)
|
||
migratedSchema := d.migrateSchema()
|
||
// 启动序列化保存goroutine
|
||
d.wg.Add(1)
|
||
go d.saveWorker()
|
||
|
||
db = d
|
||
if migratedLogAccess || migratedSchema {
|
||
d.RequestSave()
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Legacy per-user grants are grouped by identical source and label scope.
|
||
func migrateLogAccessToGroups(d *JsonDB) bool {
|
||
changed := false
|
||
for i := range d.LogSources {
|
||
source := &d.LogSources[i]
|
||
if len(source.Access) == 0 {
|
||
continue
|
||
}
|
||
byLabels := map[string]*models.LogPolicyGroup{}
|
||
for _, rule := range source.Access {
|
||
validUser := false
|
||
for _, user := range d.Users {
|
||
if user.ID == rule.UserID && user.Role != "admin" {
|
||
validUser = true
|
||
break
|
||
}
|
||
}
|
||
if !validUser {
|
||
continue
|
||
}
|
||
alreadyGrouped := false
|
||
for _, group := range d.LogPolicyGroups {
|
||
if group.SourceID != source.ID {
|
||
continue
|
||
}
|
||
for _, id := range group.UserIDs {
|
||
if id == rule.UserID {
|
||
alreadyGrouped = true
|
||
break
|
||
}
|
||
}
|
||
}
|
||
if alreadyGrouped {
|
||
continue
|
||
}
|
||
keyBytes, _ := json.Marshal(rule.Labels)
|
||
key := string(keyBytes)
|
||
group := byLabels[key]
|
||
if group == nil {
|
||
d.sequences["logPolicyGroups"]++
|
||
id := d.sequences["logPolicyGroups"]
|
||
group = &models.LogPolicyGroup{ID: id, Name: fmt.Sprintf("旧授权 · %s · %d", source.Name, id), SourceID: source.ID, Labels: rule.Labels, CreatedAt: time.Now(), UpdatedAt: time.Now()}
|
||
byLabels[key] = group
|
||
}
|
||
group.UserIDs = append(group.UserIDs, rule.UserID)
|
||
}
|
||
for _, group := range byLabels {
|
||
d.LogPolicyGroups = append(d.LogPolicyGroups, *group)
|
||
}
|
||
source.Access = nil
|
||
changed = true
|
||
}
|
||
return changed
|
||
}
|
||
|
||
// Shutdown 优雅关闭数据库,确保所有数据已落盘。
|
||
// 顺序很关键:先停防抖定时器(不再产生新请求)→ 关闭 worker 并等其退出
|
||
// (worker 可能正在处理已入队的请求)→ 此时已无并发,做最后一次 doSave 把
|
||
// 最新内存状态原子写入。注意绝不能在 worker 仍在运行时直接 Save(),否则
|
||
// 两个 doSave 并发改写同一文件会发生"后写覆盖先写"的丢更新。
|
||
func Shutdown() error {
|
||
if db == nil {
|
||
return nil
|
||
}
|
||
|
||
// 1. 取消防抖定时器,阻止新的保存请求进入 channel
|
||
db.saveMu.Lock()
|
||
if db.saveTimer != nil {
|
||
db.saveTimer.Stop()
|
||
db.saveTimer = nil
|
||
}
|
||
db.saveMu.Unlock()
|
||
|
||
// 2. 通知 worker 停止并等待其排空 saveChan、完成进行中的 doSave 后退出
|
||
close(db.stopChan)
|
||
db.wg.Wait()
|
||
|
||
// 3. worker 已退出,无并发,落盘最终状态
|
||
if err := db.doSave(); err != nil {
|
||
return fmt.Errorf("final save failed: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// saveWorker 通过channel序列化所有保存操作,防止并发写入冲突
|
||
func (d *JsonDB) saveWorker() {
|
||
defer d.wg.Done()
|
||
for {
|
||
select {
|
||
case req, ok := <-d.saveChan:
|
||
if !ok {
|
||
return
|
||
}
|
||
err := d.doSave()
|
||
if req.done != nil {
|
||
req.done <- err
|
||
}
|
||
case <-d.stopChan:
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// doSave 执行实际的原子文件写入
|
||
func (d *JsonDB) doSave() error {
|
||
start := time.Now()
|
||
err := d.doSaveInner()
|
||
// JSON DB 是全量写盘,落盘耗时是平台最主要的延迟来源之一,必须可观测
|
||
metrics.ObserveDBSave(time.Since(start).Seconds(), err == nil)
|
||
return err
|
||
}
|
||
|
||
func (d *JsonDB) doSaveInner() error {
|
||
d.mu.RLock()
|
||
data, err := json.MarshalIndent(d, "", " ")
|
||
d.mu.RUnlock()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 原子写入:先写入临时文件,再重命名
|
||
dir := filepath.Dir(d.dataFile)
|
||
tmpFile := filepath.Join(dir, fmt.Sprintf(".lokistack-%d.tmp", time.Now().UnixNano()))
|
||
|
||
if err := os.WriteFile(tmpFile, data, 0600); err != nil {
|
||
return err
|
||
}
|
||
|
||
// 确保数据落盘
|
||
f, err := os.Open(tmpFile)
|
||
if err != nil {
|
||
os.Remove(tmpFile)
|
||
return err
|
||
}
|
||
_ = f.Sync()
|
||
f.Close()
|
||
|
||
if err := os.Rename(tmpFile, d.dataFile); err != nil {
|
||
os.Remove(tmpFile)
|
||
return err
|
||
}
|
||
metrics.DBSizeBytes.Set(float64(len(data)))
|
||
|
||
return nil
|
||
}
|
||
|
||
// Save 同步保存数据库到文件(原子写入)
|
||
func (d *JsonDB) Save() error {
|
||
return d.doSave()
|
||
}
|
||
|
||
// RequestSave 请求异步保存(防抖机制+序列化channel)
|
||
// 多次快速调用会在500ms内合并为一次保存
|
||
func (d *JsonDB) RequestSave() {
|
||
d.saveMu.Lock()
|
||
defer d.saveMu.Unlock()
|
||
|
||
// 如果已有待触发的定时器,重置它(防抖)
|
||
if d.saveTimer != nil {
|
||
d.saveTimer.Stop()
|
||
}
|
||
|
||
d.saveTimer = time.AfterFunc(500*time.Millisecond, func() {
|
||
d.saveMu.Lock()
|
||
d.saveTimer = nil
|
||
d.saveMu.Unlock()
|
||
|
||
// 通过channel发送到保存worker,确保序列化
|
||
select {
|
||
case d.saveChan <- saveRequest{}:
|
||
case <-d.stopChan:
|
||
}
|
||
})
|
||
}
|
||
|
||
// RequestSaveSync 请求同步保存,等待保存完成(用于shutdown场景)
|
||
func (d *JsonDB) RequestSaveSync() error {
|
||
req := saveRequest{done: make(chan error, 1)}
|
||
select {
|
||
case d.saveChan <- req:
|
||
case <-d.stopChan:
|
||
return d.Save() // worker已停止,直接同步保存
|
||
}
|
||
return <-req.done
|
||
}
|
||
|
||
func (d *JsonDB) NextID(table string) uint {
|
||
d.mu.Lock()
|
||
defer d.mu.Unlock()
|
||
d.sequences[table]++
|
||
return d.sequences[table]
|
||
}
|
||
|
||
func (d *JsonDB) Lock() {
|
||
d.mu.Lock()
|
||
}
|
||
|
||
func (d *JsonDB) Unlock() {
|
||
d.mu.Unlock()
|
||
}
|
||
|
||
func (d *JsonDB) RLock() {
|
||
d.mu.RLock()
|
||
}
|
||
|
||
func (d *JsonDB) RUnlock() {
|
||
d.mu.RUnlock()
|
||
}
|
||
|
||
// CreateUser 创建用户并触发防抖保存
|
||
func (d *JsonDB) CreateUser(user *models.User) error {
|
||
d.mu.Lock()
|
||
defer d.mu.Unlock()
|
||
if user.ID == 0 {
|
||
user.ID = d.sequences["users"] + 1
|
||
d.sequences["users"] = user.ID
|
||
}
|
||
if user.CreatedAt.IsZero() {
|
||
now := time.Now()
|
||
user.CreatedAt = now
|
||
user.UpdatedAt = now
|
||
}
|
||
d.Users = append(d.Users, *user)
|
||
d.RequestSave()
|
||
return nil
|
||
}
|
||
|
||
// RecoverOrphanDeployments 启动时把残留的 deploying 状态部署标记为失败:
|
||
// 部署 goroutine 随进程消亡,不恢复会永久卡在「部署中」(UI 无停止入口)。
|
||
func (d *JsonDB) RecoverOrphanDeployments() {
|
||
d.Lock()
|
||
defer d.Unlock()
|
||
recovered := 0
|
||
for i := range d.Deployments {
|
||
if d.Deployments[i].Status == "deploying" {
|
||
d.Deployments[i].Status = "failed"
|
||
d.Deployments[i].StatusMessage = "服务重启导致部署中断,请重新部署"
|
||
d.Deployments[i].UpdatedAt = time.Now()
|
||
recovered++
|
||
}
|
||
}
|
||
if recovered > 0 {
|
||
d.RequestSave()
|
||
}
|
||
}
|