diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index 09a213e952..d851509d92 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -185,7 +185,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { dashboardAggregationRepository := repository.NewDashboardAggregationRepository(db) dashboardStatsCache := repository.NewDashboardCache(redisClient, configConfig) dashboardService := service.NewDashboardService(usageLogRepository, dashboardAggregationRepository, dashboardStatsCache, configConfig) - dashboardAggregationService := service.ProvideDashboardAggregationService(dashboardAggregationRepository, timingWheelService, leaderLockCache, db, configConfig) + dashboardAggregationService := service.ProvideDashboardAggregationService(dashboardAggregationRepository, timingWheelService, leaderLockCache, db, configConfig, settingRepository) dashboardHandler := admin.NewDashboardHandler(dashboardService, dashboardAggregationService) adminGroupRepository := repository.NewAdminGroupRepository(client, db) adminAccountRepository := repository.NewAdminAccountRepository(client, db, schedulerCache) diff --git a/backend/internal/repository/dashboard_aggregation_group_usage_test.go b/backend/internal/repository/dashboard_aggregation_group_usage_test.go index cbb5d3c6b6..40bca32bc4 100644 --- a/backend/internal/repository/dashboard_aggregation_group_usage_test.go +++ b/backend/internal/repository/dashboard_aggregation_group_usage_test.go @@ -249,6 +249,18 @@ func TestDashboardAggregationRepositoryCleanupUsageLogsPartitionedSortsAndInvali WillReturnResult(sqlmock.NewResult(0, 0)) mock.ExpectCommit() } + // The boundary partition is pruned by exact timestamp. tableoid is required + // because ctid alone can also identify a retained row in another partition. + mock.ExpectBegin() + mock.ExpectQuery(`SELECT id FROM usage_group_rollup_state.*FOR UPDATE`). + WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(1)) + mock.ExpectQuery(`(?s)SELECT tableoid, ctid.*WHERE created_at < \$1.*WHERE \(tableoid, ctid\) IN.*RETURNING created_at`). + WithArgs(cutoff, usageLogsCleanupBatchSize). + WillReturnRows(sqlmock.NewRows([]string{"created_at"}).AddRow(cutoff.Add(-time.Hour))) + mock.ExpectExec(`UPDATE usage_group_rollup_state`). + WithArgs(cutoff.Add(-time.Hour), "Asia/Shanghai"). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectCommit() mock.ExpectBegin() mock.ExpectQuery(`SELECT closed_before::text, retained_from.*FOR UPDATE`). WillReturnRows(sqlmock.NewRows([]string{"closed_before", "retained_from", "timezone_name"}). @@ -271,7 +283,7 @@ func TestDashboardAggregationRepositoryCleanupUsageLogsNonPartitionedFailureRoll mock.ExpectBegin() mock.ExpectQuery(`SELECT id FROM usage_group_rollup_state.*FOR UPDATE`). WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(1)) - mock.ExpectQuery(`(?s)SELECT ctid.*ORDER BY created_at ASC, id ASC.*DELETE FROM usage_logs.*RETURNING created_at`). + mock.ExpectQuery(`(?s)SELECT tableoid, ctid.*ORDER BY created_at ASC, id ASC.*DELETE FROM usage_logs.*RETURNING created_at`). WithArgs(cutoff, usageLogsCleanupBatchSize). WillReturnRows(sqlmock.NewRows([]string{"created_at"}).AddRow(deletedAt)) mock.ExpectExec(`UPDATE usage_group_rollup_state`). diff --git a/backend/internal/repository/dashboard_aggregation_repo.go b/backend/internal/repository/dashboard_aggregation_repo.go index 6701a3c999..347600df05 100644 --- a/backend/internal/repository/dashboard_aggregation_repo.go +++ b/backend/internal/repository/dashboard_aggregation_repo.go @@ -239,7 +239,10 @@ func (r *dashboardAggregationRepository) CleanupUsageLogs(ctx context.Context, c if err := r.dropUsageLogsPartitions(ctx, cutoff); err != nil { return err } - } else if err := r.cleanupUsageLogsBatches(ctx, cutoff); err != nil { + } + // Remove the expired part of the boundary partition as well, so a retention + // window measured in days does not silently round up to a whole month. + if err := r.cleanupUsageLogsBatches(ctx, cutoff); err != nil { return err } return r.SyncGroupUsageRollups(ctx, service.GroupUsageTodayStart(r.now())) @@ -261,14 +264,14 @@ func (r *dashboardAggregationRepository) cleanupUsageLogsBatches(ctx context.Con res, err := r.sql.ExecContext(ctx, ` WITH victims AS ( - SELECT ctid + SELECT tableoid, ctid FROM usage_logs WHERE created_at < $1 ORDER BY created_at ASC, id ASC LIMIT $2 ) DELETE FROM usage_logs - WHERE ctid IN (SELECT ctid FROM victims) + WHERE (tableoid, ctid) IN (SELECT tableoid, ctid FROM victims) `, cutoff.UTC(), usageLogsCleanupBatchSize) if err != nil { return err @@ -298,14 +301,14 @@ func cleanupUsageLogsBatchWithRollupInvalidation(ctx context.Context, db *sql.DB } rows, err := tx.QueryContext(ctx, ` WITH victims AS ( - SELECT ctid + SELECT tableoid, ctid FROM usage_logs WHERE created_at < $1 ORDER BY created_at ASC, id ASC LIMIT $2 ) DELETE FROM usage_logs - WHERE ctid IN (SELECT ctid FROM victims) + WHERE (tableoid, ctid) IN (SELECT tableoid, ctid FROM victims) RETURNING created_at `, cutoff.UTC(), usageLogsCleanupBatchSize) if err != nil { diff --git a/backend/internal/repository/request_log_retention_integration_test.go b/backend/internal/repository/request_log_retention_integration_test.go new file mode 100644 index 0000000000..faee600804 --- /dev/null +++ b/backend/internal/repository/request_log_retention_integration_test.go @@ -0,0 +1,46 @@ +//go:build integration + +package repository + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestRequestLogRetention_PartitionBoundaryKeepsRecentRows(t *testing.T) { + ctx := context.Background() + tx := testTx(t) + // Temporary tables shadow the real schema and disappear on rollback. + _, err := tx.ExecContext(ctx, ` + CREATE TEMP TABLE usage_logs (id bigint, created_at timestamptz) PARTITION BY RANGE (created_at); + CREATE TEMP TABLE retention_july PARTITION OF usage_logs FOR VALUES FROM ('2026-07-01') TO ('2026-08-01'); + CREATE TEMP TABLE retention_august PARTITION OF usage_logs FOR VALUES FROM ('2026-08-01') TO ('2026-09-01'); + INSERT INTO usage_logs VALUES + (1, '2026-07-17 23:59:59+00'), + (2, '2026-07-18 00:00:00+00'), + (3, '2026-08-01 00:00:00+00'); + `) + require.NoError(t, err) + var collidingCTIDs bool + require.NoError(t, tx.QueryRowContext(ctx, `SELECT a.ctid = b.ctid FROM usage_logs a, usage_logs b WHERE a.id = 1 AND b.id = 3`).Scan(&collidingCTIDs)) + require.True(t, collidingCTIDs, "fixture must exercise identical row locations in different partitions") + repo := newDashboardAggregationRepositoryWithSQL(tx) + cutoff := time.Date(2026, 7, 18, 0, 0, 0, 0, time.UTC) + for range 2 { + require.NoError(t, repo.cleanupUsageLogsBatches(ctx, cutoff)) + rows, err := tx.QueryContext(ctx, `SELECT id FROM usage_logs ORDER BY id`) + require.NoError(t, err) + var ids []int64 + for rows.Next() { + var id int64 + require.NoError(t, rows.Scan(&id)) + ids = append(ids, id) + } + require.NoError(t, rows.Err()) + require.NoError(t, rows.Close()) + require.Equal(t, []int64{2, 3}, ids) + } +} diff --git a/backend/internal/service/dashboard_aggregation_service.go b/backend/internal/service/dashboard_aggregation_service.go index 6e351d75d2..c85dec07a3 100644 --- a/backend/internal/service/dashboard_aggregation_service.go +++ b/backend/internal/service/dashboard_aggregation_service.go @@ -3,7 +3,9 @@ package service import ( "context" "database/sql" + "encoding/json" "errors" + "fmt" "log/slog" "sync/atomic" "time" @@ -55,6 +57,7 @@ type DashboardAggregationService struct { repo DashboardAggregationRepository timingWheel *TimingWheelService cfg config.DashboardAggregationConfig + settingRepo SettingRepository running int32 lastRetentionCleanup atomic.Value // time.Time @@ -95,6 +98,8 @@ func (s *DashboardAggregationService) Start() { } if !s.cfg.Enabled { logger.LegacyPrintf("service.dashboard_aggregation", "[DashboardAggregation] 聚合作业已禁用") + // Explicit runtime retention remains available without dashboard aggregation. + s.timingWheel.ScheduleRecurring("dashboard:retention", time.Minute, s.runScheduledRetention) return } go s.runStartupGroupUsageSync() @@ -360,20 +365,38 @@ func (s *DashboardAggregationService) maybeCleanupRetention(ctx context.Context, } } + usageDays, err := s.requestRetentionDays(ctx) + if err != nil { + // Never fall back to a shorter window after a settings read failure. + logger.LegacyPrintf("service.dashboard_aggregation", "[DashboardAggregation] 读取日志保留设置失败，跳过清理: %v", err) + return + } hourlyCutoff := now.AddDate(0, 0, -s.cfg.Retention.HourlyDays) dailyCutoff := now.AddDate(0, 0, -s.cfg.Retention.DailyDays) - usageCutoff := now.AddDate(0, 0, -s.cfg.Retention.UsageLogsDays) - dedupCutoff := now.AddDate(0, 0, -s.cfg.Retention.UsageBillingDedupDays) + usageCutoff := now.AddDate(0, 0, -usageDays) + dedupDays := s.cfg.Retention.UsageBillingDedupDays + if dedupDays <= 0 { + dedupDays = 365 + } + dedupDays = max(dedupDays, usageDays) + dedupCutoff := now.AddDate(0, 0, -dedupDays) - aggErr := s.repo.CleanupAggregates(ctx, hourlyCutoff, dailyCutoff) + var aggErr, usageErr, dedupErr error + if s.cfg.Enabled { + aggErr = s.repo.CleanupAggregates(ctx, hourlyCutoff, dailyCutoff) + } if aggErr != nil { logger.LegacyPrintf("service.dashboard_aggregation", "[DashboardAggregation] 聚合保留清理失败: %v", aggErr) } - usageErr := s.repo.CleanupUsageLogs(ctx, usageCutoff) + if usageDays > 0 { + usageErr = s.repo.CleanupUsageLogs(ctx, usageCutoff) + } if usageErr != nil { logger.LegacyPrintf("service.dashboard_aggregation", "[DashboardAggregation] usage_logs 保留清理失败: %v", usageErr) } - dedupErr := s.repo.CleanupUsageBillingDedup(ctx, dedupCutoff) + if usageDays > 0 { + dedupErr = s.repo.CleanupUsageBillingDedup(ctx, dedupCutoff) + } if dedupErr != nil { logger.LegacyPrintf("service.dashboard_aggregation", "[DashboardAggregation] usage_billing_dedup 保留清理失败: %v", dedupErr) } @@ -382,6 +405,51 @@ func (s *DashboardAggregationService) maybeCleanupRetention(ctx context.Context, } } +func (s *DashboardAggregationService) requestRetentionDays(ctx context.Context) (int, error) { + days := s.cfg.Retention.UsageLogsDays + if !s.cfg.Enabled { + days = 0 + } + if s.settingRepo == nil { + return days, nil + } + raw, err := s.settingRepo.GetValue(ctx, SettingKeyOpsRuntimeLogConfig) + if errors.Is(err, ErrSettingNotFound) { + return days, nil + } + if err != nil { + return 0, err + } + var cfg struct { + RequestRetentionDays *int `json:"request_retention_days"` + } + if err := json.Unmarshal([]byte(raw), &cfg); err != nil { + return 0, err + } + if cfg.RequestRetentionDays != nil { + days = *cfg.RequestRetentionDays + if days < 0 || days > 3650 { + return 0, fmt.Errorf("invalid request_retention_days: %d", days) + } + } + return days, nil +} + +func (s *DashboardAggregationService) runScheduledRetention() { + if !atomic.CompareAndSwapInt32(&s.running, 0, 1) { + return + } + defer atomic.StoreInt32(&s.running, 0) + ctx, cancel := context.WithTimeout(context.Background(), defaultDashboardAggregationTimeout) + defer cancel() + release, ok := tryAcquireSingletonLeaderLock(ctx, s.lockCache, s.db, dashboardAggregationLeaderLockKey, s.instanceID, dashboardAggregationLeaderLockTTL) + if !ok { + return + } + defer release() + s.maybeCleanupRetention(ctx, time.Now().UTC()) +} + func truncateToDayUTC(t time.Time) time.Time { t = t.UTC() return time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, time.UTC) diff --git a/backend/internal/service/dashboard_aggregation_service_test.go b/backend/internal/service/dashboard_aggregation_service_test.go index d164519330..f03d48491e 100644 --- a/backend/internal/service/dashboard_aggregation_service_test.go +++ b/backend/internal/service/dashboard_aggregation_service_test.go @@ -11,21 +11,24 @@ import ( ) type dashboardAggregationRepoTestStub struct { - aggregateCalls int - recomputeCalls int - cleanupUsageCalls int - cleanupDedupCalls int - ensurePartitionCalls int - lastStart time.Time - lastEnd time.Time - watermark time.Time - aggregateErr error - cleanupAggregatesErr error - cleanupUsageErr error - cleanupDedupErr error - ensurePartitionErr error - aggregateCtx context.Context - events *[]string + aggregateCalls int + recomputeCalls int + cleanupUsageCalls int + cleanupUsageCutoff time.Time + cleanupDedupCutoff time.Time + cleanupAggregateCalls int + cleanupDedupCalls int + ensurePartitionCalls int + lastStart time.Time + lastEnd time.Time + watermark time.Time + aggregateErr error + cleanupAggregatesErr error + cleanupUsageErr error + cleanupDedupErr error + ensurePartitionErr error + aggregateCtx context.Context + events *[]string } type dashboardAggregationRollupRepoTestStub struct { @@ -71,16 +74,19 @@ func (s *dashboardAggregationRepoTestStub) UpdateAggregationWatermark(ctx contex } func (s *dashboardAggregationRepoTestStub) CleanupAggregates(ctx context.Context, hourlyCutoff, dailyCutoff time.Time) error { + s.cleanupAggregateCalls++ return s.cleanupAggregatesErr } func (s *dashboardAggregationRepoTestStub) CleanupUsageLogs(ctx context.Context, cutoff time.Time) error { s.cleanupUsageCalls++ + s.cleanupUsageCutoff = cutoff return s.cleanupUsageErr } func (s *dashboardAggregationRepoTestStub) CleanupUsageBillingDedup(ctx context.Context, cutoff time.Time) error { s.cleanupDedupCalls++ + s.cleanupDedupCutoff = cutoff return s.cleanupDedupErr } @@ -214,6 +220,7 @@ func TestDashboardAggregationService_CleanupRetentionFailure_DoesNotRecord(t *te svc := &DashboardAggregationService{ repo: repo, cfg: config.DashboardAggregationConfig{ + Enabled: true, Retention: config.DashboardAggregationRetentionConfig{ UsageLogsDays: 1, HourlyDays: 1, @@ -234,6 +241,7 @@ func TestDashboardAggregationService_CleanupDedupFailure_DoesNotRecord(t *testin svc := &DashboardAggregationService{ repo: repo, cfg: config.DashboardAggregationConfig{ + Enabled: true, Retention: config.DashboardAggregationRetentionConfig{ UsageLogsDays: 1, HourlyDays: 1, diff --git a/backend/internal/service/ops_log_runtime.go b/backend/internal/service/ops_log_runtime.go index 637ab8d69f..481fa6ca6c 100644 --- a/backend/internal/service/ops_log_runtime.go +++ b/backend/internal/service/ops_log_runtime.go @@ -14,15 +14,23 @@ import ( ) func defaultOpsRuntimeLogConfig(cfg *config.Config) *OpsRuntimeLogConfig { + requestRetentionDays := 90 + if cfg != nil { + requestRetentionDays = cfg.DashboardAgg.Retention.UsageLogsDays + if !cfg.DashboardAgg.Enabled { + requestRetentionDays = 0 + } + } out := &OpsRuntimeLogConfig{ - Level: "info", - PersistAccessLogs: false, - 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, + RequestRetentionDays: &requestRetentionDays, } if cfg == nil { return out @@ -60,6 +68,10 @@ func normalizeOpsRuntimeLogConfig(cfg *OpsRuntimeLogConfig, defaults *OpsRuntime if cfg.RetentionDays <= 0 { cfg.RetentionDays = defaults.RetentionDays } + if cfg.RequestRetentionDays == nil && defaults.RequestRetentionDays != nil { + days := *defaults.RequestRetentionDays + cfg.RequestRetentionDays = &days + } } func validateOpsRuntimeLogConfig(cfg *OpsRuntimeLogConfig) error { @@ -85,6 +97,9 @@ func validateOpsRuntimeLogConfig(cfg *OpsRuntimeLogConfig) error { if cfg.RetentionDays < 1 || cfg.RetentionDays > 3650 { return errors.New("retention_days must be between 1 and 3650") } + if cfg.RequestRetentionDays != nil && (*cfg.RequestRetentionDays < 0 || *cfg.RequestRetentionDays > 3650) { + return errors.New("request_retention_days must be between 0 and 3650") + } return nil } @@ -139,6 +154,9 @@ func (s *OpsService) UpdateRuntimeLogConfig(ctx context.Context, req *OpsRuntime return nil, err } next := *req + if next.RequestRetentionDays == nil { + next.RequestRetentionDays = oldCfg.RequestRetentionDays + } normalizeOpsRuntimeLogConfig(&next, defaultOpsRuntimeLogConfig(s.cfg)) if err := validateOpsRuntimeLogConfig(&next); err != nil { s.auditRuntimeLogConfigFailure(operatorID, oldCfg, &next, "validation_failed: "+err.Error()) diff --git a/backend/internal/service/ops_settings_models.go b/backend/internal/service/ops_settings_models.go index 93c364153c..8a32d7a92b 100644 --- a/backend/internal/service/ops_settings_models.go +++ b/backend/internal/service/ops_settings_models.go @@ -77,11 +77,14 @@ type OpsRuntimeLogConfig struct { 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"` - UpdatedByUserID int64 `json:"updated_by_user_id,omitempty"` - Extra map[string]any `json:"extra,omitempty"` + RetentionDays int `json:"retention_days"` + // RequestRetentionDays controls usage_logs; 0 keeps request history indefinitely. + // A nil value from older clients preserves the existing setting. + RequestRetentionDays *int `json:"request_retention_days"` + Source string `json:"source,omitempty"` + UpdatedAt string `json:"updated_at,omitempty"` + UpdatedByUserID int64 `json:"updated_by_user_id,omitempty"` + Extra map[string]any `json:"extra,omitempty"` } type OpsAlertRuntimeSettings struct { diff --git a/backend/internal/service/request_log_retention_test.go b/backend/internal/service/request_log_retention_test.go new file mode 100644 index 0000000000..4cc8955f79 --- /dev/null +++ b/backend/internal/service/request_log_retention_test.go @@ -0,0 +1,109 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/pkg/logger" + "github.com/stretchr/testify/require" +) + +func TestRequestLogRetention_RuntimePolicy(t *testing.T) { + for _, tt := range []struct { + name string + raw string + enabled bool + wantDays int + wantCleanup bool + }{ + {"legacy uses deployment", `{"retention_days":30}`, true, 90, true}, + {"short window", `{"request_retention_days":7}`, true, 7, true}, + {"long window extends dedup", `{"request_retention_days":730}`, true, 730, true}, + {"forever skips request and dedup deletion", `{"request_retention_days":0}`, true, 0, false}, + {"works without aggregation", `{"request_retention_days":14}`, false, 14, true}, + {"legacy disabled stays disabled", `{"retention_days":30}`, false, 0, false}, + } { + t.Run(tt.name, func(t *testing.T) { + settings := newRuntimeSettingRepoStub() + settings.values[SettingKeyOpsRuntimeLogConfig] = tt.raw + repo := &dashboardAggregationRepoTestStub{} + svc := NewDashboardAggregationService(repo, nil, &config.Config{DashboardAgg: config.DashboardAggregationConfig{ + Enabled: tt.enabled, + Retention: config.DashboardAggregationRetentionConfig{UsageLogsDays: 90, UsageBillingDedupDays: 365, HourlyDays: 180, DailyDays: 730}, + }}) + svc.settingRepo = settings + now := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC) + svc.maybeCleanupRetention(context.Background(), now) + if tt.wantCleanup { + require.Equal(t, 1, repo.cleanupUsageCalls) + require.Equal(t, now.AddDate(0, 0, -tt.wantDays), repo.cleanupUsageCutoff) + require.Equal(t, now.AddDate(0, 0, -max(365, tt.wantDays)), repo.cleanupDedupCutoff) + } else { + require.Zero(t, repo.cleanupUsageCalls) + require.Zero(t, repo.cleanupDedupCalls) + } + if !tt.enabled { + require.Zero(t, repo.cleanupAggregateCalls) + } + // Read the updated policy at the next rolling cleanup. + settings.values[SettingKeyOpsRuntimeLogConfig] = `{"request_retention_days":180}` + next := now.Add(dashboardAggregationRetentionInterval) + svc.maybeCleanupRetention(context.Background(), next) + require.Equal(t, next.AddDate(0, 0, -180), repo.cleanupUsageCutoff) + }) + } +} + +func TestRequestLogRetention_ReadFailureSkipsDeletion(t *testing.T) { + for _, raw := range []string{`{broken`, `{"request_retention_days":-1}`, `{"request_retention_days":3651}`} { + settings := newRuntimeSettingRepoStub() + settings.values[SettingKeyOpsRuntimeLogConfig] = raw + repo := &dashboardAggregationRepoTestStub{} + svc := &DashboardAggregationService{repo: repo, settingRepo: settings} + svc.runScheduledRetention() + require.Zero(t, repo.cleanupUsageCalls) + require.Zero(t, repo.cleanupDedupCalls) + require.Nil(t, svc.lastRetentionCleanup.Load()) + } + settings := newRuntimeSettingRepoStub() + settings.getValueFn = func(string) (string, error) { return "", errors.New("database unavailable") } + svc := &DashboardAggregationService{repo: &dashboardAggregationRepoTestStub{}, settingRepo: settings} + svc.runScheduledRetention() + require.Nil(t, svc.lastRetentionCleanup.Load()) +} + +func TestRuntimeLogConfig_RequestRetentionCompatibility(t *testing.T) { + settings := newRuntimeSettingRepoStub() + settings.values[SettingKeyOpsRuntimeLogConfig] = `{"level":"info","retention_days":7,"request_retention_days":180}` + svc := &OpsService{settingRepo: settings} + require.NoError(t, logger.Init(logger.InitOptions{Level: "info", Format: "json", Output: logger.OutputOptions{ToStdout: true}})) + t.Cleanup(logger.Sync) + legacy := defaultOpsRuntimeLogConfig(nil) + legacy.RequestRetentionDays = nil + updated, err := svc.UpdateRuntimeLogConfig(context.Background(), legacy, 1) + require.NoError(t, err) + require.Equal(t, 180, *updated.RequestRetentionDays) + for _, days := range []int{0, 1, 3650} { + updated.RequestRetentionDays = &days + saved, err := svc.UpdateRuntimeLogConfig(context.Background(), updated, 1) + require.NoError(t, err) + require.Equal(t, days, *saved.RequestRetentionDays) + var persisted OpsRuntimeLogConfig + require.NoError(t, json.Unmarshal([]byte(settings.values[SettingKeyOpsRuntimeLogConfig]), &persisted)) + require.Equal(t, days, *persisted.RequestRetentionDays) + } + for _, days := range []int{-1, 3651} { + updated.RequestRetentionDays = &days + before := settings.values[SettingKeyOpsRuntimeLogConfig] + _, err := svc.UpdateRuntimeLogConfig(context.Background(), updated, 1) + require.ErrorContains(t, err, "request_retention_days") + require.Equal(t, before, settings.values[SettingKeyOpsRuntimeLogConfig]) + } + reset, err := svc.ResetRuntimeLogConfig(context.Background(), 1) + require.NoError(t, err) + require.Equal(t, 90, *reset.RequestRetentionDays) +} diff --git a/backend/internal/service/wire.go b/backend/internal/service/wire.go index 2af85ff6d8..e5aa732098 100644 --- a/backend/internal/service/wire.go +++ b/backend/internal/service/wire.go @@ -381,8 +381,9 @@ func ProvideGrokTokenProvider( } // ProvideDashboardAggregationService 创建并启动仪表盘聚合服务 -func ProvideDashboardAggregationService(repo DashboardAggregationRepository, timingWheel *TimingWheelService, lockCache LeaderLockCache, db *sql.DB, cfg *config.Config) *DashboardAggregationService { +func ProvideDashboardAggregationService(repo DashboardAggregationRepository, timingWheel *TimingWheelService, lockCache LeaderLockCache, db *sql.DB, cfg *config.Config, settingRepo SettingRepository) *DashboardAggregationService { svc := NewDashboardAggregationService(repo, timingWheel, cfg) + svc.settingRepo = settingRepo svc.SetLeaderLock(lockCache, db) svc.Start() return svc diff --git a/docs/log-retention.md b/docs/log-retention.md new file mode 100644 index 0000000000..64c90f68e8 --- /dev/null +++ b/docs/log-retention.md @@ -0,0 +1,21 @@ +# 滚动日志保留设置 + +管理员在 **运维监控 → 系统日志 → 运行时日志配置** 中设置保留期，然后点击「保存并应用」。 + +| 日志 | 可选范围 | 执行时间 | +| --- | --- | --- | +| 运维系统日志 | 1–3650 天 | 运维设置中的数据清理计划，须开启「启用数据清理」 | +| 请求日志（请求使用明细） | 1–3650 天，或永久保留 | 每 6 小时清理一次，保存后在下一次清理生效 | + +下拉框提供 1、3、7、14、30、60、90、180、365 天，也可以选择「自定义天数」。按天清理只删除早于截止时间的记录；截止时间上的记录保留。请求日志按月分区时，也会清理截止日期所在分区中的过期记录。 + +## 配置兼容 + +- 设置保存在 `ops_runtime_log_config`：`retention_days` 控制运维系统日志，`request_retention_days` 控制请求日志，后者为 `0` 时表示永久保留。 +- 没有请求日志设置时，沿用 `dashboard_aggregation.retention.usage_logs_days`（默认 90 天）；原来禁用仪表盘聚合的部署继续保留请求日志，不自动开启删除。 +- 显式保存请求日志保留天数后，即使仪表盘聚合关闭，滚动清理仍会运行。 +- 旧客户端没有传入 `request_retention_days` 时，保留已有值。「重置默认值」恢复部署配置。 +- 计费去重记录的保留期至少与请求日志一样长；请求日志永久保留时，也停止清理计费去重记录。 +- 读取请求日志保留设置失败或设置无效时，跳过清理，避免回退到更短的保留期。 + +运维错误日志和指标继续使用运维设置中的各自保留策略。本页的运维保留期针对数据库中的系统日志；磁盘日志文件仍使用部署配置中的轮转策略。 diff --git a/docs/screenshots/log-retention.png b/docs/screenshots/log-retention.png new file mode 100644 index 0000000000..fe45f97f2a Binary files /dev/null and b/docs/screenshots/log-retention.png differ diff --git a/frontend/src/api/admin/ops.ts b/frontend/src/api/admin/ops.ts index a5ca947e11..8754f19583 100644 --- a/frontend/src/api/admin/ops.ts +++ b/frontend/src/api/admin/ops.ts @@ -822,6 +822,7 @@ export interface OpsRuntimeLogConfig { caller: boolean stacktrace_level: 'none' | 'error' | 'fatal' retention_days: number + request_retention_days: number source?: string updated_at?: string updated_by_user_id?: number diff --git a/frontend/src/i18n/locales/en/admin/ops.ts b/frontend/src/i18n/locales/en/admin/ops.ts index afdbd36db6..8576736b9c 100644 --- a/frontend/src/i18n/locales/en/admin/ops.ts +++ b/frontend/src/i18n/locales/en/admin/ops.ts @@ -41,8 +41,14 @@ export default { stacktraceThreshold: 'Stacktrace threshold', samplingInitial: 'Sampling initial', samplingThereafter: 'Sampling thereafter', - retentionDays: 'Retention days', - retentionDaysHint: 'Applied by the scheduled data-cleanup job.', + retentionDays: 'Operations log retention days', + requestRetentionDays: 'Request log retention days', + requestRetentionDaysHint: 'Request usage records are pruned every 6 hours. Changes apply on the next cleanup. Keeping records forever uses increasing storage.', + retentionDaysInvalid: 'Keep operations logs for 1–3650 days; keep request logs for 1–3650 days or choose Forever.', + retentionDaysOption: '{days} days', + retentionDaysCustom: 'Custom days', + retentionForever: 'Forever', + retentionDaysHint: 'Applied on the data-cleanup schedule when cleanup is enabled in Operations Settings.', caller: 'caller', sampling: 'sampling', persistAccessLogs: 'Store access logs in database', @@ -69,6 +75,7 @@ export default { logDetails: 'Log Details', loadFailed: 'Failed to load system logs', runtimeConfigActive: 'Runtime log configuration is active', + runtimeConfigLoadFailed: 'Failed to load log configuration. Refresh and try again.', runtimeConfigSaveFailed: 'Failed to save log configuration', resetRuntimeConfigConfirm: 'Reset to startup configuration (env/yaml) and apply immediately?', runtimeConfigReset: 'Reset to startup log configuration', diff --git a/frontend/src/i18n/locales/zh/admin/ops.ts b/frontend/src/i18n/locales/zh/admin/ops.ts index c7c017e4b9..ace6fff1e4 100644 --- a/frontend/src/i18n/locales/zh/admin/ops.ts +++ b/frontend/src/i18n/locales/zh/admin/ops.ts @@ -41,8 +41,14 @@ export default { stacktraceThreshold: '堆栈阈值', samplingInitial: '采样初始条数', samplingThereafter: '后续采样间隔', - retentionDays: '保留天数', - retentionDaysHint: '由定时数据清理任务执行。', + retentionDays: '运维日志保留天数', + requestRetentionDays: '请求日志保留天数', + requestRetentionDaysHint: '请求使用明细每 6 小时滚动清理一次，保存后在下次清理生效。永久保留会持续占用存储空间。', + retentionDaysInvalid: '运维日志须保留 1–3650 天；请求日志须保留 1–3650 天或选择永久保留。', + retentionDaysOption: '{days} 天', + retentionDaysCustom: '自定义天数', + retentionForever: '永久保留', + retentionDaysHint: '按运维设置中启用的数据清理计划执行。', caller: '调用方', sampling: '采样', persistAccessLogs: '将访问日志写入数据库', @@ -69,6 +75,7 @@ export default { logDetails: '日志详情', loadFailed: '加载系统日志失败', runtimeConfigActive: '运行时日志配置已生效', + runtimeConfigLoadFailed: '日志配置加载失败，请刷新后重试。', runtimeConfigSaveFailed: '保存日志配置失败', resetRuntimeConfigConfirm: '确定要重置为启动配置（env/yaml）并立即应用吗？', runtimeConfigReset: '已重置为启动日志配置', diff --git a/frontend/src/views/admin/ops/components/LogRetentionSelect.vue b/frontend/src/views/admin/ops/components/LogRetentionSelect.vue new file mode 100644 index 0000000000..21254ebc38 --- /dev/null +++ b/frontend/src/views/admin/ops/components/LogRetentionSelect.vue @@ -0,0 +1,43 @@ + + + diff --git a/frontend/src/views/admin/ops/components/OpsSystemLogTable.vue b/frontend/src/views/admin/ops/components/OpsSystemLogTable.vue index e360be3ce3..3c6c1c2bf6 100644 --- a/frontend/src/views/admin/ops/components/OpsSystemLogTable.vue +++ b/frontend/src/views/admin/ops/components/OpsSystemLogTable.vue @@ -2,6 +2,7 @@ import { computed, onMounted, reactive, ref, watch } from 'vue' import { useMediaQuery } from '@vueuse/core' import { useI18n } from 'vue-i18n' +import LogRetentionSelect from './LogRetentionSelect.vue' import { opsAPI, type OpsRuntimeLogConfig, type OpsSystemLog, type OpsSystemLogSinkHealth } from '@/api/admin/ops' import Pagination from '@/components/common/Pagination.vue' import Select from '@/components/common/Select.vue' @@ -38,6 +39,7 @@ const health = ref({ }) const runtimeLoading = ref(false) +const runtimeConfigLoaded = ref(false) const runtimeSaving = ref(false) const runtimeConfig = reactive({ level: 'info', @@ -47,7 +49,8 @@ const runtimeConfig = reactive({ sampling_thereafter: 100, caller: true, stacktrace_level: 'error', - retention_days: 30 + retention_days: 30, + request_retention_days: 90 }) const filters = reactive({ @@ -229,6 +232,7 @@ const fetchHealth = async () => { const loadRuntimeConfig = async () => { runtimeLoading.value = true + runtimeConfigLoaded.value = false try { const cfg = await opsAPI.getRuntimeLogConfig() runtimeConfig.level = cfg.level @@ -239,14 +243,23 @@ const loadRuntimeConfig = async () => { runtimeConfig.caller = cfg.caller runtimeConfig.stacktrace_level = cfg.stacktrace_level runtimeConfig.retention_days = cfg.retention_days + runtimeConfig.request_retention_days = cfg.request_retention_days ?? 90 + runtimeConfigLoaded.value = true } catch (err: any) { console.error('[OpsSystemLogTable] Failed to load runtime log config', err) + appStore.showError(t('admin.ops.systemLogs.runtimeConfigLoadFailed')) } finally { runtimeLoading.value = false } } const saveRuntimeConfig = async () => { + if (!runtimeConfigLoaded.value || runtimeLoading.value) return + if (!Number.isInteger(runtimeConfig.retention_days) || runtimeConfig.retention_days < 1 || runtimeConfig.retention_days > 3650 || + !Number.isInteger(runtimeConfig.request_retention_days) || runtimeConfig.request_retention_days < 0 || runtimeConfig.request_retention_days > 3650) { + appStore.showError(t('admin.ops.systemLogs.retentionDaysInvalid')) + return + } runtimeSaving.value = true try { const saved = await opsAPI.updateRuntimeLogConfig({ ...runtimeConfig }) @@ -258,6 +271,7 @@ const saveRuntimeConfig = async () => { runtimeConfig.caller = saved.caller runtimeConfig.stacktrace_level = saved.stacktrace_level runtimeConfig.retention_days = saved.retention_days + runtimeConfig.request_retention_days = saved.request_retention_days ?? 90 appStore.showSuccess(t('admin.ops.systemLogs.runtimeConfigActive')) } catch (err: any) { console.error('[OpsSystemLogTable] Failed to save runtime log config', err) @@ -282,6 +296,7 @@ const resetRuntimeConfig = async () => { runtimeConfig.caller = saved.caller runtimeConfig.stacktrace_level = saved.stacktrace_level runtimeConfig.retention_days = saved.retention_days + runtimeConfig.request_retention_days = saved.request_retention_days ?? 90 appStore.showSuccess(t('admin.ops.systemLogs.runtimeConfigReset')) await fetchHealth() } catch (err: any) { @@ -422,9 +437,14 @@ onMounted(async () => { +
@@ -442,7 +462,7 @@ onMounted(async () => {
-