feat: add configurable rolling log retention

This commit is contained in:
haruka
2026-09-15 14:21:15 +08:00
parent 32682a4f84
commit 6582260bc8
17 changed files with 479 additions and 49 deletions
+1 -1
View File
@@ -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)
@@ -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`).
@@ -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 {
@@ -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)
}
}
@@ -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)
@@ -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,
+26 -8
View File
@@ -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())
@@ -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 {
@@ -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)
}
+2 -1
View File
@@ -373,8 +373,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