feat(settings,analysis): 日志保留时间可配(发送历史保留 7 天)+ 分析页手动刷新;修复我引入的 RWMutex 死锁
=== 0. 【严重】修复上一版引入的死锁(平台写路径全卡死)===
上一版把保留期读取写成 `GetRetentionSettings()`(内部 db.RLock()),
却在 `db.Lock()` 已持有的情况下调用(recordDelivery / RecordAudit 的裁剪路径)。
Go 的 sync.RWMutex **不可重入** → 第一次写投递历史/审计日志就永久死锁。
现象:登录 28 秒返回空(登录也要写审计)、所有写接口挂住。
go build / go vet / 之前的单测(不触发写路径)全都查不出来。
修法:
- 拆出 retentionFromDBLocked(db) / alertDeliveryRetentionLocked(db) /
auditLogRetentionLocked(db),调用方已持锁时用无锁版本
- upsertSettingLocked 里**不能**用 db.NextID()(内部 d.mu.Lock(),同样死锁),
改为锁内自算 max(id)+1
- 新增 retention_test.go:持写锁调用裁剪函数,2 秒超时未返回即判死锁
实测修复后:登录 103ms(原 28000ms)、panic/deadlock 命中 0。
=== 1. 平台数据保留期可配(系统设置 → 通用配置 → 日志保留时间)===
新增 GET/PUT /api/v1/settings/retention:
alertDeliveryDays 告警发送历史保留天数,默认 7(原写死"最近 50 条")
auditLogDays 审计日志保留天数,默认 90(原写死常量)
保存后立即按新策略裁剪一次,不必等下次触发。
校验:发送历史 1~365 天、审计 1~3650 天,超出返回明确中文原因(不是 500)。
注意与「Loki 日志保留期」区分:那个是部署级配置(组件部署 → 保留策略)。
=== 2. 发送历史:50 条硬上限 → 保留 7 天 ===
后端 Deliveries 接口原来写死 `len(out) < 50`,现在按保留期取(响应上限 500 可调),
标题改为「发送历史(保留 N 天)」,N 从设置里读。
=== 3. 菜单「日志服务」→「组件部署」===
=== 4. 分析页:去掉「全部命名空间」+ 改手动刷新 ===
- 移除「全部命名空间」选项:{namespace=~".+"} 要扫全集群(约 370GB/天),
一次分析几十秒,得不偿失;历史值 __all__ 也会被忽略并回退到策略/第一个 ns
- 筛选改为「选择值 / 已生效值」分离:切换命名空间或时间范围**不再自动查询**,
只把刷新按钮点亮并提示「筛选条件已改,点刷新生效」;
点一次才应用并取数。
原因:一次分析要跑多条 count_over_time,切一下下拉就重查一轮,
快速连点会把 Loki 打满(并发护栏 429),还拖慢别人。
=== 实测(9739c7c092e7)===
登录 103ms / token 273 字符
GET retention → {alertDeliveryDays:7, auditLogDays:90}
PUT 0 天 → 400「告警发送历史保留天数需在 1~365 之间」
平台日志 panic/deadlock 命中 0
go test ./... 全绿(含 2 个新死锁回归用例)
This commit is contained in:
1 parent
7369a17598
commit
5e59347c40
27 files changed
+491
-54
No files matched your search
@@ -227,6 +227,37 @@ func recordCooldownThrottled(ruleID uint) bool {
|
||||
return true
|
||||
}
|
||||
|
||||
// pruneDeliveriesLocked 按保留期 + 条数上限裁剪投递历史(调用方必须已持锁)。
|
||||
func pruneDeliveriesLocked(db *repository.JsonDB) {
|
||||
cut := 0
|
||||
// 注意:调用方已持写锁,必须用无锁读取版本,否则 RWMutex 会死锁
|
||||
deadline := time.Now().Add(-alertDeliveryRetentionLocked(db))
|
||||
for cut < len(db.AlertDeliveries) && db.AlertDeliveries[cut].CreatedAt.Before(deadline) {
|
||||
cut++
|
||||
}
|
||||
if remain := len(db.AlertDeliveries) - cut; remain > maxAlertDeliveries {
|
||||
cut += remain - maxAlertDeliveries
|
||||
}
|
||||
if cut > 0 {
|
||||
db.AlertDeliveries = append([]models.AlertDelivery{}, db.AlertDeliveries[cut:]...)
|
||||
}
|
||||
}
|
||||
|
||||
// PruneAlertDeliveries 按当前保留期裁剪一次(供保存设置后立即生效)。
|
||||
func PruneAlertDeliveries() {
|
||||
db := repository.GetDB()
|
||||
if db == nil {
|
||||
return
|
||||
}
|
||||
db.Lock()
|
||||
defer db.Unlock()
|
||||
before := len(db.AlertDeliveries)
|
||||
pruneDeliveriesLocked(db)
|
||||
if len(db.AlertDeliveries) != before {
|
||||
db.RequestSave()
|
||||
}
|
||||
}
|
||||
|
||||
func recordDelivery(ruleID uint, ruleName string, count int, status, reason string) {
|
||||
db := repository.GetDB()
|
||||
newID := db.NextID("alertDeliveries")
|
||||
@@ -235,10 +266,9 @@ func recordDelivery(ruleID uint, ruleName string, count int, status, reason stri
|
||||
ID: newID, RuleID: ruleID, RuleName: ruleName,
|
||||
Count: count, Status: status, Reason: reason,
|
||||
})
|
||||
// 只保留最近 50 条
|
||||
if len(db.AlertDeliveries) > 50 {
|
||||
db.AlertDeliveries = db.AlertDeliveries[len(db.AlertDeliveries)-50:]
|
||||
}
|
||||
// 保留策略:按 告警发送历史保留天数(默认 7 天,系统设置可改)+ 条数硬上限兜底。
|
||||
// 原来写死 只留最近 50 条,真实发送记录一小时内就被冷却记录挤光了。
|
||||
pruneDeliveriesLocked(db)
|
||||
db.RequestSave()
|
||||
db.Unlock()
|
||||
}
|
||||
|
||||
@@ -15,8 +15,7 @@ const (
|
||||
// maxAuditLogs 审计日志条数上限。JSON DB 是全量读进内存、全量写盘,
|
||||
// 不设上限的话审计表会把内存和落盘时间一起拖垮。
|
||||
maxAuditLogs = 20000
|
||||
// auditRetention 审计日志保留天数(合规常见要求 90 天)
|
||||
auditRetention = 90 * 24 * time.Hour
|
||||
// 保留期改由系统设置配置(见 services/retention.go)
|
||||
)
|
||||
|
||||
// RecordAudit 写入一条审计日志。
|
||||
@@ -60,7 +59,9 @@ func RecordAudit(entry models.AuditLog) {
|
||||
func pruneAuditLocked(db *repository.JsonDB) {
|
||||
logs := db.AuditLogs
|
||||
cut := 0
|
||||
deadline := time.Now().Add(-auditRetention)
|
||||
// 保留期来自 系统设置->通用配置->日志保留时间,默认 90 天
|
||||
// 注意:调用方已持写锁,必须用无锁读取版本,否则 RWMutex 会死锁
|
||||
deadline := time.Now().Add(-auditLogRetentionLocked(db))
|
||||
|
||||
// 1) 先按保留期裁
|
||||
for cut < len(logs) && logs[cut].CreatedAt.Before(deadline) {
|
||||
@@ -75,6 +76,21 @@ func pruneAuditLocked(db *repository.JsonDB) {
|
||||
}
|
||||
}
|
||||
|
||||
// PruneAuditLogs 按当前保留期裁剪一次审计日志(供保存设置后立即生效)。
|
||||
func PruneAuditLogs() {
|
||||
db := repository.GetDB()
|
||||
if db == nil {
|
||||
return
|
||||
}
|
||||
db.Lock()
|
||||
defer db.Unlock()
|
||||
before := len(db.AuditLogs)
|
||||
pruneAuditLocked(db)
|
||||
if len(db.AuditLogs) != before {
|
||||
db.RequestSave()
|
||||
}
|
||||
}
|
||||
|
||||
// ListAuditLogs 分页查询审计日志(按时间倒序)。
|
||||
func ListAuditLogs(q models.AuditQuery) ([]models.AuditLog, int64) {
|
||||
db := repository.GetDB()
|
||||
|
||||
@@ -3,9 +3,9 @@ package services
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"lokistack/models"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"lokistack/models"
|
||||
"testing"
|
||||
)
|
||||
|
||||
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
// indent 必须与该模板里 `external_labels:` 的层级匹配:
|
||||
// - 主机模板(4 空格)→ 子项 6 空格
|
||||
// - DaemonSet 模板(6 空格)→ 子项 8 空格
|
||||
//
|
||||
// 早先写死 8 空格,主机模式一旦配了标签就会生成非法 YAML,Promtail 直接装不上。
|
||||
func promtailExternalLabels(cluster string, labelsJSON string, indent string) string {
|
||||
lines := []string{fmt.Sprintf("%scluster: %s", indent, yamlScalar(cluster))}
|
||||
|
||||
@@ -11,13 +11,13 @@ import (
|
||||
// 统一一个值要么被限流(COS 503 SlowDown),要么白等(S3 能扛更多)。
|
||||
func TestCompactorDeleteWorkersPerStorageType(t *testing.T) {
|
||||
cases := map[string]int{
|
||||
"cos": 16, // 实测 150 触发 503 SlowDown
|
||||
"obs": 32, // 华为 OBS 保守值
|
||||
"minio": 32, // 自建,取决于磁盘
|
||||
"azure": 32, // 存储账户限流较严
|
||||
"oss": 64, // 阿里云 OSS QPS 较高
|
||||
"cos": 16, // 实测 150 触发 503 SlowDown
|
||||
"obs": 32, // 华为 OBS 保守值
|
||||
"minio": 32, // 自建,取决于磁盘
|
||||
"azure": 32, // 存储账户限流较严
|
||||
"oss": 64, // 阿里云 OSS QPS 较高
|
||||
"gcs": 64,
|
||||
"s3": 96, // AWS S3 限流宽松
|
||||
"s3": 96, // AWS S3 限流宽松
|
||||
"": defaultCompactorDeleteWorkers,
|
||||
"unknown": defaultCompactorDeleteWorkers,
|
||||
}
|
||||
|
||||
@@ -7,10 +7,10 @@ import (
|
||||
"regexp"
|
||||
"time"
|
||||
|
||||
"lokistack/models"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/rest"
|
||||
"k8s.io/client-go/tools/clientcmd"
|
||||
"lokistack/models"
|
||||
)
|
||||
|
||||
// resolveKubeconfig 取集群最终的 kubeconfig 文本(优先内存字段,其次磁盘文件)
|
||||
|
||||
@@ -301,7 +301,6 @@ func patchDeployment(ctx context.Context, cluster *models.Cluster, namespace, na
|
||||
return err
|
||||
}
|
||||
|
||||
|
||||
// podDNSBroken 在指定 deployment 的容器里探测 hostname 是否可解析(wget /ready 连通性)。
|
||||
func podDNSBroken(ctx context.Context, cluster *models.Cluster, namespace, deployment, container, host string) bool {
|
||||
pods, err := listDeploymentPods(ctx, cluster, namespace, deployment)
|
||||
@@ -355,7 +354,6 @@ func listDeploymentPods(ctx context.Context, cluster *models.Cluster, namespace,
|
||||
return names, nil
|
||||
}
|
||||
|
||||
|
||||
// lokiNamespaceOf 从 endpoint 的 svc FQDN 提取 Loki 命名空间(log.ns.svc → ns)。
|
||||
func lokiNamespaceOf(loki *models.LokiDeployment, endpoint string) string {
|
||||
if loki != nil && loki.Namespace != "" {
|
||||
|
||||
@@ -123,12 +123,12 @@ type lokiConfigData struct {
|
||||
CompactorStoreType string
|
||||
// CompactorDeleteWorkers 过期 chunk 删除并发,按对象存储类型取默认值
|
||||
CompactorDeleteWorkers int
|
||||
RingInstanceAddr string // 单副本 127.0.0.1;多副本 0.0.0.0(须可被对端寻址)
|
||||
RingKVStore string // 单副本 inmemory;多副本 memberlist
|
||||
MemberlistBlock string // 多副本时非空,注入 memberlist 顶级配置
|
||||
RulerStorageBlock string // 对象存储时非空,注入 ruler.storage.s3/gcs/azure 嵌套配置
|
||||
MultiReplica bool
|
||||
FilesystemStorage bool
|
||||
RingInstanceAddr string // 单副本 127.0.0.1;多副本 0.0.0.0(须可被对端寻址)
|
||||
RingKVStore string // 单副本 inmemory;多副本 memberlist
|
||||
MemberlistBlock string // 多副本时非空,注入 memberlist 顶级配置
|
||||
RulerStorageBlock string // 对象存储时非空,注入 ruler.storage.s3/gcs/azure 嵌套配置
|
||||
MultiReplica bool
|
||||
FilesystemStorage bool
|
||||
}
|
||||
|
||||
func (d *LokiDeployer) generateLokiConfig() string {
|
||||
@@ -181,20 +181,20 @@ func (d *LokiDeployer) generateLokiConfig() string {
|
||||
}
|
||||
|
||||
data := lokiConfigData{
|
||||
AuthEnabled: d.deployment.AuthEnabled,
|
||||
ReplicationFactor: d.replicationFactor(),
|
||||
RetentionPeriod: retentionHours,
|
||||
StorageConfigBlock: storageBlock,
|
||||
ObjectStore: objectStore,
|
||||
RulerStoreType: rulerStore,
|
||||
AuthEnabled: d.deployment.AuthEnabled,
|
||||
ReplicationFactor: d.replicationFactor(),
|
||||
RetentionPeriod: retentionHours,
|
||||
StorageConfigBlock: storageBlock,
|
||||
ObjectStore: objectStore,
|
||||
RulerStoreType: rulerStore,
|
||||
CompactorStoreType: compactorStore,
|
||||
CompactorDeleteWorkers: compactorDeleteWorkers(d.deployment.StorageConfig),
|
||||
RingInstanceAddr: ringAddr,
|
||||
RingKVStore: ringStore,
|
||||
MemberlistBlock: memberlistBlock,
|
||||
RulerStorageBlock: d.buildRulerStorageBlock(),
|
||||
MultiReplica: d.deployment.Replicas >= 2,
|
||||
FilesystemStorage: objectStore == "filesystem",
|
||||
RingInstanceAddr: ringAddr,
|
||||
RingKVStore: ringStore,
|
||||
MemberlistBlock: memberlistBlock,
|
||||
RulerStorageBlock: d.buildRulerStorageBlock(),
|
||||
MultiReplica: d.deployment.Replicas >= 2,
|
||||
FilesystemStorage: objectStore == "filesystem",
|
||||
}
|
||||
|
||||
tmpl, err := template.New("loki-config").Parse(lokiConfigTemplate)
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
package services
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"lokistack/models"
|
||||
"lokistack/repository"
|
||||
)
|
||||
|
||||
// 平台自身数据的保留期设置(不是 Loki 日志的保留期)。
|
||||
//
|
||||
// 为什么需要可配:不同团队对"告警发送历史留多久 / 审计日志留多久"要求不同 —
|
||||
// 合规场景要 90~180 天,小团队 7 天就够。写死在代码里只能改一次发一次版。
|
||||
//
|
||||
// 注意区分:Loki 的日志保留期是**部署级**配置(组件部署 → 保留策略),
|
||||
// 这里管的是平台自己产生的小数据(投递历史、审计日志)。
|
||||
const (
|
||||
settingKeyAlertDeliveryDays = "alertDeliveryRetentionDays"
|
||||
settingKeyAuditLogDays = "auditLogRetentionDays"
|
||||
|
||||
defaultAlertDeliveryDays = 7
|
||||
defaultAuditLogDays = 90
|
||||
|
||||
// 硬上限:即使配置得很大,也不能让 JSON 文件型 DB 无界增长
|
||||
maxAlertDeliveries = 20000
|
||||
)
|
||||
|
||||
// RetentionSettings 平台数据保留期(天)
|
||||
type RetentionSettings struct {
|
||||
AlertDeliveryDays int `json:"alertDeliveryDays"`
|
||||
AuditLogDays int `json:"auditLogDays"`
|
||||
}
|
||||
|
||||
// GetRetentionSettings 读取保留期配置,未配置时返回默认值。
|
||||
func GetRetentionSettings() RetentionSettings {
|
||||
db := repository.GetDB()
|
||||
if db == nil {
|
||||
return retentionDefaults()
|
||||
}
|
||||
db.RLock()
|
||||
defer db.RUnlock()
|
||||
return retentionFromDBLocked(db)
|
||||
}
|
||||
|
||||
func retentionDefaults() RetentionSettings {
|
||||
return RetentionSettings{
|
||||
AlertDeliveryDays: defaultAlertDeliveryDays,
|
||||
AuditLogDays: defaultAuditLogDays,
|
||||
}
|
||||
}
|
||||
|
||||
// retentionFromDBLocked 从已持有的锁内读取配置,**不会再加锁**。
|
||||
//
|
||||
// 为什么必须单独一个函数:Go 的 sync.RWMutex 不可重入。
|
||||
// 调用方已经 db.Lock() 时再调 GetRetentionSettings()(内部 RLock)会**永久死锁** ——
|
||||
// 实测把整个平台写路径卡死(登录都要写审计,于是登录 28 秒返回空)。
|
||||
func retentionFromDBLocked(db *repository.JsonDB) RetentionSettings {
|
||||
out := retentionDefaults()
|
||||
for _, item := range db.Settings {
|
||||
switch item.Key {
|
||||
case settingKeyAlertDeliveryDays:
|
||||
if v, err := strconv.Atoi(item.Value); err == nil && v > 0 {
|
||||
out.AlertDeliveryDays = v
|
||||
}
|
||||
case settingKeyAuditLogDays:
|
||||
if v, err := strconv.Atoi(item.Value); err == nil && v > 0 {
|
||||
out.AuditLogDays = v
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// SaveRetentionSettings 保存保留期配置(天数)。
|
||||
func SaveRetentionSettings(alertDays, auditDays int) error {
|
||||
if alertDays < 1 || alertDays > 365 {
|
||||
return errors.New("告警发送历史保留天数需在 1~365 之间")
|
||||
}
|
||||
if auditDays < 1 || auditDays > 3650 {
|
||||
return errors.New("审计日志保留天数需在 1~3650 之间")
|
||||
}
|
||||
db := repository.GetDB()
|
||||
if db == nil {
|
||||
return errors.New("数据库未初始化")
|
||||
}
|
||||
db.Lock()
|
||||
upsertSettingLocked(db, settingKeyAlertDeliveryDays, strconv.Itoa(alertDays))
|
||||
upsertSettingLocked(db, settingKeyAuditLogDays, strconv.Itoa(auditDays))
|
||||
db.RequestSave()
|
||||
db.Unlock()
|
||||
|
||||
// 立刻按新策略裁剪一次,避免"改了配置但旧数据还在"
|
||||
PruneAlertDeliveries()
|
||||
PruneAuditLogs()
|
||||
return nil
|
||||
}
|
||||
|
||||
// upsertSettingLocked 写入/更新一条设置(调用方必须已持写锁)。
|
||||
func upsertSettingLocked(db *repository.JsonDB, key, value string) {
|
||||
for i := range db.Settings {
|
||||
if db.Settings[i].Key == key {
|
||||
db.Settings[i].Value = value
|
||||
return
|
||||
}
|
||||
}
|
||||
// 不能用 db.NextID():它内部会 d.mu.Lock(),而调用方已经持有写锁,
|
||||
// sync.Mutex 不可重入 → 永久死锁(这个坑在审计日志里踩过一次)。
|
||||
// 这里在锁内自己算最大 ID + 1。
|
||||
nextID := uint(1)
|
||||
for _, item := range db.Settings {
|
||||
if item.ID >= nextID {
|
||||
nextID = item.ID + 1
|
||||
}
|
||||
}
|
||||
db.Settings = append(db.Settings, models.SystemSetting{
|
||||
ID: nextID,
|
||||
Key: key,
|
||||
Value: value,
|
||||
})
|
||||
}
|
||||
|
||||
// alertDeliveryRetentionLocked 返回投递历史保留时长(调用方必须已持锁)。
|
||||
func alertDeliveryRetentionLocked(db *repository.JsonDB) time.Duration {
|
||||
return time.Duration(retentionFromDBLocked(db).AlertDeliveryDays) * 24 * time.Hour
|
||||
}
|
||||
|
||||
// auditLogRetentionLocked 返回审计日志保留时长(调用方必须已持锁)。
|
||||
func auditLogRetentionLocked(db *repository.JsonDB) time.Duration {
|
||||
return time.Duration(retentionFromDBLocked(db).AuditLogDays) * 24 * time.Hour
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package services
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"lokistack/repository"
|
||||
)
|
||||
|
||||
// 回归:持写锁时裁剪逻辑绝不能再去拿读锁。
|
||||
//
|
||||
// 线上事故:一个 day-retention 配置读取被写成在 db.Lock() 内部调用
|
||||
// GetRetentionSettings()(内部 db.RLock())。Go 的 sync.RWMutex 不可重入,
|
||||
// 于是**第一次写投递历史/审计日志就永久死锁** —— 平台表现是"登录 28 秒返回空",
|
||||
// 因为登录也要写审计。go build / go vet / 单测(不触发写路径)全都查不出来。
|
||||
func TestPruneWhileLockedDoesNotDeadlock(t *testing.T) {
|
||||
db := &repository.JsonDB{}
|
||||
db.Lock()
|
||||
defer db.Unlock()
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
pruneDeliveriesLocked(db)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("pruneDeliveriesLocked 在持有写锁时死锁(内部又去拿了读锁)")
|
||||
}
|
||||
}
|
||||
|
||||
// 同理:审计日志裁剪。
|
||||
func TestPruneAuditWhileLockedDoesNotDeadlock(t *testing.T) {
|
||||
db := &repository.JsonDB{}
|
||||
db.Lock()
|
||||
defer db.Unlock()
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
pruneAuditLocked(db)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("pruneAuditLocked 在持有写锁时死锁(内部又去拿了读锁)")
|
||||
}
|
||||
}
|
||||
@@ -177,8 +177,8 @@ func stripBucketPrefix(endpoint, bucket string) string {
|
||||
|
||||
// bucketPathFor 根据 endpoint 风格决定桶路径(用于建桶 PUT)。
|
||||
//
|
||||
// 1) 区域端点(推荐): cos.ap-hongkong.myqcloud.com → 路径式 /<bucket>
|
||||
// 2) 虚拟主机式: mybucket-1250000000.cos.<region>… → 路径为 /
|
||||
// 1. 区域端点(推荐): cos.ap-hongkong.myqcloud.com → 路径式 /<bucket>
|
||||
// 2. 虚拟主机式: mybucket-1250000000.cos.<region>… → 路径为 /
|
||||
func bucketPathFor(endpoint, bucket string) string {
|
||||
if endpointHasBucket(endpoint, bucket) {
|
||||
return "/"
|
||||
|
||||
Reference in new issue
Block a user