github/sub2api
4
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-20 00:18:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #6424 from cyhhao/fix/bound-ops-system-log-storage

Browse Source 
fix(ops): bound database-backed system log growth
This commit is contained in:
Wesley Liddick
2026-09-08 20:43:54 +08:00
committed by GitHub
parent 43569bb44c 163eb7ffae
commit 95acbf1f03
18 changed files with 418 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/config/config.go
+7 -3
View File
@@ -1613,10 +1613,10 @@ type OpsCleanupConfig struct {
Enabled bool `mapstructure:"enabled"`
Schedule string `mapstructure:"schedule"`
// Retention days (0 disables that cleanup target).
//
// vNext requirement: default 30 days across ops datasets.
// Retention days. Error and metrics targets accept 0 as an explicit truncate;
// system logs require a positive value because their runtime setting is bounded.
ErrorLogRetentionDays int `mapstructure:"error_log_retention_days"`
SystemLogRetentionDays int `mapstructure:"system_log_retention_days"`
MinuteMetricsRetentionDays int `mapstructure:"minute_metrics_retention_days"`
HourlyMetricsRetentionDays int `mapstructure:"hourly_metrics_retention_days"`
}
@@ -2251,6 +2251,7 @@ func setDefaults() {
viper.SetDefault("ops.cleanup.schedule", "0 2 * * *")
// Retention days: vNext defaults to 30 days across ops datasets.
viper.SetDefault("ops.cleanup.error_log_retention_days", 30)
viper.SetDefault("ops.cleanup.system_log_retention_days", 30)
viper.SetDefault("ops.cleanup.minute_metrics_retention_days", 30)
viper.SetDefault("ops.cleanup.hourly_metrics_retention_days", 30)
viper.SetDefault("ops.aggregation.enabled", true)
@@ -3674,6 +3675,9 @@ func (c *Config) Validate() error {
if c.Ops.Cleanup.ErrorLogRetentionDays < 0 {
return fmt.Errorf("ops.cleanup.error_log_retention_days must be non-negative")
}
if c.Ops.Cleanup.Enabled && c.Ops.Cleanup.SystemLogRetentionDays <= 0 {
return fmt.Errorf("ops.cleanup.system_log_retention_days must be positive when ops cleanup is enabled")
}
if c.Ops.Cleanup.MinuteMetricsRetentionDays < 0 {
return fmt.Errorf("ops.cleanup.minute_metrics_retention_days must be non-negative")
}
backend/internal/config/config_test.go
+20
View File
@@ -1191,6 +1191,21 @@ func TestLoadDefaultUsageCleanupConfig(t *testing.T) {
}
}
func TestLoadDefaultOpsCleanupConfig(t *testing.T) {
resetViperWithJWTSecret(t)
cfg, err := Load()
if err != nil {
t.Fatalf("Load() error: %v", err)
}
if !cfg.Ops.Cleanup.Enabled {
t.Fatal("Ops.Cleanup.Enabled = false, want true")
}
if cfg.Ops.Cleanup.SystemLogRetentionDays != 30 {
t.Fatalf("Ops.Cleanup.SystemLogRetentionDays = %d, want 30", cfg.Ops.Cleanup.SystemLogRetentionDays)
}
}
func TestValidateUsageCleanupConfigEnabled(t *testing.T) {
resetViperWithJWTSecret(t)
@@ -2137,6 +2152,11 @@ func TestValidateConfigErrors(t *testing.T) {
mutate: func(c *Config) { c.Ops.Cleanup.ErrorLogRetentionDays = -1 },
wantErr: "ops.cleanup.error_log_retention_days",
},
{
name: "ops cleanup system log retention",
mutate: func(c *Config) { c.Ops.Cleanup.SystemLogRetentionDays = 0 },
wantErr: "ops.cleanup.system_log_retention_days",
},
{
name: "ops cleanup minute retention",
mutate: func(c *Config) { c.Ops.Cleanup.MinuteMetricsRetentionDays = -1 },
backend/internal/service/ops_cleanup_overlay_test.go
+73 -8
View File
@@ -45,7 +45,8 @@ func TestComputeEffective_FallbackToCfgWhenSettingsAbsent(t *testing.T) {
svc.computeEffectiveLocked(context.Background())
if svc.effective != base {
want := opsCleanupEffectiveConfig{OpsCleanupConfig: base, SystemLogRetentionDays: 30}
if svc.effective != want {
t.Fatalf("expected effective == cfg base, got %#v", svc.effective)
}
}
@@ -70,12 +71,15 @@ func TestComputeEffective_SettingsOverridesAll(t *testing.T) {
svc.computeEffectiveLocked(context.Background())
want := config.OpsCleanupConfig{
Enabled: true,
Schedule: "0 * * * *",
ErrorLogRetentionDays: 0,
MinuteMetricsRetentionDays: 7,
HourlyMetricsRetentionDays: 14,
want := opsCleanupEffectiveConfig{
OpsCleanupConfig: config.OpsCleanupConfig{
Enabled: true,
Schedule: "0 * * * *",
ErrorLogRetentionDays: 0,
MinuteMetricsRetentionDays: 7,
HourlyMetricsRetentionDays: 14,
},
SystemLogRetentionDays: 30,
}
if svc.effective != want {
t.Fatalf("effective mismatch:\nwant %#v\n got %#v", want, svc.effective)
@@ -156,11 +160,72 @@ func TestComputeEffective_BadJSONFallsBackToCfg(t *testing.T) {
svc.computeEffectiveLocked(context.Background())
if svc.effective != base {
want := opsCleanupEffectiveConfig{OpsCleanupConfig: base, SystemLogRetentionDays: 30}
if svc.effective != want {
t.Fatalf("expected fallback to cfg on bad JSON, got %#v", svc.effective)
}
}
func TestComputeEffective_RuntimeConfigOwnsSystemLogRetention(t *testing.T) {
repo := newRuntimeSettingRepoStub()
writeAdvancedSettings(t, repo, OpsDataRetentionSettings{
CleanupEnabled: true,
CleanupSchedule: "0 2 * * *",
ErrorLogRetentionDays: 90,
})
repo.values[SettingKeyOpsRuntimeLogConfig] = `{"retention_days":7}`
svc := makeOverlayService(repo, config.OpsCleanupConfig{ErrorLogRetentionDays: 30})
svc.computeEffectiveLocked(context.Background())
if svc.effective.ErrorLogRetentionDays != 90 {
t.Fatalf("error log retention = %d, want 90", svc.effective.ErrorLogRetentionDays)
}
if svc.effective.SystemLogRetentionDays != 7 {
t.Fatalf("system log retention = %d, want 7", svc.effective.SystemLogRetentionDays)
}
}
func TestComputeEffective_InvalidRuntimeConfigDoesNotDiscardAdvancedSettings(t *testing.T) {
repo := newRuntimeSettingRepoStub()
writeAdvancedSettings(t, repo, OpsDataRetentionSettings{
CleanupEnabled: true,
CleanupSchedule: "0 * * * *",
ErrorLogRetentionDays: 14,
})
repo.values[SettingKeyOpsRuntimeLogConfig] = `{invalid-json}`
svc := makeOverlayService(repo, config.OpsCleanupConfig{ErrorLogRetentionDays: 30})
svc.computeEffectiveLocked(context.Background())
if !svc.effective.Enabled || svc.effective.ErrorLogRetentionDays != 14 {
t.Fatalf("advanced settings were discarded: %#v", svc.effective)
}
if svc.effective.SystemLogRetentionDays != 30 {
t.Fatalf("system log retention = %d, want default 30", svc.effective.SystemLogRetentionDays)
}
}
func TestComputeEffective_PartialAdvancedSettingsUseSafeDefaults(t *testing.T) {
repo := newRuntimeSettingRepoStub()
repo.values[SettingKeyOpsAdvancedSettings] = `{"data_retention":{"cleanup_enabled":true}}`
svc := makeOverlayService(repo, config.OpsCleanupConfig{
Schedule: "0 3 * * *",
ErrorLogRetentionDays: 90,
MinuteMetricsRetentionDays: 90,
HourlyMetricsRetentionDays: 90,
})
svc.computeEffectiveLocked(context.Background())
if svc.effective.Schedule != opsCleanupDefaultSchedule ||
svc.effective.ErrorLogRetentionDays != 30 ||
svc.effective.MinuteMetricsRetentionDays != 30 ||
svc.effective.HourlyMetricsRetentionDays != 30 {
t.Fatalf("partial settings did not use safe defaults: %#v", svc.effective)
}
}
// 验证 OpsService.UpdateOpsAdvancedSettings 写入后会调用 cleanupReloader.Reload。
type fakeCleanupReloader struct {
calls int
backend/internal/service/ops_cleanup_service.go
+55 -31
View File
@@ -4,7 +4,6 @@ import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
@@ -59,11 +58,16 @@ type OpsCleanupService struct {
cron *cron.Cron
started bool
stopped bool
effective config.OpsCleanupConfig
effective opsCleanupEffectiveConfig
warnNoRedisOnce sync.Once
}
type opsCleanupEffectiveConfig struct {
config.OpsCleanupConfig
SystemLogRetentionDays int
}
func NewOpsCleanupService(
opsRepo OpsRepository,
db *sql.DB,
@@ -166,9 +170,10 @@ func (s *OpsCleanupService) applyScheduleLocked(ctx context.Context) error {
c.Start()
s.cron = c
logger.LegacyPrintf("service.ops_cleanup",
"[OpsCleanup] scheduled (schedule=%q tz=%s retention_days=err:%d/min:%d/hour:%d)",
"[OpsCleanup] scheduled (schedule=%q tz=%s retention_days=err:%d/system:%d/min:%d/hour:%d)",
schedule, loc.String(),
s.effective.ErrorLogRetentionDays,
s.effective.SystemLogRetentionDays,
s.effective.MinuteMetricsRetentionDays,
s.effective.HourlyMetricsRetentionDays,
)
@@ -198,11 +203,14 @@ func (s *OpsCleanupService) Reload(ctx context.Context) error {
// - Schedule：settings 非空时覆盖，否则保留 cfg
// - *RetentionDays：settings >=0 时覆盖（包括 0=TRUNCATE），<0 沿用 cfg
//
// 若 settings 表无该 key（ErrSettingNotFound）或解析失败，整体 fallback 到 cfg.Ops.Cleanup。
// System log retention has a separate owner: settings.ops_runtime_log_config.retention_days.
// Missing or invalid settings fall back independently instead of disabling all cleanup overlays.
func (s *OpsCleanupService) computeEffectiveLocked(ctx context.Context) {
base := config.OpsCleanupConfig{}
base := opsCleanupEffectiveConfig{
SystemLogRetentionDays: defaultOpsRuntimeLogConfig(s.cfg).RetentionDays,
}
if s.cfg != nil {
base = s.cfg.Ops.Cleanup
base.OpsCleanupConfig = s.cfg.Ops.Cleanup
}
defer func() { s.effective = base }()
@@ -212,38 +220,54 @@ func (s *OpsCleanupService) computeEffectiveLocked(ctx context.Context) {
if ctx == nil {
ctx = context.Background()
}
raw, err := s.settingRepo.GetValue(ctx, SettingKeyOpsAdvancedSettings)
values, err := s.settingRepo.GetMultiple(ctx, []string{
SettingKeyOpsAdvancedSettings,
SettingKeyOpsRuntimeLogConfig,
})
if err != nil {
if !errors.Is(err, ErrSettingNotFound) {
logger.LegacyPrintf("service.ops_cleanup",
"[OpsCleanup] read advanced settings failed, using cfg: %v", err)
}
return
}
var adv OpsAdvancedSettings
if err := json.Unmarshal([]byte(raw), &adv); err != nil {
logger.LegacyPrintf("service.ops_cleanup",
"[OpsCleanup] parse advanced settings failed, using cfg: %v", err)
"[OpsCleanup] read cleanup settings failed, using defaults: %v", err)
return
}
dr := adv.DataRetention
base.Enabled = dr.CleanupEnabled
if sched := strings.TrimSpace(dr.CleanupSchedule); sched != "" {
base.Schedule = sched
if raw, ok := values[SettingKeyOpsAdvancedSettings]; ok {
adv := defaultOpsAdvancedSettingsForConfig(s.cfg)
if err := json.Unmarshal([]byte(raw), adv); err != nil {
logger.LegacyPrintf("service.ops_cleanup",
"[OpsCleanup] parse advanced settings failed, using cfg: %v", err)
} else {
dr := adv.DataRetention
base.Enabled = dr.CleanupEnabled
if sched := strings.TrimSpace(dr.CleanupSchedule); sched != "" {
base.Schedule = sched
}
if dr.ErrorLogRetentionDays >= 0 {
base.ErrorLogRetentionDays = dr.ErrorLogRetentionDays
}
if dr.MinuteMetricsRetentionDays >= 0 {
base.MinuteMetricsRetentionDays = dr.MinuteMetricsRetentionDays
}
if dr.HourlyMetricsRetentionDays >= 0 {
base.HourlyMetricsRetentionDays = dr.HourlyMetricsRetentionDays
}
}
}
if dr.ErrorLogRetentionDays >= 0 {
base.ErrorLogRetentionDays = dr.ErrorLogRetentionDays
}
if dr.MinuteMetricsRetentionDays >= 0 {
base.MinuteMetricsRetentionDays = dr.MinuteMetricsRetentionDays
}
if dr.HourlyMetricsRetentionDays >= 0 {
base.HourlyMetricsRetentionDays = dr.HourlyMetricsRetentionDays
if raw, ok := values[SettingKeyOpsRuntimeLogConfig]; ok {
defaults := defaultOpsRuntimeLogConfig(s.cfg)
runtimeCfg := *defaults
if err := json.Unmarshal([]byte(raw), &runtimeCfg); err != nil {
logger.LegacyPrintf("service.ops_cleanup",
"[OpsCleanup] parse runtime log config failed, using default system-log retention: %v", err)
} else {
normalizeOpsRuntimeLogConfig(&runtimeCfg, defaults)
if runtimeCfg.RetentionDays >= 1 && runtimeCfg.RetentionDays <= 3650 {
base.SystemLogRetentionDays = runtimeCfg.RetentionDays
}
}
}
}
// snapshotEffective 取一份 effective 副本（runCleanupOnce 等读路径使用）。
func (s *OpsCleanupService) snapshotEffective() config.OpsCleanupConfig {
func (s *OpsCleanupService) snapshotEffective() opsCleanupEffectiveConfig {
s.mu.Lock()
defer s.mu.Unlock()
return s.effective
@@ -305,8 +329,8 @@ func (s *OpsCleanupService) runCleanupOnce(ctx context.Context) (opsCleanupDelet
{effective.ErrorLogRetentionDays, "ops_error_logs", "created_at", false, &out.errorLogs},
{effective.ErrorLogRetentionDays, "ops_ingress_reject_aggregates", "bucket_start", false, &out.ingressRejects},
{effective.ErrorLogRetentionDays, "ops_alert_events", "created_at", false, &out.alertEvents},
{effective.ErrorLogRetentionDays, "ops_system_logs", "created_at", false, &out.systemLogs},
{effective.ErrorLogRetentionDays, "ops_system_log_cleanup_audits", "created_at", false, &out.logAudits},
{effective.SystemLogRetentionDays, "ops_system_logs", "created_at", false, &out.systemLogs},
{effective.SystemLogRetentionDays, "ops_system_log_cleanup_audits", "created_at", false, &out.logAudits},
{effective.MinuteMetricsRetentionDays, "ops_system_metrics", "created_at", false, &out.systemMetrics},
{effective.HourlyMetricsRetentionDays, "ops_metrics_hourly", "bucket_start", false, &out.hourlyPreagg},
{effective.HourlyMetricsRetentionDays, "ops_metrics_daily", "bucket_date", true, &out.dailyPreagg},
backend/internal/service/ops_log_runtime.go
+36 -14
View File
@@ -15,13 +15,14 @@ import (
func defaultOpsRuntimeLogConfig(cfg *config.Config) *OpsRuntimeLogConfig {
out := &OpsRuntimeLogConfig{
Level: "info",
EnableSampling: false,
SamplingInitial: 100,
SamplingNext: 100,
Caller: true,
StacktraceLevel: "error",
RetentionDays: 30,
Level: "info",
PersistAccessLogs: false,
EnableSampling: false,
SamplingInitial: 100,
SamplingNext: 100,
Caller: true,
StacktraceLevel: "error",
RetentionDays: 30,
}
if cfg == nil {
return out
@@ -32,8 +33,8 @@ func defaultOpsRuntimeLogConfig(cfg *config.Config) *OpsRuntimeLogConfig {
out.SamplingNext = cfg.Log.Sampling.Thereafter
out.Caller = cfg.Log.Caller
out.StacktraceLevel = strings.ToLower(strings.TrimSpace(cfg.Log.StacktraceLevel))
if cfg.Ops.Cleanup.ErrorLogRetentionDays > 0 {
out.RetentionDays = cfg.Ops.Cleanup.ErrorLogRetentionDays
if cfg.Ops.Cleanup.SystemLogRetentionDays > 0 {
out.RetentionDays = cfg.Ops.Cleanup.SystemLogRetentionDays
}
return out
}
@@ -144,7 +145,7 @@ func (s *OpsService) UpdateRuntimeLogConfig(ctx context.Context, req *OpsRuntime
return nil, err
}
if err := applyOpsRuntimeLogConfig(&next); err != nil {
if err := s.applyRuntimeLogConfig(&next); err != nil {
s.auditRuntimeLogConfigFailure(operatorID, oldCfg, &next, "apply_failed: "+err.Error())
return nil, err
}
@@ -159,12 +160,13 @@ func (s *OpsService) UpdateRuntimeLogConfig(ctx context.Context, req *OpsRuntime
}
if err := s.settingRepo.Set(ctx, SettingKeyOpsRuntimeLogConfig, string(encoded)); err != nil {
// 存储失败时回滚到旧配置，避免内存状态与持久化状态不一致。
_ = applyOpsRuntimeLogConfig(oldCfg)
_ = s.applyRuntimeLogConfig(oldCfg)
s.auditRuntimeLogConfigFailure(operatorID, oldCfg, &next, "persist_failed: "+err.Error())
return nil, err
}
s.auditRuntimeLogConfigChange(operatorID, oldCfg, &next, "updated")
s.reloadOpsCleanup(ctx)
return &next, nil
}
@@ -191,14 +193,14 @@ func (s *OpsService) ResetRuntimeLogConfig(ctx context.Context, operatorID int64
s.auditRuntimeLogConfigFailure(operatorID, oldCfg, resetCfg, "reset_validation_failed: "+err.Error())
return nil, err
}
if err := applyOpsRuntimeLogConfig(resetCfg); err != nil {
if err := s.applyRuntimeLogConfig(resetCfg); err != nil {
s.auditRuntimeLogConfigFailure(operatorID, oldCfg, resetCfg, "reset_apply_failed: "+err.Error())
return nil, err
}
// 清理 runtime 覆盖配置，回退到 env/yaml baseline。
if err := s.settingRepo.Delete(ctx, SettingKeyOpsRuntimeLogConfig); err != nil && !errors.Is(err, ErrSettingNotFound) {
_ = applyOpsRuntimeLogConfig(oldCfg)
_ = s.applyRuntimeLogConfig(oldCfg)
s.auditRuntimeLogConfigFailure(operatorID, oldCfg, resetCfg, "reset_persist_failed: "+err.Error())
return nil, err
}
@@ -209,9 +211,29 @@ func (s *OpsService) ResetRuntimeLogConfig(ctx context.Context, operatorID int64
resetCfg.UpdatedByUserID = operatorID
s.auditRuntimeLogConfigChange(operatorID, oldCfg, resetCfg, "reset")
s.reloadOpsCleanup(ctx)
return resetCfg, nil
}
func (s *OpsService) applyRuntimeLogConfig(cfg *OpsRuntimeLogConfig) error {
if err := applyOpsRuntimeLogConfig(cfg); err != nil {
return err
}
if s != nil && s.systemLogSink != nil {
s.systemLogSink.SetPersistAccessLogs(cfg.PersistAccessLogs)
}
return nil
}
func (s *OpsService) reloadOpsCleanup(ctx context.Context) {
if s == nil || s.cleanupReloader == nil {
return
}
if err := s.cleanupReloader.Reload(ctx); err != nil {
logger.LegacyPrintf("service.ops_log_runtime", "[OpsLogRuntime] cleanup reload failed: %v", err)
}
}
func applyOpsRuntimeLogConfig(cfg *OpsRuntimeLogConfig) error {
if cfg == nil {
return fmt.Errorf("nil runtime log config")
@@ -238,7 +260,7 @@ func (s *OpsService) applyRuntimeLogConfigOnStartup(ctx context.Context) {
if err != nil {
return
}
_ = applyOpsRuntimeLogConfig(cfg)
_ = s.applyRuntimeLogConfig(cfg)
}
func (s *OpsService) auditRuntimeLogConfigChange(operatorID int64, oldCfg *OpsRuntimeLogConfig, newCfg *OpsRuntimeLogConfig, action string) {
backend/internal/service/ops_log_runtime_test.go
+55 -22
View File
@@ -21,6 +21,15 @@ type runtimeSettingRepoStub struct {
deleteFn func(key string) error
}
type runtimeCleanupReloader struct {
calls int
}
func (r *runtimeCleanupReloader) Reload(context.Context) error {
r.calls++
return nil
}
func newRuntimeSettingRepoStub() *runtimeSettingRepoStub {
return &runtimeSettingRepoStub{
values: map[string]string{},
@@ -181,7 +190,8 @@ func TestResetRuntimeLogConfig_ShouldFallbackToBaseline(t *testing.T) {
},
Ops: config.OpsConfig{
Cleanup: config.OpsCleanupConfig{
ErrorLogRetentionDays: 45,
ErrorLogRetentionDays: 7,
SystemLogRetentionDays: 45,
},
},
},
@@ -282,7 +292,8 @@ func TestUpdateRuntimeLogConfig_PersistFailureRollback(t *testing.T) {
}
svc := &OpsService{
settingRepo: repo,
settingRepo: repo,
systemLogSink: &OpsSystemLogSink{},
cfg: &config.Config{
Log: config.LogConfig{
Level: "info",
@@ -311,13 +322,14 @@ func TestUpdateRuntimeLogConfig_PersistFailureRollback(t *testing.T) {
}
_, err := svc.UpdateRuntimeLogConfig(context.Background(), &OpsRuntimeLogConfig{
Level: "debug",
EnableSampling: false,
SamplingInitial: 100,
SamplingNext: 100,
Caller: true,
StacktraceLevel: "error",
RetentionDays: 30,
Level: "debug",
PersistAccessLogs: true,
EnableSampling: false,
SamplingInitial: 100,
SamplingNext: 100,
Caller: true,
StacktraceLevel: "error",
RetentionDays: 30,
}, 5)
if err == nil {
t.Fatalf("expected persist error")
@@ -326,15 +338,19 @@ func TestUpdateRuntimeLogConfig_PersistFailureRollback(t *testing.T) {
if logger.CurrentLevel() != "info" {
t.Fatalf("logger level should rollback to info, got %s", logger.CurrentLevel())
}
if svc.systemLogSink.persistAccessLogs.Load() {
t.Fatal("access-log persistence should rollback after setting write failure")
}
}
func TestApplyRuntimeLogConfigOnStartup(t *testing.T) {
repo := newRuntimeSettingRepoStub()
cfgRaw := `{"level":"debug","enable_sampling":false,"sampling_initial":100,"sampling_thereafter":100,"caller":true,"stacktrace_level":"error","retention_days":30}`
cfgRaw := `{"level":"debug","persist_access_logs":true,"enable_sampling":false,"sampling_initial":100,"sampling_thereafter":100,"caller":true,"stacktrace_level":"error","retention_days":30}`
repo.values[SettingKeyOpsRuntimeLogConfig] = cfgRaw
svc := &OpsService{
settingRepo: repo,
settingRepo: repo,
systemLogSink: &OpsSystemLogSink{},
cfg: &config.Config{
Log: config.LogConfig{
Level: "info",
@@ -366,6 +382,9 @@ func TestApplyRuntimeLogConfigOnStartup(t *testing.T) {
if logger.CurrentLevel() != "debug" {
t.Fatalf("expected startup apply debug, got %s", logger.CurrentLevel())
}
if !svc.systemLogSink.persistAccessLogs.Load() {
t.Fatal("startup config did not enable access-log persistence")
}
}
func TestDefaultNormalizeAndValidateRuntimeLogConfig(t *testing.T) {
@@ -382,13 +401,17 @@ func TestDefaultNormalizeAndValidateRuntimeLogConfig(t *testing.T) {
},
Ops: config.OpsConfig{
Cleanup: config.OpsCleanupConfig{
ErrorLogRetentionDays: 7,
ErrorLogRetentionDays: 7,
SystemLogRetentionDays: 14,
},
},
})
if defaults.Level != "debug" || defaults.StacktraceLevel != "fatal" || defaults.RetentionDays != 7 {
if defaults.Level != "debug" || defaults.StacktraceLevel != "fatal" || defaults.RetentionDays != 14 {
t.Fatalf("unexpected defaults: %+v", defaults)
}
if defaults.PersistAccessLogs {
t.Fatal("access-log persistence should be disabled by default")
}
cfg := &OpsRuntimeLogConfig{
Level: " ",
@@ -403,7 +426,7 @@ func TestDefaultNormalizeAndValidateRuntimeLogConfig(t *testing.T) {
if cfg.Level != "debug" || cfg.StacktraceLevel != "fatal" {
t.Fatalf("normalize level/stacktrace failed: %+v", cfg)
}
if cfg.SamplingInitial != 50 || cfg.SamplingNext != 20 || cfg.RetentionDays != 7 {
if cfg.SamplingInitial != 50 || cfg.SamplingNext != 20 || cfg.RetentionDays != 14 {
t.Fatalf("normalize numeric defaults failed: %+v", cfg)
}
if err := validateOpsRuntimeLogConfig(cfg); err != nil {
@@ -489,8 +512,11 @@ func TestUpdateRuntimeLogConfig_PreconditionErrors(t *testing.T) {
func TestUpdateRuntimeLogConfig_Success(t *testing.T) {
repo := newRuntimeSettingRepoStub()
reloader := &runtimeCleanupReloader{}
svc := &OpsService{
settingRepo: repo,
settingRepo: repo,
systemLogSink: &OpsSystemLogSink{},
cleanupReloader: reloader,
cfg: &config.Config{
Log: config.LogConfig{
Level: "info",
@@ -519,13 +545,14 @@ func TestUpdateRuntimeLogConfig_Success(t *testing.T) {
}
next, err := svc.UpdateRuntimeLogConfig(context.Background(), &OpsRuntimeLogConfig{
Level: "debug",
EnableSampling: false,
SamplingInitial: 100,
SamplingNext: 100,
Caller: true,
StacktraceLevel: "error",
RetentionDays: 30,
Level: "debug",
PersistAccessLogs: true,
EnableSampling: false,
SamplingInitial: 100,
SamplingNext: 100,
Caller: true,
StacktraceLevel: "error",
RetentionDays: 30,
}, 2)
if err != nil {
t.Fatalf("UpdateRuntimeLogConfig() error: %v", err)
@@ -536,6 +563,12 @@ func TestUpdateRuntimeLogConfig_Success(t *testing.T) {
if logger.CurrentLevel() != "debug" {
t.Fatalf("expected applied level debug, got %s", logger.CurrentLevel())
}
if !svc.systemLogSink.persistAccessLogs.Load() {
t.Fatal("runtime update did not enable access-log persistence")
}
if reloader.calls != 1 {
t.Fatalf("cleanup reload calls = %d, want 1", reloader.calls)
}
}
func TestResetRuntimeLogConfig_IgnoreNotFoundDelete(t *testing.T) {
backend/internal/service/ops_runtime_snapshot_test.go
+39 -4
View File
@@ -114,28 +114,60 @@ func TestOpsRuntimeSettingsAdministrativeUpdatesAreImmediatelyVisible(t *testing
}
func TestOpsRuntimeSettingsBackgroundRefreshConverges(t *testing.T) {
repo := &opsRuntimeRefreshRepo{values: map[string]string{SettingKeyOpsMonitoringEnabled: "false"}}
svc := &OpsService{settingRepo: repo}
repo := &opsRuntimeRefreshRepo{values: map[string]string{
SettingKeyOpsMonitoringEnabled: "false",
SettingKeyOpsRuntimeLogConfig: `{"persist_access_logs":false}`,
}}
sink := &OpsSystemLogSink{}
svc := &OpsService{settingRepo: repo, systemLogSink: sink}
svc.initRuntimeSettings(context.Background())
if svc.IsMonitoringEnabled(context.Background()) {
t.Fatal("initial monitoring state = true, want false")
}
repo.set(SettingKeyOpsMonitoringEnabled, "true")
repo.set(SettingKeyOpsRuntimeLogConfig, `{"persist_access_logs":true}`)
svc.startRuntimeSettingsRefresh(context.Background(), 5*time.Millisecond, 0, 50*time.Millisecond)
t.Cleanup(svc.StopRuntimeSettingsRefresh)
waitForOpsRefresh(t, time.Second, func() bool {
return svc.IsMonitoringEnabled(context.Background()) && svc.RuntimeSettingsRefreshHealth().SuccessTotal > 0
return svc.IsMonitoringEnabled(context.Background()) &&
sink.persistAccessLogs.Load() &&
svc.RuntimeSettingsRefreshHealth().SuccessTotal > 0
})
}
func TestOpsRuntimeSettingsRefreshUsesSafeAccessLogDefault(t *testing.T) {
repo := &opsRuntimeRefreshRepo{values: map[string]string{
SettingKeyOpsRuntimeLogConfig: `{"persist_access_logs":true}`,
}}
sink := &OpsSystemLogSink{}
svc := &OpsService{settingRepo: repo, systemLogSink: sink}
svc.initRuntimeSettings(context.Background())
if !sink.persistAccessLogs.Load() {
t.Fatal("valid runtime config did not enable access-log persistence")
}
repo.set(SettingKeyOpsRuntimeLogConfig, `{invalid-json}`)
if err := svc.RefreshRuntimeSettings(context.Background()); err != nil {
t.Fatalf("RefreshRuntimeSettings() error = %v", err)
}
if sink.persistAccessLogs.Load() {
t.Fatal("invalid runtime config should fail closed for access-log persistence")
}
}
func TestOpsRuntimeSettingsRefreshFailuresKeepLastKnownGoodSnapshot(t *testing.T) {
repo := &opsRuntimeRefreshRepo{values: map[string]string{
SettingKeyOpsMonitoringEnabled: "false",
SettingKeyOpsAdvancedSettings: `{"ignore_no_available_accounts":true}`,
SettingKeyOpsRuntimeLogConfig: `{"persist_access_logs":true}`,
}}
svc := &OpsService{settingRepo: repo}
sink := &OpsSystemLogSink{}
svc := &OpsService{settingRepo: repo, systemLogSink: sink}
svc.initRuntimeSettings(context.Background())
if !sink.persistAccessLogs.Load() {
t.Fatal("initial access-log setting was not applied")
}
repo.fail.Store(true)
svc.startRuntimeSettingsRefresh(context.Background(), 5*time.Millisecond, 0, 50*time.Millisecond)
t.Cleanup(svc.StopRuntimeSettingsRefresh)
@@ -149,6 +181,9 @@ func TestOpsRuntimeSettingsRefreshFailuresKeepLastKnownGoodSnapshot(t *testing.T
if !svc.OpsAdvancedSettingsSnapshot().IgnoreNoAvailableAccounts {
t.Fatal("failed refresh overwrote last known advanced settings")
}
if !sink.persistAccessLogs.Load() {
t.Fatal("failed refresh overwrote last known access-log setting")
}
}
func TestOpsRuntimeSettingsRefreshStopEndsLifecycle(t *testing.T) {
backend/internal/service/ops_service.go
+17 -4
View File
@@ -177,7 +177,7 @@ func (s *OpsService) initRuntimeSettings(ctx context.Context) {
if s == nil {
return
}
defaults := defaultOpsAdvancedSettings()
defaults := defaultOpsAdvancedSettingsForConfig(s.cfg)
s.runtimeSettings.Store(&opsRuntimeSettingsSnapshot{monitoringEnabled: true, advanced: *defaults})
_ = s.RefreshRuntimeSettings(ctx)
}
@@ -198,6 +198,7 @@ func (s *OpsService) RefreshRuntimeSettings(ctx context.Context) error {
values, err := s.settingRepo.GetMultiple(ctx, []string{
SettingKeyOpsMonitoringEnabled,
SettingKeyOpsAdvancedSettings,
SettingKeyOpsRuntimeLogConfig,
})
if err != nil {
return err
@@ -207,15 +208,27 @@ func (s *OpsService) RefreshRuntimeSettings(ctx context.Context) error {
if raw, ok := values[SettingKeyOpsMonitoringEnabled]; ok {
monitoringEnabled = parseOpsMonitoringEnabled(raw)
}
advanced := defaultOpsAdvancedSettings()
advanced := defaultOpsAdvancedSettingsForConfig(s.cfg)
if raw, ok := values[SettingKeyOpsAdvancedSettings]; ok {
if err := json.Unmarshal([]byte(raw), advanced); err != nil {
advanced = defaultOpsAdvancedSettings()
advanced = defaultOpsAdvancedSettingsForConfig(s.cfg)
}
}
normalizeOpsAdvancedSettings(advanced)
s.runtimeSettings.Store(&opsRuntimeSettingsSnapshot{monitoringEnabled: monitoringEnabled, advanced: *advanced})
if s.systemLogSink != nil {
persistAccessLogs := false
if raw, ok := values[SettingKeyOpsRuntimeLogConfig]; ok {
var runtimeCfg struct {
PersistAccessLogs bool `json:"persist_access_logs"`
}
if json.Unmarshal([]byte(raw), &runtimeCfg) == nil {
persistAccessLogs = runtimeCfg.PersistAccessLogs
}
}
s.systemLogSink.SetPersistAccessLogs(persistAccessLogs)
}
return nil
}
@@ -360,7 +373,7 @@ func (s *OpsService) SetMonitoringEnabled(enabled bool) {
}
s.runtimeSettingsMu.Lock()
current := s.runtimeSettings.Load()
next := &opsRuntimeSettingsSnapshot{monitoringEnabled: enabled, advanced: *defaultOpsAdvancedSettings()}
next := &opsRuntimeSettingsSnapshot{monitoringEnabled: enabled, advanced: *defaultOpsAdvancedSettingsForConfig(s.cfg)}
if current != nil {
next.advanced = current.advanced
}
backend/internal/service/ops_settings.go
+15 -2
View File
@@ -4,9 +4,11 @@ import (
"context"
"encoding/json"
"errors"
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
"strings"
"time"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
)
const (
@@ -360,7 +362,9 @@ func (s *OpsService) UpdateOpsAlertRuntimeSettings(ctx context.Context, cfg *Ops
func defaultOpsAdvancedSettings() *OpsAdvancedSettings {
return &OpsAdvancedSettings{
DataRetention: OpsDataRetentionSettings{
CleanupEnabled: false,
// Keep the DB-backed default aligned with ops.cleanup.enabled so a
// first settings read cannot silently disable configured retention.
CleanupEnabled: true,
CleanupSchedule: opsCleanupDefaultSchedule,
ErrorLogRetentionDays: 30,
MinuteMetricsRetentionDays: 30,
@@ -382,6 +386,14 @@ func defaultOpsAdvancedSettings() *OpsAdvancedSettings {
}
}
func defaultOpsAdvancedSettingsForConfig(cfg *config.Config) *OpsAdvancedSettings {
defaults := defaultOpsAdvancedSettings()
if cfg != nil {
defaults.DataRetention.CleanupEnabled = cfg.Ops.Cleanup.Enabled
}
return defaults
}
func normalizeOpsAdvancedSettings(cfg *OpsAdvancedSettings) {
if cfg == nil {
return
@@ -456,6 +468,7 @@ func (s *OpsService) OpsAdvancedSettingsSnapshot() OpsAdvancedSettings {
if snapshot := s.runtimeSettings.Load(); snapshot != nil {
return snapshot.advanced
}
return *defaultOpsAdvancedSettingsForConfig(s.cfg)
}
return *defaultOpsAdvancedSettings()
}
backend/internal/service/ops_settings_advanced_test.go
+18
View File
@@ -23,11 +23,29 @@ func TestGetOpsAdvancedSettings_DefaultSnapshotHidesOpenAITokenStats(t *testing.
if !cfg.DisplayAlertEvents {
t.Fatalf("DisplayAlertEvents = false, want true by default")
}
if !cfg.DataRetention.CleanupEnabled {
t.Fatal("data cleanup should be enabled by default")
}
if repo.getValueCalls != 0 || repo.getMultipleCalls != 0 {
t.Fatalf("hot-path snapshot read touched repository: get=%d get_multiple=%d", repo.getValueCalls, repo.getMultipleCalls)
}
}
func TestGetOpsAdvancedSettings_DefaultCleanupFollowsDeploymentConfig(t *testing.T) {
svc := &OpsService{cfg: &config.Config{Ops: config.OpsConfig{
Cleanup: config.OpsCleanupConfig{Enabled: false},
}}}
svc.initRuntimeSettings(context.Background())
cfg, err := svc.GetOpsAdvancedSettings(context.Background())
if err != nil {
t.Fatalf("GetOpsAdvancedSettings() error = %v", err)
}
if cfg.DataRetention.CleanupEnabled {
t.Fatal("data cleanup should follow the disabled deployment baseline")
}
}
func TestUpdateOpsAdvancedSettings_PersistsOpenAITokenStatsVisibility(t *testing.T) {
repo := newRuntimeSettingRepoStub()
svc := &OpsService{settingRepo: repo}
backend/internal/service/ops_settings_models.go
+8 -6
View File
@@ -69,12 +69,14 @@ type OpsMetricThresholds struct {
}
type OpsRuntimeLogConfig struct {
Level string `json:"level"`
EnableSampling bool `json:"enable_sampling"`
SamplingInitial int `json:"sampling_initial"`
SamplingNext int `json:"sampling_thereafter"`
Caller bool `json:"caller"`
StacktraceLevel string `json:"stacktrace_level"`
Level string `json:"level"`
PersistAccessLogs bool `json:"persist_access_logs"`
EnableSampling bool `json:"enable_sampling"`
SamplingInitial int `json:"sampling_initial"`
SamplingNext int `json:"sampling_thereafter"`
Caller bool `json:"caller"`
StacktraceLevel string `json:"stacktrace_level"`
// RetentionDays is the authoritative retention for ops_system_logs.
RetentionDays int `json:"retention_days"`
Source string `json:"source,omitempty"`
UpdatedAt string `json:"updated_at,omitempty"`
backend/internal/service/ops_system_log_sink.go
+15 -5
View File
@@ -42,10 +42,11 @@ type OpsSystemLogSink struct {
cancel context.CancelFunc
wg sync.WaitGroup
droppedCount uint64
writeFailed uint64
writtenCount uint64
totalDelayNs uint64
droppedCount uint64
writeFailed uint64
writtenCount uint64
totalDelayNs uint64
persistAccessLogs atomic.Bool
lastError atomic.Value
}
@@ -147,6 +148,15 @@ func (s *OpsSystemLogSink) WriteLogEvent(event *logger.LogEvent) {
}
}
// SetPersistAccessLogs controls whether high-volume request access logs are
// copied into PostgreSQL. Warning/error and audit events are always retained.
func (s *OpsSystemLogSink) SetPersistAccessLogs(enabled bool) {
if s == nil {
return
}
s.persistAccessLogs.Store(enabled)
}
func (s *OpsSystemLogSink) shouldIndex(event *logger.LogEvent) bool {
if event != nil && event.Fields != nil {
if skip, _ := event.Fields[logger.OpsSystemLogSkipField].(bool); skip {
@@ -167,7 +177,7 @@ func (s *OpsSystemLogSink) shouldIndex(event *logger.LogEvent) bool {
}
}
if strings.Contains(component, "http.access") {
return true
return s.persistAccessLogs.Load()
}
if strings.Contains(component, "audit") {
return true
backend/internal/service/ops_system_log_sink_test.go
+26 -4
View File
@@ -32,9 +32,9 @@ func TestOpsSystemLogSink_ShouldIndex(t *testing.T) {
want: true,
},
{
name: "access component",
name: "access component disabled by default",
event: &logger.LogEvent{Level: "info", Component: "http.access"},
want: true,
want: false,
},
{
name: "rejected access excluded from database sink",
@@ -46,13 +46,13 @@ func TestOpsSystemLogSink_ShouldIndex(t *testing.T) {
want: false,
},
{
name: "access component from fields (real zap path)",
name: "access component from fields disabled by default",
event: &logger.LogEvent{
Level: "info",
Component: "",
Fields: map[string]any{"component": "http.access"},
},
want: true,
want: false,
},
{
name: "audit component",
@@ -82,6 +82,28 @@ func TestOpsSystemLogSink_ShouldIndex(t *testing.T) {
}
}
func TestOpsSystemLogSink_ShouldIndexAccessLogsOnlyWhenEnabled(t *testing.T) {
sink := &OpsSystemLogSink{}
sink.SetPersistAccessLogs(true)
if !sink.shouldIndex(&logger.LogEvent{Level: "info", Component: "http.access"}) {
t.Fatal("access log should be indexed after explicit opt-in")
}
if !sink.shouldIndex(&logger.LogEvent{
Level: "info",
Fields: map[string]any{"component": "http.access"},
}) {
t.Fatal("field-based access log should be indexed after explicit opt-in")
}
if sink.shouldIndex(&logger.LogEvent{
Level: "info",
Component: "http.access",
Fields: map[string]any{logger.OpsSystemLogSkipField: true},
}) {
t.Fatal("explicit skip marker must override access-log persistence")
}
}
func TestOpsSystemLogSink_WriteLogEvent_ShouldDropWhenQueueFull(t *testing.T) {
sink := &OpsSystemLogSink{
queue: make(chan *logger.LogEvent, 1),
frontend/src/api/admin/ops.ts
+1
View File
@@ -814,6 +814,7 @@ export interface OpsAggregationSettings {
export interface OpsRuntimeLogConfig {
level: 'debug' | 'info' | 'warn' | 'error'
persist_access_logs: boolean
enable_sampling: boolean
sampling_initial: number
sampling_thereafter: number
frontend/src/i18n/locales/en/admin/ops.ts
+3
View File
@@ -42,8 +42,11 @@ export default {
samplingInitial: 'Sampling initial',
samplingThereafter: 'Sampling thereafter',
retentionDays: 'Retention days',
retentionDaysHint: 'Applied by the scheduled data-cleanup job.',
caller: 'caller',
sampling: 'sampling',
persistAccessLogs: 'Store access logs in database',
persistAccessLogsHint: 'Disabled by default because access logs add one indexed database row per request. Warning, error, and audit logs are always stored.',
saveAndApply: 'Save and apply',
resetDefaults: 'Reset defaults',
latestWriteError: 'Latest write error:',
frontend/src/i18n/locales/zh/admin/ops.ts
+3
View File
@@ -42,8 +42,11 @@ export default {
samplingInitial: '采样初始条数',
samplingThereafter: '后续采样间隔',
retentionDays: '保留天数',
retentionDaysHint: '由定时数据清理任务执行。',
caller: '调用方',
sampling: '采样',
persistAccessLogs: '将访问日志写入数据库',
persistAccessLogsHint: '默认关闭，因为访问日志会为每个请求新增一条带索引的数据库记录。警告、错误和审计日志始终会保留。',
saveAndApply: '保存并应用',
resetDefaults: '重置默认值',
latestWriteError: '最近写入错误：',
frontend/src/views/admin/ops/components/OpsSystemLogTable.vue
+10
View File
@@ -41,6 +41,7 @@ const runtimeLoading = ref(false)
const runtimeSaving = ref(false)
const runtimeConfig = reactive<OpsRuntimeLogConfig>({
level: 'info',
persist_access_logs: false,
enable_sampling: false,
sampling_initial: 100,
sampling_thereafter: 100,
@@ -231,6 +232,7 @@ const loadRuntimeConfig = async () => {
try {
const cfg = await opsAPI.getRuntimeLogConfig()
runtimeConfig.level = cfg.level
runtimeConfig.persist_access_logs = cfg.persist_access_logs
runtimeConfig.enable_sampling = cfg.enable_sampling
runtimeConfig.sampling_initial = cfg.sampling_initial
runtimeConfig.sampling_thereafter = cfg.sampling_thereafter
@@ -249,6 +251,7 @@ const saveRuntimeConfig = async () => {
try {
const saved = await opsAPI.updateRuntimeLogConfig({ ...runtimeConfig })
runtimeConfig.level = saved.level
runtimeConfig.persist_access_logs = saved.persist_access_logs
runtimeConfig.enable_sampling = saved.enable_sampling
runtimeConfig.sampling_initial = saved.sampling_initial
runtimeConfig.sampling_thereafter = saved.sampling_thereafter
@@ -272,6 +275,7 @@ const resetRuntimeConfig = async () => {
try {
const saved = await opsAPI.resetRuntimeLogConfig()
runtimeConfig.level = saved.level
runtimeConfig.persist_access_logs = saved.persist_access_logs
runtimeConfig.enable_sampling = saved.enable_sampling
runtimeConfig.sampling_initial = saved.sampling_initial
runtimeConfig.sampling_thereafter = saved.sampling_thereafter
@@ -419,6 +423,7 @@ onMounted(async () => {
<label class="text-xs text-gray-600 dark:text-gray-300">
{{ t('admin.ops.systemLogs.retentionDays') }}
<input v-model.number="runtimeConfig.retention_days" type="number" min="1" max="3650" class="input mt-1" />
<span class="mt-1 block text-[11px] text-gray-500 dark:text-gray-400">{{ t('admin.ops.systemLogs.retentionDaysHint') }}</span>
</label>
<div class="md:col-span-2 xl:col-span-6">
<div class="grid gap-3 lg:grid-cols-[minmax(0,1fr)_auto] lg:items-end">
@@ -431,6 +436,10 @@ onMounted(async () => {
<input v-model="runtimeConfig.enable_sampling" type="checkbox" />
{{ t('admin.ops.systemLogs.sampling') }}
</label>
<label class="inline-flex items-center gap-2 text-xs text-gray-600 dark:text-gray-300">
<input v-model="runtimeConfig.persist_access_logs" type="checkbox" />
{{ t('admin.ops.systemLogs.persistAccessLogs') }}
</label>
</div>
<div class="flex flex-wrap items-center gap-2 lg:justify-end">
<button type="button" class="btn btn-primary btn-sm" :disabled="runtimeSaving" @click="saveRuntimeConfig">
@@ -443,6 +452,7 @@ onMounted(async () => {
</div>
</div>
</div>
<p class="mt-2 text-xs text-gray-500 dark:text-gray-400">{{ t('admin.ops.systemLogs.persistAccessLogsHint') }}</p>
<p v-if="health.last_error" class="mt-2 text-xs text-red-600 dark:text-red-400">{{ t('admin.ops.systemLogs.latestWriteError') }} {{ health.last_error }}</p>
</div>
frontend/src/views/admin/ops/components/__tests__/OpsSystemLogTable.spec.ts
+17
View File
@@ -53,6 +53,7 @@ const PaginationStub = defineComponent({
const runtimeConfig = {
level: 'info',
persist_access_logs: false,
enable_sampling: false,
sampling_initial: 100,
sampling_thereafter: 100,
@@ -126,6 +127,22 @@ describe('OpsSystemLogTable host support', () => {
expect(mockCleanupSystemLogs).toHaveBeenCalledWith(expect.objectContaining({ host: 'api-node-2' }))
})
it('keeps database access-log persistence opt-in', async () => {
const wrapper = mount(OpsSystemLogTable, {
global: {
stubs: {
Select: SelectStub,
Pagination: PaginationStub,
},
},
})
await flushPromises()
const label = wrapper.findAll('label').find((item) => item.text().includes('admin.ops.systemLogs.persistAccessLogs'))
expect(label).toBeDefined()
expect((label!.find('input').element as HTMLInputElement).checked).toBe(false)
})
it.each([
['zh', zhLocale],
['en', enLocale],