From 22b95db2f9c26d95bf5777d4ef321149d163c6d1 Mon Sep 17 00:00:00 2001 From: break Date: Thu, 20 Aug 2026 12:03:17 +0800 Subject: [PATCH] =?UTF-8?q?=E6=AF=8F=E6=97=A5=E6=B8=85=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/audit-retention-simulate/main.go | 66 ++- cmd/worker/main.go | 40 +- docs/deployment/production-runbook.md | 9 + .../application/auditarchive/integration.go | 25 +- .../application/auditarchive/retention.go | 505 ++++++------------ internal/query/retention/retention.go | 102 ++-- internal/task/audit_monthly_retention.go | 96 +--- internal/task/integration_archive.go | 46 +- .../.openspec.yaml | 2 + .../design.md | 68 +++ .../proposal.md | 28 + .../specs/external-integration/spec.md | 19 + .../specs/operations-audit/spec.md | 41 ++ .../tasks.md | 19 + openspec/specs/external-integration/spec.md | 19 + openspec/specs/operations-audit/spec.md | 43 ++ pkg/bootstrap/directories.go | 10 +- pkg/config/config.go | 17 +- pkg/config/defaults/config.yaml | 8 +- pkg/config/loader.go | 5 + pkg/constants/audit_archive.go | 8 +- pkg/constants/constants.go | 2 +- pkg/logger/retention.go | 17 + 23 files changed, 614 insertions(+), 581 deletions(-) create mode 100644 openspec/changes/archive/2026-08-20-daily-audit-log-retention/.openspec.yaml create mode 100644 openspec/changes/archive/2026-08-20-daily-audit-log-retention/design.md create mode 100644 openspec/changes/archive/2026-08-20-daily-audit-log-retention/proposal.md create mode 100644 openspec/changes/archive/2026-08-20-daily-audit-log-retention/specs/external-integration/spec.md create mode 100644 openspec/changes/archive/2026-08-20-daily-audit-log-retention/specs/operations-audit/spec.md create mode 100644 openspec/changes/archive/2026-08-20-daily-audit-log-retention/tasks.md create mode 100644 pkg/logger/retention.go diff --git a/cmd/audit-retention-simulate/main.go b/cmd/audit-retention-simulate/main.go index b641f62..3923cea 100644 --- a/cmd/audit-retention-simulate/main.go +++ b/cmd/audit-retention-simulate/main.go @@ -1,4 +1,4 @@ -// Command audit-retention-simulate 在测试环境演练完整自然月归档与清理边界。 +// Command audit-retention-simulate 在测试环境演练逐日归档与清理边界。 package main import ( @@ -6,22 +6,21 @@ import ( "crypto/sha256" "fmt" "os" + "path/filepath" "strings" "time" "github.com/bytedance/sonic" - "github.com/hibiken/asynq" "go.uber.org/zap" "gorm.io/datatypes" "gorm.io/gorm" auditarchive "github.com/break/junhong_cmp_fiber/internal/application/auditarchive" - auditinfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/audit" "github.com/break/junhong_cmp_fiber/internal/model" - taskapp "github.com/break/junhong_cmp_fiber/internal/task" "github.com/break/junhong_cmp_fiber/pkg/config" "github.com/break/junhong_cmp_fiber/pkg/constants" "github.com/break/junhong_cmp_fiber/pkg/database" + logpkg "github.com/break/junhong_cmp_fiber/pkg/logger" "github.com/break/junhong_cmp_fiber/pkg/storage" ) @@ -67,6 +66,11 @@ func run(ctx context.Context) error { return fmt.Errorf("仅允许显式确认的测试数据库，当前数据库为 %q", cfg.Database.DBName) } logger := zap.NewNop() + if err := os.MkdirAll(filepath.Dir(cfg.Logging.RetentionLog.Filename), 0755); err != nil { + return err + } + retentionLogger, syncRetentionLogger := logpkg.NewRetentionLogger(cfg.Logging.Level, logpkg.LogRotationConfig{Filename: cfg.Logging.RetentionLog.Filename, MaxSize: cfg.Logging.RetentionLog.MaxSize, MaxBackups: cfg.Logging.RetentionLog.MaxBackups, MaxAge: cfg.Logging.RetentionLog.MaxAge, Compress: cfg.Logging.RetentionLog.Compress}) + defer func() { _ = syncRetentionLogger() }() db, err := database.InitPostgreSQL(&cfg.Database, logger) if err != nil { return err @@ -81,8 +85,7 @@ func run(ctx context.Context) error { if err != nil { return err } - auditWriter := auditinfra.NewWriter(auditinfra.NewRegistry(), nil) - service, err := auditarchive.NewService(db, provider, simulationInstance, auditWriter) + service, err := auditarchive.NewService(db, provider, simulationInstance) if err != nil { return err } @@ -108,14 +111,17 @@ func run(ctx context.Context) error { } } - payload, _ := sonic.Marshal(taskapp.AuditMonthlyRetentionPayload{ArchiveMonth: simulationMonth}) - dryRunTask := asynq.NewTask(constants.TaskTypeAuditMonthlyRetention, payload) - if err := taskapp.NewAuditMonthlyRetentionHandler(service, logger, false).Handle(ctx, dryRunTask); err != nil { - return fmt.Errorf("月度只读演练失败: %w", err) - } - dryRun, err := service.ValidateMonth(ctx, monthStart) - if err != nil { - return err + var dryRun auditarchive.RetentionResult + for date := monthStart; date.Before(monthEnd); date = date.AddDate(0, 0, 1) { + result, retainErr := service.RetainDate(ctx, date, false) + if retainErr != nil { + return fmt.Errorf("逐日只读演练失败: %w", retainErr) + } + retentionLogger.Info("日留存仿真只读校验通过", zap.String("archive_date", result.ArchiveDate)) + dryRun.EventCount += result.EventCount + dryRun.ResourceCount += result.ResourceCount + dryRun.IntegrationCount += result.IntegrationCount + dryRun.EstimatedBatches += result.EstimatedBatches } summary, err := collectBeforeCleanup(ctx, db, monthStart, monthEnd, dryRun) if err != nil { @@ -125,16 +131,24 @@ func run(ctx context.Context) error { return fmt.Errorf("只归档模式写入了 %d 个清理断点", summary.CleanupMarkersBefore) } - cleanupTask := asynq.NewTask(constants.TaskTypeAuditMonthlyRetention, payload) - if err := taskapp.NewAuditMonthlyRetentionHandler(service, logger, true).Handle(ctx, cleanupTask); err != nil { - return fmt.Errorf("隔离测试库物理清理演练失败: %w", err) + for date := monthStart; date.Before(monthEnd); date = date.AddDate(0, 0, 1) { + result, retainErr := service.RetainDate(ctx, date, true) + if retainErr != nil { + return fmt.Errorf("隔离测试库逐日物理清理演练失败: %w", retainErr) + } + retentionLogger.Info("日留存仿真物理清理完成", zap.String("archive_date", result.ArchiveDate)) } if err := collectAfterCleanup(ctx, db, monthStart, monthEnd, &summary); err != nil { return err } - if summary.TargetRowsAfterCleanup != 0 || summary.BoundaryRowsAfterCleanup != 6 || summary.CleanupMarkersAfter != int64(summary.Days*2) || !summary.RetentionAuditRecorded { - return fmt.Errorf("清理范围复核失败: target=%d boundary=%d markers=%d audit=%t", - summary.TargetRowsAfterCleanup, summary.BoundaryRowsAfterCleanup, summary.CleanupMarkersAfter, summary.RetentionAuditRecorded) + if summary.TargetRowsAfterCleanup != 0 || summary.BoundaryRowsAfterCleanup != 6 || summary.CleanupMarkersAfter != int64(summary.Days*2) { + return fmt.Errorf("清理范围复核失败: target=%d boundary=%d markers=%d", summary.TargetRowsAfterCleanup, summary.BoundaryRowsAfterCleanup, summary.CleanupMarkersAfter) + } + if err := syncRetentionLogger(); err != nil { + return fmt.Errorf("刷新日留存仿真日志失败: %w", err) + } + if _, err := os.Stat(cfg.Logging.RetentionLog.Filename); err != nil { + return fmt.Errorf("独立日留存日志未生成: %w", err) } encoded, _ := sonic.MarshalIndent(summary, "", " ") fmt.Println(string(encoded)) @@ -261,7 +275,12 @@ func archiveMonth(ctx context.Context, db *gorm.DB, service *auditarchive.Servic Updates(map[string]any{"result": constants.IntegrationResultSuccess, "updated_at": time.Now()}).Error; err != nil { return err } - return service.FinalizeIntegrationMonth(ctx, start) + for date := start; date.Before(end); date = date.AddDate(0, 0, 1) { + if err := service.FinalizeIntegrationDate(ctx, date); err != nil { + return err + } + } + return nil } func collectBeforeCleanup(ctx context.Context, db *gorm.DB, start, end time.Time, dryRun auditarchive.RetentionResult) (simulationSummary, error) { @@ -311,10 +330,5 @@ func collectAfterCleanup(ctx context.Context, db *gorm.DB, start, end time.Time, Count(&summary.CleanupMarkersAfter).Error; err != nil { return err } - var auditCount int64 - if err := db.WithContext(ctx).Model(&model.AuditEvent{}).Where("event_id = ?", "evt_retention_"+strings.ReplaceAll(simulationMonth, "-", "_")).Count(&auditCount).Error; err != nil { - return err - } - summary.RetentionAuditRecorded = auditCount == 1 return nil } diff --git a/cmd/worker/main.go b/cmd/worker/main.go index b6e2d9e..8c66760 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -122,6 +122,11 @@ func runWorker(cfg *config.Config) { defer func() { _ = logger.Sync() // 忽略 sync 错误 }() + retentionLogger, syncRetentionLogger := logger.NewRetentionLogger(cfg.Logging.Level, logger.LogRotationConfig{ + Filename: cfg.Logging.RetentionLog.Filename, MaxSize: cfg.Logging.RetentionLog.MaxSize, + MaxBackups: cfg.Logging.RetentionLog.MaxBackups, MaxAge: cfg.Logging.RetentionLog.MaxAge, Compress: cfg.Logging.RetentionLog.Compress, + }) + defer func() { _ = syncRetentionLogger() }() appLogger := logger.GetAppLogger() ctx, cancel := context.WithCancel(context.Background()) @@ -148,7 +153,7 @@ func runWorker(cfg *config.Config) { taskHandler.RegisterHandlers() registerWeComApprovalTasks(taskHandler.GetMux(), runtime, cfg, appLogger) registerAgentRechargeRecoveryTask(taskHandler.GetMux(), runtime, appLogger) - registerAuditArchiveTask(taskHandler.GetMux(), runtime, cfg.Worker.AuditRetentionCleanupEnabled, appLogger) + registerAuditArchiveTask(taskHandler.GetMux(), runtime, cfg.Worker.AuditRetentionCleanupEnabled, appLogger, retentionLogger) outboxHandler := outbox.NewHandler(runtime.outboxConsumers) taskHandler.GetMux().HandleFunc(constants.TaskTypeOutboxDeliver, outboxHandler.Handle) startOutboxRelay(ctx, runtime, cfg.Worker.InstanceName, appLogger) @@ -794,43 +799,32 @@ func registerAsynqScheduleTasks(asynqScheduler *asynq.Scheduler) error { )); err != nil { return fmt.Errorf("注册 Integration Log 每日冷归档定时任务失败: %w", err) } - if _, err := asynqScheduler.Register("CRON_TZ=Asia/Shanghai 0 5 1 * *", asynq.NewTask( - constants.TaskTypeIntegrationMonthlyFinalize, - nil, - asynq.MaxRetry(10), - asynq.Timeout(6*time.Hour), - asynq.Unique(27*24*time.Hour), - asynq.Queue(constants.QueueForTaskType(constants.TaskTypeIntegrationMonthlyFinalize)), - )); err != nil { - return fmt.Errorf("注册 Integration Log 月度最终版本复核任务失败: %w", err) - } - if _, err := asynqScheduler.Register("CRON_TZ=Asia/Shanghai 0 6 1 * *", asynq.NewTask( - constants.TaskTypeAuditMonthlyRetention, + if _, err := asynqScheduler.Register("CRON_TZ=Asia/Shanghai 0 5 * * *", asynq.NewTask( + constants.TaskTypeAuditDailyRetention, nil, asynq.MaxRetry(10), asynq.Timeout(12*time.Hour), - asynq.Unique(27*24*time.Hour), - asynq.Queue(constants.QueueForTaskType(constants.TaskTypeAuditMonthlyRetention)), + asynq.Unique(23*time.Hour), + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeAuditDailyRetention)), )); err != nil { - return fmt.Errorf("注册月度日志留存演练或清理任务失败: %w", err) + return fmt.Errorf("注册日志日留存演练或清理任务失败: %w", err) } return nil } // registerAuditArchiveTask 注册 Audit 与 Integration 冷归档任务处理器。 -func registerAuditArchiveTask(mux *asynq.ServeMux, runtime *workerRuntime, cleanupEnabled bool, appLogger *zap.Logger) { +func registerAuditArchiveTask(mux *asynq.ServeMux, runtime *workerRuntime, cleanupEnabled bool, appLogger, retentionLogger *zap.Logger) { if runtime.storageSvc == nil { appLogger.Warn("对象存储未配置，审计归档任务将在执行时重试") mux.HandleFunc(constants.TaskTypeAuditDailyArchive, task.NewAuditDailyArchiveHandler(nil, appLogger).Handle) integrationHandler := task.NewIntegrationArchiveHandler(nil, appLogger) mux.HandleFunc(constants.TaskTypeIntegrationDailyArchive, integrationHandler.HandleDaily) - mux.HandleFunc(constants.TaskTypeIntegrationMonthlyFinalize, integrationHandler.HandleMonthlyFinalize) - mux.HandleFunc(constants.TaskTypeAuditMonthlyRetention, task.NewAuditMonthlyRetentionHandler(nil, appLogger, cleanupEnabled).Handle) + mux.HandleFunc(constants.TaskTypeAuditDailyRetention, task.NewAuditRetentionHandler(nil, retentionLogger, cleanupEnabled).Handle) return } auditWriter, ok := runtime.workerResult.Services.PaymentAudit.(*auditInfra.Writer) if !ok || auditWriter == nil { - appLogger.Fatal("初始化月度日志留存清理失败：统一审计 Writer 未配置") + appLogger.Fatal("初始化日志日留存清理失败：统一审计 Writer 未配置") } service, err := auditArchiveApp.NewService(runtime.db, runtime.storageSvc.Provider(), constants.AuditArchiveInstanceID, auditWriter) if err != nil { @@ -839,13 +833,11 @@ func registerAuditArchiveTask(mux *asynq.ServeMux, runtime *workerRuntime, clean mux.HandleFunc(constants.TaskTypeAuditDailyArchive, task.NewAuditDailyArchiveHandler(service, appLogger).Handle) integrationHandler := task.NewIntegrationArchiveHandler(service, appLogger) mux.HandleFunc(constants.TaskTypeIntegrationDailyArchive, integrationHandler.HandleDaily) - mux.HandleFunc(constants.TaskTypeIntegrationMonthlyFinalize, integrationHandler.HandleMonthlyFinalize) - mux.HandleFunc(constants.TaskTypeAuditMonthlyRetention, task.NewAuditMonthlyRetentionHandler(service, appLogger, cleanupEnabled).Handle) + mux.HandleFunc(constants.TaskTypeAuditDailyRetention, task.NewAuditRetentionHandler(service, retentionLogger, cleanupEnabled).Handle) appLogger.Info("注册审计归档任务处理器", zap.String("audit_task_type", constants.TaskTypeAuditDailyArchive), zap.String("integration_daily_task_type", constants.TaskTypeIntegrationDailyArchive), - zap.String("integration_monthly_task_type", constants.TaskTypeIntegrationMonthlyFinalize), - zap.String("retention_task_type", constants.TaskTypeAuditMonthlyRetention), + zap.String("retention_task_type", constants.TaskTypeAuditDailyRetention), zap.Bool("retention_cleanup_enabled", cleanupEnabled)) } diff --git a/docs/deployment/production-runbook.md b/docs/deployment/production-runbook.md index cb09fe8..f80692b 100644 --- a/docs/deployment/production-runbook.md +++ b/docs/deployment/production-runbook.md @@ -88,3 +88,12 @@ DB_PASSWORD='<密码>' DB_NAME=<库名> DB_SSLMODE=<模式> \ ## 已确认数据库备份 每日凌晨 02:00 自动备份 `junhong_cmp_prod`：数据库运行在 Docker 容器 `postgres` 中，备份脚本执行 `pg_dump -Fc -Z 6`，写入 `/data/backups/postgresql/<库名>_<时间>.dump`，同时生成 MD5 文件并以 `pg_restore --list` 校验结构；保留 30 天。发布前仍须人工新建一次备份并确认校验通过，不能只依赖凌晨的最近备份。恢复命令待维护者实际演练或确认后补充。 + +## 日审计日志留存启用与恢复 + +日留存由调度 Worker 每日处理 Asia/Shanghai 的昨天及更早连续积压日期；`JUNHONG_WORKER_AUDIT_RETENTION_CLEANUP_ENABLED=false` 时仅验证归档、对象、manifest、数据库数量和日期连续性，不写清理断点、不删除在线数据。 + +1. 发布新 Worker 后保持开关关闭。维护者先核对 `tb_log_archive_run` 中 Audit 与 Integration 两个来源从历史最早在线日期起没有缺失账本；缺失日期必须先受控补归档，不能跳过失败日。 +2. 以关闭开关的 Worker 完成只读演练，观察 `/opt/junhong_cmp/worker/logs/audit-retention.log`：每个日期应有来源计数和耗时；若出现日期、来源、失败分类或安全错误摘要，先修复该日归档或 pending 记录后再演练。 +3. 低峰期将调度 Worker 的 `JUNHONG_WORKER_AUDIT_RETENTION_CLEANUP_ENABLED=true`，重启该 Worker，并持续观察上述独立日志、`tb_log_archive_run.cleanup_started_at` / `cleaned_at` 断点及表大小。任何日期失败都会阻断该日和后续日期，不得手工跳过。 +4. 异常时立即将开关改回 `false` 并重启调度 Worker。已清理日期按已验证的对象和 manifest 执行归档恢复；尚未清理或阻断的日期仍保留在线，无需数据库恢复。恢复后先重新执行只读演练，再决定是否重新开启清理。 diff --git a/internal/application/auditarchive/integration.go b/internal/application/auditarchive/integration.go index 0b9f3e2..0bcdf79 100644 --- a/internal/application/auditarchive/integration.go +++ b/internal/application/auditarchive/integration.go @@ -57,26 +57,9 @@ func (s *Service) ArchiveIntegrationDate(ctx context.Context, archiveDate time.T return s.archiveIntegrationDate(ctx, archiveDate, false) } -// FinalizePreviousIntegrationMonth 复核并终结上一个完整自然月的 Integration Log 归档。 -func (s *Service) FinalizePreviousIntegrationMonth(ctx context.Context) error { - now := time.Now().In(s.location) - return s.FinalizeIntegrationMonth(ctx, now.AddDate(0, -1, 0)) -} - -// FinalizeIntegrationMonth 逐日复核指定完整自然月，并为变化内容创建最终 revision。 -func (s *Service) FinalizeIntegrationMonth(ctx context.Context, month time.Time) error { - monthStart := time.Date(month.In(s.location).Year(), month.In(s.location).Month(), 1, 0, 0, 0, 0, s.location) - currentMonth := time.Now().In(s.location) - currentMonthStart := time.Date(currentMonth.Year(), currentMonth.Month(), 1, 0, 0, 0, 0, s.location) - if !monthStart.Before(currentMonthStart) { - return fmt.Errorf("只能终结已经结束的 Integration Log 完整自然月") - } - for date := monthStart; date.Before(monthStart.AddDate(0, 1, 0)); date = date.AddDate(0, 0, 1) { - if err := s.archiveIntegrationDate(ctx, date, true); err != nil { - return fmt.Errorf("终结 %s Integration Log 归档失败: %w", date.Format(time.DateOnly), err) - } - } - return nil +// FinalizeIntegrationDate 为指定已结束自然日形成 Integration Log 最终归档版本。 +func (s *Service) FinalizeIntegrationDate(ctx context.Context, archiveDate time.Time) error { + return s.archiveIntegrationDate(ctx, archiveDate, true) } func (s *Service) archiveIntegrationDate(ctx context.Context, archiveDate time.Time, final bool) error { @@ -105,8 +88,6 @@ func (s *Service) archiveIntegrationDate(ctx context.Context, archiveDate time.T return pendingErr } if pending > 0 { - _ = s.db.WithContext(ctx).Model(&model.LogArchiveRun{}).Where("id = ?", run.ID). - Updates(map[string]any{"is_final": false, "error_summary": "存在 pending Integration Log，无法形成最终归档", "updated_at": time.Now()}).Error return fmt.Errorf("仍有 %d 条 pending Integration Log，无法形成最终归档", pending) } } diff --git a/internal/application/auditarchive/retention.go b/internal/application/auditarchive/retention.go index 075ccf9..a35294a 100644 --- a/internal/application/auditarchive/retention.go +++ b/internal/application/auditarchive/retention.go @@ -11,7 +11,6 @@ import ( "time" "github.com/bytedance/sonic" - "gorm.io/gorm" "github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/pkg/constants" @@ -19,25 +18,18 @@ import ( const maxManifestBytes = 1024 * 1024 -// RetentionAudit 描述月度物理清理的统一审计事实。 +// RetentionAudit 保留系统审计 Writer 的输入兼容类型。 type RetentionAudit struct { - EventID string - Month string - Summary string - Result string - ErrorSummary string - RangeStart time.Time - RangeEnd time.Time - EventCount int64 - ResourceCount int64 - IntegrationCount int64 - ManifestKeys []string - DurationMS int64 + EventID, Month, Summary, Result, ErrorSummary string + RangeStart, RangeEnd time.Time + EventCount, ResourceCount, IntegrationCount int64 + ManifestKeys []string + DurationMS int64 } -// RetentionResult 是月度留存清理的结构化执行结果。 +// RetentionResult 是单日留存处理结果。 type RetentionResult struct { - Month string + ArchiveDate string EventCount int64 ResourceCount int64 IntegrationCount int64 @@ -46,193 +38,143 @@ type RetentionResult struct { Duration time.Duration } +// RetentionBlockedError 描述阻断日期推进的安全上下文。 +type RetentionBlockedError struct { + ArchiveDate string + Source string + Err error +} + +func (e *RetentionBlockedError) Error() string { + return e.ArchiveDate + " " + e.Source + ": " + e.Err.Error() +} +func (e *RetentionBlockedError) Unwrap() error { return e.Err } + type retentionRuns struct { - audit []*model.LogArchiveRun - integration []*model.LogArchiveRun + audit *model.LogArchiveRun + integration *model.LogArchiveRun } -// CleanupPreviousMonth 校验并物理清理上一个完整自然月的在线审计日志。 -func (s *Service) CleanupPreviousMonth(ctx context.Context) (RetentionResult, error) { - now := time.Now().In(s.location) - return s.CleanupMonth(ctx, now.AddDate(0, -1, 0)) -} - -// ValidatePreviousMonth 只读校验上一个完整自然月的归档与清理门禁。 -func (s *Service) ValidatePreviousMonth(ctx context.Context) (RetentionResult, error) { - now := time.Now().In(s.location) - return s.ValidateMonth(ctx, now.AddDate(0, -1, 0)) -} - -// ValidateMonth 只读校验指定完整自然月，不写清理断点且不删除在线数据。 -func (s *Service) ValidateMonth(ctx context.Context, month time.Time) (result RetentionResult, err error) { +// RetainPendingDays 从最早仍在线的日期连续处理至昨天。cleanup 为 false 时只读校验。 +func (s *Service) RetainPendingDays(ctx context.Context, cleanup bool) ([]RetentionResult, error) { if s.db == nil || s.store == nil { - return result, fmt.Errorf("日志留存演练数据库或对象存储未配置") + return nil, fmt.Errorf("日志日留存数据库或对象存储未配置") } - start, end, err := s.retentionMonthRange(month) + start, end, err := s.pendingRetentionRange(ctx) if err != nil { - return result, err + return nil, err } + results := make([]RetentionResult, 0) + for date := start; date.Before(end); date = date.AddDate(0, 0, 1) { + result, err := s.retainDate(ctx, date, cleanup) + if err != nil { + source := "retention" + if strings.HasPrefix(err.Error(), "Audit") { + source = constants.AuditArchiveSource + } + if strings.HasPrefix(err.Error(), "Integration") { + source = constants.IntegrationArchiveSource + } + return results, &RetentionBlockedError{ArchiveDate: date.Format(time.DateOnly), Source: source, Err: err} + } + results = append(results, result) + } + return results, nil +} + +// RetainDate 校验并按需清理指定已结束自然日，供受控演练使用。 +func (s *Service) RetainDate(ctx context.Context, date time.Time, cleanup bool) (RetentionResult, error) { + return s.retainDate(ctx, date, cleanup) +} + +func (s *Service) pendingRetentionRange(ctx context.Context) (time.Time, time.Time, error) { + today := time.Now().In(s.location) + end := time.Date(today.Year(), today.Month(), today.Day(), 0, 0, 0, 0, s.location) + var earliest *time.Time + for _, item := range []struct{ table, column string }{{"tb_audit_event", "created_at"}, {"tb_integration_log", "created_at"}} { + var value *time.Time + if err := s.db.WithContext(ctx).Table(item.table).Select("MIN(" + item.column + ")").Scan(&value).Error; err != nil { + return time.Time{}, time.Time{}, fmt.Errorf("查询日留存起点失败: %w", err) + } + if value != nil && (earliest == nil || value.Before(*earliest)) { + local := value.In(s.location) + earliest = &local + } + } + if earliest == nil { + return end, end, nil + } + start := time.Date(earliest.Year(), earliest.Month(), earliest.Day(), 0, 0, 0, 0, s.location) + return start, end, nil +} + +func (s *Service) retainDate(ctx context.Context, date time.Time, cleanup bool) (RetentionResult, error) { startedAt := time.Now() - result.Month = start.Format("2006-01") - runs, err := s.loadRetentionRuns(ctx, start, end) + var existing []model.LogArchiveRun + if err := s.db.WithContext(ctx).Where("source IN ? AND archive_date = ? AND instance_id = ?", []string{constants.AuditArchiveSource, constants.IntegrationArchiveSource}, date.Format(time.DateOnly), s.instanceID).Find(&existing).Error; err != nil { + return RetentionResult{}, fmt.Errorf("读取日归档账本失败: %w", err) + } + statuses := map[string]string{} + for _, run := range existing { + statuses[run.Source] = run.Status + } + if statuses[constants.AuditArchiveSource] != constants.ArchiveStatusSuccess { + if err := s.ArchiveDate(ctx, date); err != nil { + return RetentionResult{}, fmt.Errorf("Audit 归档失败: %w", err) + } + } + if statuses[constants.IntegrationArchiveSource] != constants.ArchiveStatusSuccess { + if err := s.ArchiveIntegrationDate(ctx, date); err != nil { + return RetentionResult{}, fmt.Errorf("Integration Log 归档失败: %w", err) + } + } + if err := s.FinalizeIntegrationDate(ctx, date); err != nil { + return RetentionResult{}, fmt.Errorf("Integration Log 最终归档失败: %w", err) + } + runs, err := s.loadRetentionRuns(ctx, date) if err != nil { - return result, err + return RetentionResult{}, err } - if err := s.validateRetentionRuns(ctx, start, end, runs); err != nil { - return result, err + if err := s.validateAuditRetentionDay(ctx, date, runs.audit); err != nil { + return RetentionResult{}, fmt.Errorf("Audit 完整性校验失败: %w", err) } - summarizeRetentionRuns(runs, &result) + if err := s.validateIntegrationRetentionDay(ctx, date, runs.integration); err != nil { + return RetentionResult{}, fmt.Errorf("Integration Log 完整性校验失败: %w", err) + } + result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), EventCount: runs.audit.EventCount, ResourceCount: runs.audit.ResourceCount, IntegrationCount: runs.integration.RecordCount, ManifestKeys: []string{runs.audit.ManifestKey, runs.integration.ManifestKey}} result.EstimatedBatches = estimatedRetentionBatches(result) + if cleanup { + if err := s.cleanupAuditDate(ctx, date, runs.audit); err != nil { + return result, err + } + if err := s.cleanupIntegrationDate(ctx, date, runs.integration); err != nil { + return result, err + } + } result.Duration = time.Since(startedAt) return result, nil } -// CleanupMonth 校验归档硬门禁后按固定顺序物理清理指定完整自然月。 -func (s *Service) CleanupMonth(ctx context.Context, month time.Time) (result RetentionResult, cleanupErr error) { - if s.db == nil || s.store == nil || s.audit == nil { - return result, fmt.Errorf("日志留存清理数据库、对象存储或审计 Writer 未配置") - } - start, end, err := s.retentionMonthRange(month) - if err != nil { - return result, err - } - startedAt := time.Now() - result.Month = start.Format("2006-01") - cleanupErr = s.executeRetention(ctx, start, end, &result) - result.Duration = time.Since(startedAt) - if auditErr := s.recordRetentionAudit(ctx, start, end, result, cleanupErr); auditErr != nil { - if cleanupErr != nil { - return result, fmt.Errorf("%w；记录留存清理失败审计失败: %v", cleanupErr, auditErr) - } - return result, fmt.Errorf("记录留存清理成功审计失败: %w", auditErr) - } - return result, cleanupErr -} - -func (s *Service) retentionMonthRange(month time.Time) (time.Time, time.Time, error) { - start := time.Date(month.In(s.location).Year(), month.In(s.location).Month(), 1, 0, 0, 0, 0, s.location) - end := start.AddDate(0, 1, 0) - now := time.Now().In(s.location) - currentMonth := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, s.location) - if !end.Before(currentMonth) && !end.Equal(currentMonth) { - return time.Time{}, time.Time{}, fmt.Errorf("只能清理已经结束的完整自然月") - } - return start, end, nil -} - -func (s *Service) executeRetention(ctx context.Context, start, end time.Time, result *RetentionResult) error { - started, err := s.retentionCleanupStarted(ctx, start, end) - if err != nil { - return err - } - if !started { - lastDay := end.AddDate(0, 0, -1) - if err := s.ArchiveDate(ctx, lastDay); err != nil { - return fmt.Errorf("完成上月最后一天 Audit 归档失败: %w", err) - } - if err := s.ArchiveIntegrationDate(ctx, lastDay); err != nil { - return fmt.Errorf("完成上月最后一天 Integration Log 归档失败: %w", err) - } - if err := s.FinalizeIntegrationMonth(ctx, start); err != nil { - return err - } - } - runs, err := s.loadRetentionRuns(ctx, start, end) - if err != nil { - return err - } - if err := s.validateRetentionRuns(ctx, start, end, runs); err != nil { - return err - } - summarizeRetentionRuns(runs, result) - if err := s.cleanupAuditMonth(ctx, start, end, runs.audit); err != nil { - return err - } - return s.cleanupIntegrationMonth(ctx, start, end, runs.integration) -} - -func (s *Service) retentionCleanupStarted(ctx context.Context, start, end time.Time) (bool, error) { - var count int64 - err := s.db.WithContext(ctx).Model(&model.LogArchiveRun{}). - Where("archive_date >= ? AND archive_date < ? AND instance_id = ? AND cleanup_started_at IS NOT NULL", - start.Format(time.DateOnly), end.Format(time.DateOnly), s.instanceID). - Count(&count).Error - if err != nil { - return false, fmt.Errorf("读取月度清理断点失败: %w", err) - } - return count > 0, nil -} - -func (s *Service) loadRetentionRuns(ctx context.Context, start, end time.Time) (retentionRuns, error) { +func (s *Service) loadRetentionRuns(ctx context.Context, date time.Time) (retentionRuns, error) { var rows []model.LogArchiveRun - err := s.db.WithContext(ctx).Where( - "source IN ? AND archive_date >= ? AND archive_date < ? AND instance_id = ?", - []string{constants.AuditArchiveSource, constants.IntegrationArchiveSource}, start.Format(time.DateOnly), end.Format(time.DateOnly), s.instanceID, - ).Order("archive_date ASC, source ASC").Find(&rows).Error - if err != nil { - return retentionRuns{}, fmt.Errorf("读取月度归档账本失败: %w", err) + if err := s.db.WithContext(ctx).Where("source IN ? AND archive_date = ? AND instance_id = ?", []string{constants.AuditArchiveSource, constants.IntegrationArchiveSource}, date.Format(time.DateOnly), s.instanceID).Find(&rows).Error; err != nil { + return retentionRuns{}, fmt.Errorf("读取日归档账本失败: %w", err) } - days := int(end.Sub(start).Hours() / 24) - if len(rows) != days*2 { - return retentionRuns{}, fmt.Errorf("月度归档账本缺日：期望 %d 条，实际 %d 条", days*2, len(rows)) - } - runs := retentionRuns{audit: make([]*model.LogArchiveRun, 0, days), integration: make([]*model.LogArchiveRun, 0, days)} + runs := retentionRuns{} for index := range rows { - run := &rows[index] - switch run.Source { + switch rows[index].Source { case constants.AuditArchiveSource: - runs.audit = append(runs.audit, run) + runs.audit = &rows[index] case constants.IntegrationArchiveSource: - runs.integration = append(runs.integration, run) + runs.integration = &rows[index] } } - if len(runs.audit) != days || len(runs.integration) != days { - return retentionRuns{}, fmt.Errorf("月度 Audit 或 Integration 归档账本不完整") + if runs.audit == nil || runs.integration == nil { + return retentionRuns{}, fmt.Errorf("Audit 或 Integration Log 归档账本缺失") } return runs, nil } -func (s *Service) validateRetentionRuns(ctx context.Context, start, end time.Time, runs retentionRuns) error { - if err := validateCleanupLedgerState(runs.audit); err != nil { - return fmt.Errorf("Audit 清理断点非法: %w", err) - } - if err := validateCleanupLedgerState(runs.integration); err != nil { - return fmt.Errorf("Integration 清理断点非法: %w", err) - } - for index := range runs.audit { - date := start.AddDate(0, 0, index) - if err := s.validateAuditRetentionDay(ctx, date, runs.audit[index]); err != nil { - return fmt.Errorf("%s Audit 清理门禁失败: %w", date.Format(time.DateOnly), err) - } - if err := s.validateIntegrationRetentionDay(ctx, date, runs.integration[index]); err != nil { - return fmt.Errorf("%s Integration 清理门禁失败: %w", date.Format(time.DateOnly), err) - } - } - return nil -} - -func validateCleanupLedgerState(runs []*model.LogArchiveRun) error { - started, cleaned := 0, 0 - for _, run := range runs { - if run.CleanupStartedAt != nil { - started++ - } - if run.CleanedAt != nil { - cleaned++ - } - } - if started != 0 && started != len(runs) { - return fmt.Errorf("清理开始断点不是整月原子状态") - } - if cleaned != 0 && cleaned != len(runs) { - return fmt.Errorf("清理完成断点不是整月原子状态") - } - if cleaned > 0 && started == 0 { - return fmt.Errorf("清理完成但缺少开始断点") - } - return nil -} - func (s *Service) validateAuditRetentionDay(ctx context.Context, date time.Time, run *model.LogArchiveRun) error { if err := validateRunBase(run, date, constants.AuditArchiveSchemaVersion, false); err != nil { return err @@ -258,25 +200,21 @@ func (s *Service) validateIntegrationRetentionDay(ctx context.Context, date time if err != nil { return err } - if run.CleanedAt != nil { - if count != 0 { - return fmt.Errorf("已标记清理完成但数据库仍有 %d 条记录", count) + if run.CleanedAt != nil && count != 0 { + return fmt.Errorf("已标记清理完成但数据库仍有 %d 条记录", count) + } + if run.CleanupStartedAt != nil && count > run.RecordCount { + return fmt.Errorf("续跑窗口记录数超过最终归档数量") + } + if run.CleanupStartedAt == nil { + file, err := s.buildIntegrationArchiveFile(ctx, run.RangeStart, run.RangeEnd) + if err != nil { + return err } - return nil - } - if run.CleanupStartedAt != nil { - if count > run.RecordCount { - return fmt.Errorf("续跑窗口记录数超过最终归档数量") + defer os.Remove(file.path) + if file.recordCount != run.RecordCount || file.sha256 != run.SHA256 { + return fmt.Errorf("数据库当前 Integration 内容与最终 revision 不一致") } - return nil - } - file, err := s.buildIntegrationArchiveFile(ctx, run.RangeStart, run.RangeEnd) - if err != nil { - return err - } - defer os.Remove(file.path) - if file.recordCount != run.RecordCount || file.sha256 != run.SHA256 { - return fmt.Errorf("数据库当前 Integration 内容与最终 revision 不一致") } return nil } @@ -285,8 +223,7 @@ func validateRunBase(run *model.LogArchiveRun, date time.Time, schema string, fi if run.Status != constants.ArchiveStatusSuccess || run.SchemaVersion != schema { return fmt.Errorf("归档状态或 schema version 不符合清理要求") } - if run.ArchiveDate.Format(time.DateOnly) != date.Format(time.DateOnly) || - !run.RangeStart.Equal(date) || !run.RangeEnd.Equal(date.AddDate(0, 0, 1)) { + if run.ArchiveDate.Format(time.DateOnly) != date.Format(time.DateOnly) || !run.RangeStart.Equal(date) || !run.RangeEnd.Equal(date.AddDate(0, 0, 1)) { return fmt.Errorf("归档日期或半开时间范围不一致") } if final && !run.IsFinal { @@ -297,21 +234,14 @@ func validateRunBase(run *model.LogArchiveRun, date time.Time, schema string, fi } return nil } - func validateRemainingCounts(run *model.LogArchiveRun, events, resources int64) error { - if run.CleanedAt != nil { - if events != 0 || resources != 0 { - return fmt.Errorf("已标记清理完成但数据库仍有事件或资源") - } - return nil + if run.CleanedAt != nil && (events != 0 || resources != 0) { + return fmt.Errorf("已标记清理完成但数据库仍有事件或资源") } - if run.CleanupStartedAt != nil { - if events > run.EventCount || resources > run.ResourceCount { - return fmt.Errorf("续跑窗口数量超过已归档数量") - } - return nil + if run.CleanupStartedAt != nil && (events > run.EventCount || resources > run.ResourceCount) { + return fmt.Errorf("续跑窗口数量超过已归档数量") } - if events != run.EventCount || resources != run.ResourceCount { + if run.CleanupStartedAt == nil && (events != run.EventCount || resources != run.ResourceCount) { return fmt.Errorf("数据库事件或资源数量与 manifest 不一致") } return nil @@ -322,43 +252,27 @@ func (s *Service) validateAuditManifest(ctx context.Context, run *model.LogArchi if err := s.readManifest(ctx, run.ManifestKey, &manifest); err != nil { return err } - if manifest.Source != run.Source || manifest.SchemaVersion != run.SchemaVersion || manifest.Status != constants.ArchiveStatusSuccess || - manifest.ArchiveDate != run.ArchiveDate.Format(time.DateOnly) || manifest.Timezone != constants.AuditArchiveTimezone || - manifest.InstanceID != run.InstanceID || !manifest.RangeStart.Equal(run.RangeStart) || !manifest.RangeEnd.Equal(run.RangeEnd) || - manifest.EventCount != run.EventCount || manifest.ResourceCount != run.ResourceCount || - manifest.CompressedBytes != run.CompressedBytes || manifest.ObjectKey != run.ObjectKey || - manifest.SHA256 != run.SHA256 || manifest.Revision != run.Revision { + if manifest.Source != run.Source || manifest.SchemaVersion != run.SchemaVersion || manifest.Status != constants.ArchiveStatusSuccess || manifest.ArchiveDate != run.ArchiveDate.Format(time.DateOnly) || manifest.Timezone != constants.AuditArchiveTimezone || manifest.InstanceID != run.InstanceID || !manifest.RangeStart.Equal(run.RangeStart) || !manifest.RangeEnd.Equal(run.RangeEnd) || manifest.EventCount != run.EventCount || manifest.ResourceCount != run.ResourceCount || manifest.CompressedBytes != run.CompressedBytes || manifest.ObjectKey != run.ObjectKey || manifest.SHA256 != run.SHA256 || manifest.Revision != run.Revision { return fmt.Errorf("Audit manifest 与 ledger 不一致") } - if err := s.verifyObject(ctx, run.ManifestKey, -1, map[string]string{ - "source": constants.AuditArchiveSource, "data-sha256": run.SHA256, "revision": strconv.Itoa(run.Revision), - }); err != nil { + if err := s.verifyObject(ctx, run.ManifestKey, -1, map[string]string{"source": constants.AuditArchiveSource, "data-sha256": run.SHA256, "revision": strconv.Itoa(run.Revision)}); err != nil { return err } return s.verifyRetentionObject(ctx, run, false) } - func (s *Service) validateIntegrationManifest(ctx context.Context, run *model.LogArchiveRun) error { var manifest integrationArchiveManifest if err := s.readManifest(ctx, run.ManifestKey, &manifest); err != nil { return err } - if manifest.Source != run.Source || manifest.SchemaVersion != run.SchemaVersion || manifest.Status != constants.ArchiveStatusSuccess || !manifest.Final || - manifest.ArchiveDate != run.ArchiveDate.Format(time.DateOnly) || manifest.Timezone != constants.AuditArchiveTimezone || - manifest.InstanceID != run.InstanceID || !manifest.RangeStart.Equal(run.RangeStart) || !manifest.RangeEnd.Equal(run.RangeEnd) || - manifest.RecordCount != run.RecordCount || manifest.CompressedBytes != run.CompressedBytes || - manifest.ObjectKey != run.ObjectKey || manifest.SHA256 != run.SHA256 || manifest.Revision != run.Revision { + if manifest.Source != run.Source || manifest.SchemaVersion != run.SchemaVersion || manifest.Status != constants.ArchiveStatusSuccess || !manifest.Final || manifest.ArchiveDate != run.ArchiveDate.Format(time.DateOnly) || manifest.Timezone != constants.AuditArchiveTimezone || manifest.InstanceID != run.InstanceID || !manifest.RangeStart.Equal(run.RangeStart) || !manifest.RangeEnd.Equal(run.RangeEnd) || manifest.RecordCount != run.RecordCount || manifest.CompressedBytes != run.CompressedBytes || manifest.ObjectKey != run.ObjectKey || manifest.SHA256 != run.SHA256 || manifest.Revision != run.Revision { return fmt.Errorf("Integration manifest 与最终 ledger 不一致") } - if err := s.verifyObject(ctx, run.ManifestKey, -1, map[string]string{ - "source": constants.IntegrationArchiveSource, "data-sha256": run.SHA256, - "revision": strconv.Itoa(run.Revision), "final": "true", - }); err != nil { + if err := s.verifyObject(ctx, run.ManifestKey, -1, map[string]string{"source": constants.IntegrationArchiveSource, "data-sha256": run.SHA256, "revision": strconv.Itoa(run.Revision), "final": "true"}); err != nil { return err } return s.verifyRetentionObject(ctx, run, true) } - func (s *Service) readManifest(ctx context.Context, key string, target any) error { object, err := s.store.Stat(ctx, key) if err != nil { @@ -384,13 +298,8 @@ func (s *Service) readManifest(ctx context.Context, key string, target any) erro } return nil } - func (s *Service) verifyRetentionObject(ctx context.Context, run *model.LogArchiveRun, final bool) error { - metadata := map[string]string{ - "schema-version": run.SchemaVersion, "source": run.Source, - "archive-date": run.RangeStart.Format(time.DateOnly), "timezone": constants.AuditArchiveTimezone, - "sha256": run.SHA256, "revision": strconv.Itoa(run.Revision), - } + metadata := map[string]string{"schema-version": run.SchemaVersion, "source": run.Source, "archive-date": run.RangeStart.Format(time.DateOnly), "timezone": constants.AuditArchiveTimezone, "sha256": run.SHA256, "revision": strconv.Itoa(run.Revision)} if run.Source == constants.AuditArchiveSource { metadata["event-count"] = strconv.FormatInt(run.EventCount, 10) metadata["resource-count"] = strconv.FormatInt(run.ResourceCount, 10) @@ -419,53 +328,35 @@ func (s *Service) verifyRetentionObject(ctx context.Context, run *model.LogArchi } return nil } - -func summarizeRetentionRuns(runs retentionRuns, result *RetentionResult) { - result.ManifestKeys = make([]string, 0, len(runs.audit)+len(runs.integration)) - for _, run := range runs.audit { - result.EventCount += run.EventCount - result.ResourceCount += run.ResourceCount - result.ManifestKeys = append(result.ManifestKeys, run.ManifestKey) - } - for _, run := range runs.integration { - result.IntegrationCount += run.RecordCount - result.ManifestKeys = append(result.ManifestKeys, run.ManifestKey) - } -} - func estimatedRetentionBatches(result RetentionResult) int64 { - batchSize := int64(constants.AuditRetentionDeleteBatchSize) - return (result.EventCount+batchSize-1)/batchSize + - (result.ResourceCount+batchSize-1)/batchSize + - (result.IntegrationCount+batchSize-1)/batchSize + batch := int64(constants.AuditRetentionDeleteBatchSize) + return (result.EventCount+batch-1)/batch + (result.ResourceCount+batch-1)/batch + (result.IntegrationCount+batch-1)/batch } - -func (s *Service) cleanupAuditMonth(ctx context.Context, start, end time.Time, runs []*model.LogArchiveRun) error { - if allRunsCleaned(runs) { +func (s *Service) cleanupAuditDate(ctx context.Context, date time.Time, run *model.LogArchiveRun) error { + if run.CleanedAt != nil { return nil } - if err := s.markCleanupStarted(ctx, constants.AuditArchiveSource, start, end); err != nil { + if err := s.markCleanupStarted(ctx, constants.AuditArchiveSource, date); err != nil { return err } - if err := s.deleteAuditResources(ctx, start, end); err != nil { + if err := s.deleteAuditResources(ctx, date, date.AddDate(0, 0, 1)); err != nil { return err } - if err := s.deleteAuditEvents(ctx, start, end); err != nil { + if err := s.deleteAuditEvents(ctx, date, date.AddDate(0, 0, 1)); err != nil { return err } - return s.markCleaned(ctx, constants.AuditArchiveSource, start, end) + return s.markCleaned(ctx, constants.AuditArchiveSource, date) } - -func (s *Service) cleanupIntegrationMonth(ctx context.Context, start, end time.Time, runs []*model.LogArchiveRun) error { - if allRunsCleaned(runs) { +func (s *Service) cleanupIntegrationDate(ctx context.Context, date time.Time, run *model.LogArchiveRun) error { + if run.CleanedAt != nil { return nil } - if err := s.markCleanupStarted(ctx, constants.IntegrationArchiveSource, start, end); err != nil { + if err := s.markCleanupStarted(ctx, constants.IntegrationArchiveSource, date); err != nil { return err } + end := date.AddDate(0, 0, 1) for { - subquery := s.db.Model(&model.IntegrationLog{}).Select("id"). - Where("created_at >= ? AND created_at < ?", start, end).Order("id ASC").Limit(constants.AuditRetentionDeleteBatchSize) + subquery := s.db.Model(&model.IntegrationLog{}).Select("id").Where("created_at >= ? AND created_at < ?", date, end).Order("id ASC").Limit(constants.AuditRetentionDeleteBatchSize) deleted := s.db.WithContext(ctx).Where("id IN (?)", subquery).Delete(&model.IntegrationLog{}) if deleted.Error != nil { return fmt.Errorf("分批物理删除 Integration Log 失败: %w", deleted.Error) @@ -474,15 +365,11 @@ func (s *Service) cleanupIntegrationMonth(ctx context.Context, start, end time.T break } } - return s.markCleaned(ctx, constants.IntegrationArchiveSource, start, end) + return s.markCleaned(ctx, constants.IntegrationArchiveSource, date) } - func (s *Service) deleteAuditResources(ctx context.Context, start, end time.Time) error { for { - subquery := s.db.Model(&model.AuditEventResource{}).Select("tb_audit_event_resource.id"). - Joins("JOIN tb_audit_event ON tb_audit_event.id = tb_audit_event_resource.audit_event_id"). - Where("tb_audit_event.created_at >= ? AND tb_audit_event.created_at < ?", start, end). - Order("tb_audit_event_resource.id ASC").Limit(constants.AuditRetentionDeleteBatchSize) + subquery := s.db.Model(&model.AuditEventResource{}).Select("tb_audit_event_resource.id").Joins("JOIN tb_audit_event ON tb_audit_event.id = tb_audit_event_resource.audit_event_id").Where("tb_audit_event.created_at >= ? AND tb_audit_event.created_at < ?", start, end).Order("tb_audit_event_resource.id ASC").Limit(constants.AuditRetentionDeleteBatchSize) deleted := s.db.WithContext(ctx).Where("id IN (?)", subquery).Delete(&model.AuditEventResource{}) if deleted.Error != nil { return fmt.Errorf("分批物理删除 Audit Event Resource 失败: %w", deleted.Error) @@ -492,11 +379,9 @@ func (s *Service) deleteAuditResources(ctx context.Context, start, end time.Time } } } - func (s *Service) deleteAuditEvents(ctx context.Context, start, end time.Time) error { for { - subquery := s.db.Model(&model.AuditEvent{}).Select("id"). - Where("created_at >= ? AND created_at < ?", start, end).Order("id ASC").Limit(constants.AuditRetentionDeleteBatchSize) + subquery := s.db.Model(&model.AuditEvent{}).Select("id").Where("created_at >= ? AND created_at < ?", start, end).Order("id ASC").Limit(constants.AuditRetentionDeleteBatchSize) deleted := s.db.WithContext(ctx).Where("id IN (?)", subquery).Delete(&model.AuditEvent{}) if deleted.Error != nil { return fmt.Errorf("分批物理删除 Audit Event 失败: %w", deleted.Error) @@ -506,75 +391,19 @@ func (s *Service) deleteAuditEvents(ctx context.Context, start, end time.Time) e } } } - -func (s *Service) markCleanupStarted(ctx context.Context, source string, start, end time.Time) error { +func (s *Service) markCleanupStarted(ctx context.Context, source string, date time.Time) error { now := time.Now() - result := s.db.WithContext(ctx).Model(&model.LogArchiveRun{}). - Where("source = ? AND archive_date >= ? AND archive_date < ? AND instance_id = ? AND cleanup_started_at IS NULL", - source, start.Format(time.DateOnly), end.Format(time.DateOnly), s.instanceID). - Updates(map[string]any{"cleanup_started_at": now, "updated_at": now}) + result := s.db.WithContext(ctx).Model(&model.LogArchiveRun{}).Where("source = ? AND archive_date = ? AND instance_id = ? AND cleanup_started_at IS NULL", source, date.Format(time.DateOnly), s.instanceID).Updates(map[string]any{"cleanup_started_at": now, "updated_at": now}) if result.Error != nil { - return fmt.Errorf("记录月度清理开始断点失败: %w", result.Error) - } - return s.validateCleanupMarkerCount(ctx, source, start, end, "cleanup_started_at IS NOT NULL", "开始") -} - -func (s *Service) markCleaned(ctx context.Context, source string, start, end time.Time) error { - now := time.Now() - result := s.db.WithContext(ctx).Model(&model.LogArchiveRun{}). - Where("source = ? AND archive_date >= ? AND archive_date < ? AND instance_id = ? AND cleanup_started_at IS NOT NULL", - source, start.Format(time.DateOnly), end.Format(time.DateOnly), s.instanceID). - Updates(map[string]any{"cleaned_at": now, "updated_at": now}) - if result.Error != nil { - return fmt.Errorf("记录月度清理完成断点失败: %w", result.Error) - } - return s.validateCleanupMarkerCount(ctx, source, start, end, "cleaned_at IS NOT NULL", "完成") -} - -func (s *Service) validateCleanupMarkerCount(ctx context.Context, source string, start, end time.Time, marker, label string) error { - var count int64 - err := s.db.WithContext(ctx).Model(&model.LogArchiveRun{}). - Where("source = ? AND archive_date >= ? AND archive_date < ? AND instance_id = ? AND "+marker, - source, start.Format(time.DateOnly), end.Format(time.DateOnly), s.instanceID). - Count(&count).Error - if err != nil { - return fmt.Errorf("复核月度清理%s断点失败: %w", label, err) - } - expected := int64(end.Sub(start).Hours() / 24) - if count != expected { - return fmt.Errorf("月度清理%s断点不完整：期望 %d 条，实际 %d 条", label, expected, count) + return fmt.Errorf("记录日清理开始断点失败: %w", result.Error) } return nil } - -func allRunsCleaned(runs []*model.LogArchiveRun) bool { - return len(runs) > 0 && runs[0].CleanedAt != nil -} - -func (s *Service) recordRetentionAudit(ctx context.Context, start, end time.Time, result RetentionResult, cleanupErr error) error { - audit := RetentionAudit{ - Month: result.Month, RangeStart: start, RangeEnd: end, - EventCount: result.EventCount, ResourceCount: result.ResourceCount, - IntegrationCount: result.IntegrationCount, ManifestKeys: result.ManifestKeys, - DurationMS: result.Duration.Milliseconds(), Result: constants.AuditResultSuccess, - Summary: "完成已归档在线日志月度物理清理", - EventID: "evt_retention_" + strings.ReplaceAll(result.Month, "-", "_"), +func (s *Service) markCleaned(ctx context.Context, source string, date time.Time) error { + now := time.Now() + result := s.db.WithContext(ctx).Model(&model.LogArchiveRun{}).Where("source = ? AND archive_date = ? AND instance_id = ? AND cleanup_started_at IS NOT NULL", source, date.Format(time.DateOnly), s.instanceID).Updates(map[string]any{"cleaned_at": now, "updated_at": now}) + if result.Error != nil { + return fmt.Errorf("记录日清理完成断点失败: %w", result.Error) } - if cleanupErr != nil { - audit.EventID = "" - audit.Result = constants.AuditResultFailed - audit.Summary = "已归档在线日志月度物理清理失败" - audit.ErrorSummary = truncateRetentionError(cleanupErr) - } - return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { - return s.audit.WriteRetentionCleanup(ctx, tx, audit) - }) -} - -func truncateRetentionError(err error) string { - value := []rune(err.Error()) - if len(value) > 500 { - value = value[:500] - } - return string(value) + return nil } diff --git a/internal/query/retention/retention.go b/internal/query/retention/retention.go index 13211c7..0ed0c4f 100644 --- a/internal/query/retention/retention.go +++ b/internal/query/retention/retention.go @@ -13,75 +13,108 @@ import ( "github.com/break/junhong_cmp_fiber/pkg/errors" ) -// Source 表示受在线留存边界约束的数据源。 type Source string const ( - // SourceAudit 表示统一审计事件。 - SourceAudit Source = constants.AuditArchiveSource - // SourceIntegration 表示外部交互日志。 + SourceAudit Source = constants.AuditArchiveSource SourceIntegration Source = constants.IntegrationArchiveSource ) -// Info 是查询响应公开的在线留存边界。 type Info struct { OnlineFrom time.Time `json:"online_from" description:"当前可在线查询的最早时间"` ArchivedBefore *time.Time `json:"archived_before" description:"早于该时间的数据已归档；尚未清理时为空"` Timezone string `json:"timezone" description:"留存自然日时区"` } -// Load 从归档账本读取已完成物理清理的数据边界。 +// Load 仅公开从最早在线日期开始连续完成物理清理的边界，绝不跨越清理空洞。 func Load(ctx context.Context, db *gorm.DB, sources ...Source) (Info, error) { location, err := time.LoadLocation(constants.AuditArchiveTimezone) if err != nil { return Info{}, errors.Wrap(errors.CodeInternalError, err, "加载审计留存时区失败") } - now := time.Now().In(location) - info := Info{OnlineFrom: time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, location), Timezone: constants.AuditArchiveTimezone} + boundaries := make([]sourceRetention, 0, len(sources)) for _, source := range sources { - boundary, cleaned, err := sourceBoundary(ctx, db, source, location) + boundary, err := sourceBoundary(ctx, db, source, location) if err != nil { return Info{}, err } - if boundary.Before(info.OnlineFrom) && info.ArchivedBefore == nil { - info.OnlineFrom = boundary + boundaries = append(boundaries, boundary) + } + now := time.Now().In(location) + fallback := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, location) + info := Info{OnlineFrom: fallback, Timezone: constants.AuditArchiveTimezone} + if len(boundaries) == 0 { + return info, nil + } + for _, boundary := range boundaries { + if boundary.onlineFrom.Before(info.OnlineFrom) { + info.OnlineFrom = boundary.onlineFrom } - if cleaned && (info.ArchivedBefore == nil || boundary.After(*info.ArchivedBefore)) { - value := boundary - info.ArchivedBefore = &value - info.OnlineFrom = boundary + } + if len(boundaries) == 1 && boundaries[0].cleaned { + value := boundaries[0].onlineFrom + info.ArchivedBefore = &value + return info, nil + } + if len(boundaries) > 1 { + common := boundaries[0].onlineFrom + allCleaned := boundaries[0].cleaned + for _, boundary := range boundaries[1:] { + if boundary.onlineFrom.Before(common) { + common = boundary.onlineFrom + } + allCleaned = allCleaned && boundary.cleaned + } + if allCleaned { + info.OnlineFrom = common + info.ArchivedBefore = &common } } return info, nil } -func sourceBoundary(ctx context.Context, db *gorm.DB, source Source, location *time.Location) (time.Time, bool, error) { - var cleanedEnd sql.NullTime - if err := db.WithContext(ctx).Model(&model.LogArchiveRun{}). - Where("source = ? AND cleaned_at IS NOT NULL", source). - Select("MAX(range_end)").Scan(&cleanedEnd).Error; err != nil { - return time.Time{}, false, errors.Wrap(errors.CodeDatabaseError, err, "查询审计留存清理边界失败") - } - if cleanedEnd.Valid { - return cleanedEnd.Time.In(location), true, nil - } +type sourceRetention struct { + onlineFrom time.Time + cleaned bool +} - var earliest sql.NullTime - table, column := "tb_audit_event", "occurred_at" +func sourceBoundary(ctx context.Context, db *gorm.DB, source Source, location *time.Location) (sourceRetention, error) { + table, column := "tb_audit_event", "created_at" if source == SourceIntegration { table, column = "tb_integration_log", "created_at" } - if err := db.WithContext(ctx).Table(table).Select("MIN(" + column + ")").Scan(&earliest).Error; err != nil { - return time.Time{}, false, errors.Wrap(errors.CodeDatabaseError, err, "查询审计在线数据边界失败") + var earliestOnline, earliestLedger sql.NullTime + if err := db.WithContext(ctx).Table(table).Select("MIN(" + column + ")").Scan(&earliestOnline).Error; err != nil { + return sourceRetention{}, errors.Wrap(errors.CodeDatabaseError, err, "查询审计在线数据边界失败") } - if earliest.Valid { - return earliest.Time.In(location), false, nil + if err := db.WithContext(ctx).Model(&model.LogArchiveRun{}).Where("source = ?", source).Select("MIN(archive_date)").Scan(&earliestLedger).Error; err != nil { + return sourceRetention{}, errors.Wrap(errors.CodeDatabaseError, err, "查询审计留存账本边界失败") } - now := time.Now().In(location) - return time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, location), false, nil + if !earliestOnline.Valid && !earliestLedger.Valid { + now := time.Now().In(location) + return sourceRetention{onlineFrom: time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, location)}, nil + } + start := earliestLedger.Time.In(location) + if earliestOnline.Valid && (!earliestLedger.Valid || earliestOnline.Time.Before(earliestLedger.Time)) { + start = earliestOnline.Time.In(location) + } + start = time.Date(start.Year(), start.Month(), start.Day(), 0, 0, 0, 0, location) + var rows []model.LogArchiveRun + if err := db.WithContext(ctx).Where("source = ? AND archive_date >= ?", source, start.Format(time.DateOnly)).Order("archive_date ASC").Find(&rows).Error; err != nil { + return sourceRetention{}, errors.Wrap(errors.CodeDatabaseError, err, "查询审计留存清理边界失败") + } + expected := start + for _, row := range rows { + date := row.ArchiveDate + date = time.Date(date.Year(), date.Month(), date.Day(), 0, 0, 0, 0, location) + if !date.Equal(expected) || row.CleanedAt == nil { + break + } + expected = expected.AddDate(0, 0, 1) + } + return sourceRetention{onlineFrom: expected, cleaned: !expected.Equal(start)}, nil } -// NormalizeRange 将缺省范围收敛到在线窗口，并拒绝归档或跨边界查询。 func NormalizeRange(info Info, from, to *time.Time, maxRange ...time.Duration) (*time.Time, *time.Time, error) { explicitFrom := from != nil if from != nil && info.ArchivedBefore != nil && from.Before(info.OnlineFrom) { @@ -110,7 +143,6 @@ func NormalizeRange(info Info, from, to *time.Time, maxRange ...time.Duration) ( } return from, to, nil } - func archivedError(info Info) error { return errors.NewWithData(errors.CodeAuditDataArchived, map[string]any{"retention": info}) } diff --git a/internal/task/audit_monthly_retention.go b/internal/task/audit_monthly_retention.go index 6d67a13..e9ec063 100644 --- a/internal/task/audit_monthly_retention.go +++ b/internal/task/audit_monthly_retention.go @@ -2,101 +2,47 @@ package task import ( "context" + stderrors "errors" "fmt" - "time" - "github.com/bytedance/sonic" "github.com/hibiken/asynq" "go.uber.org/zap" "github.com/break/junhong_cmp_fiber/internal/application/auditarchive" ) -// AuditMonthlyRetentionPayload 是人工补跑月度清理时可选的任务载荷。 -type AuditMonthlyRetentionPayload struct { - ArchiveMonth string `json:"archive_month"` -} - -// AuditMonthlyRetentionHandler 处理归档完整性门禁与上月在线日志物理清理。 -type AuditMonthlyRetentionHandler struct { +// AuditRetentionHandler 连续校验并按日物理清理已结束的在线日志。 +type AuditRetentionHandler struct { service *auditarchive.Service logger *zap.Logger cleanupEnabled bool } -// NewAuditMonthlyRetentionHandler 创建月度日志留存清理处理器。 -func NewAuditMonthlyRetentionHandler(service *auditarchive.Service, logger *zap.Logger, cleanupEnabled bool) *AuditMonthlyRetentionHandler { - return &AuditMonthlyRetentionHandler{service: service, logger: logger, cleanupEnabled: cleanupEnabled} +// NewAuditRetentionHandler 创建日留存处理器。 +func NewAuditRetentionHandler(service *auditarchive.Service, logger *zap.Logger, cleanupEnabled bool) *AuditRetentionHandler { + return &AuditRetentionHandler{service: service, logger: logger, cleanupEnabled: cleanupEnabled} } -// Handle 校验整月归档后按固定顺序分批物理删除 PostgreSQL 在线日志。 -func (h *AuditMonthlyRetentionHandler) Handle(ctx context.Context, task *asynq.Task) error { - if !h.cleanupEnabled { - return h.handleDryRun(ctx, task) - } +// Handle 在关闭清理开关时只读校验，开启后从最早未完成日连续删除。 +func (h *AuditRetentionHandler) Handle(ctx context.Context, _ *asynq.Task) error { if h.service == nil { - return fmt.Errorf("月度日志留存清理服务未配置") + return fmt.Errorf("日志日留存服务未配置") } - startedAt := time.Now() - var result auditarchive.RetentionResult - var err error - if len(task.Payload()) == 0 { - result, err = h.service.CleanupPreviousMonth(ctx) - } else { - var payload AuditMonthlyRetentionPayload - if unmarshalErr := sonic.Unmarshal(task.Payload(), &payload); unmarshalErr != nil { - return fmt.Errorf("解析月度日志留存清理任务载荷失败: %w", unmarshalErr) - } - month, parseErr := parseArchiveMonth(payload.ArchiveMonth) - if parseErr != nil { - return parseErr - } - result, err = h.service.CleanupMonth(ctx, month) - } - fields := []zap.Field{ - zap.String("archive_month", result.Month), zap.Int64("audit_event_count", result.EventCount), - zap.Int64("event_resource_count", result.ResourceCount), zap.Int64("integration_log_count", result.IntegrationCount), - zap.Duration("duration", time.Since(startedAt)), zap.Int("manifest_count", len(result.ManifestKeys)), + results, err := h.service.RetainPendingDays(ctx, h.cleanupEnabled) + for _, result := range results { + h.logger.Info("日志日留存处理完成", + zap.String("archive_date", result.ArchiveDate), zap.Bool("cleanup_enabled", h.cleanupEnabled), + zap.Int64("audit_event_count", result.EventCount), zap.Int64("event_resource_count", result.ResourceCount), + zap.Int64("integration_log_count", result.IntegrationCount), zap.Duration("duration", result.Duration)) } if err != nil { - fields = append(fields, zap.String("severity", "critical"), zap.Error(err)) - h.logger.Error("月度日志留存清理失败，PostgreSQL 整月清理已阻断或等待断点续跑", fields...) + var blocked *auditarchive.RetentionBlockedError + if stderrors.As(err, &blocked) { + h.logger.Error("日志日留存日期推进已阻断", zap.String("archive_date", blocked.ArchiveDate), zap.String("source", blocked.Source), zap.String("failure_category", "archive_or_validation"), zap.Error(err)) + } else { + h.logger.Error("日志日留存日期推进已阻断", zap.String("source", "retention"), zap.String("failure_category", "internal"), zap.Error(err)) + } return err } - h.logger.Info("月度日志留存清理完成", fields...) - return nil -} - -func (h *AuditMonthlyRetentionHandler) handleDryRun(ctx context.Context, task *asynq.Task) error { - if h.service == nil { - return fmt.Errorf("月度日志留存演练服务未配置") - } - var result auditarchive.RetentionResult - var err error - if len(task.Payload()) == 0 { - result, err = h.service.ValidatePreviousMonth(ctx) - } else { - var payload AuditMonthlyRetentionPayload - if unmarshalErr := sonic.Unmarshal(task.Payload(), &payload); unmarshalErr != nil { - return fmt.Errorf("解析月度日志留存演练任务载荷失败: %w", unmarshalErr) - } - month, parseErr := parseArchiveMonth(payload.ArchiveMonth) - if parseErr != nil { - return parseErr - } - result, err = h.service.ValidateMonth(ctx, month) - } - fields := []zap.Field{ - zap.Bool("cleanup_enabled", false), zap.String("archive_month", result.Month), - zap.Int64("audit_event_count", result.EventCount), zap.Int64("event_resource_count", result.ResourceCount), - zap.Int64("integration_log_count", result.IntegrationCount), zap.Int64("estimated_cleanup_batches", result.EstimatedBatches), - zap.Duration("validation_duration", result.Duration), zap.Int("manifest_count", len(result.ManifestKeys)), - } - if err != nil { - fields = append(fields, zap.String("severity", "critical"), zap.Error(err)) - h.logger.Error("审计日志月度只读演练失败，物理清理保持关闭", fields...) - return err - } - h.logger.Info("审计日志月度只读演练通过，物理清理保持关闭", fields...) return nil } diff --git a/internal/task/integration_archive.go b/internal/task/integration_archive.go index 6ccd9ad..298ff40 100644 --- a/internal/task/integration_archive.go +++ b/internal/task/integration_archive.go @@ -18,12 +18,7 @@ type IntegrationDailyArchivePayload struct { ArchiveDate string `json:"archive_date"` } -// IntegrationMonthlyFinalizePayload 是人工月度复核时可选的任务载荷。 -type IntegrationMonthlyFinalizePayload struct { - ArchiveMonth string `json:"archive_month"` -} - -// IntegrationArchiveHandler 处理 Integration Log 每日归档与月度最终复核。 +// IntegrationArchiveHandler 处理 Integration Log 每日归档。 type IntegrationArchiveHandler struct { service *auditarchive.Service logger *zap.Logger @@ -61,33 +56,6 @@ func (h *IntegrationArchiveHandler) HandleDaily(ctx context.Context, task *asynq return nil } -// HandleMonthlyFinalize 执行上一个完整自然月复核，或按任务载荷复核指定月份。 -func (h *IntegrationArchiveHandler) HandleMonthlyFinalize(ctx context.Context, task *asynq.Task) error { - if h.service == nil { - return fmt.Errorf("Integration Log 归档服务未配置") - } - var err error - if len(task.Payload()) == 0 { - err = h.service.FinalizePreviousIntegrationMonth(ctx) - } else { - var payload IntegrationMonthlyFinalizePayload - if unmarshalErr := sonic.Unmarshal(task.Payload(), &payload); unmarshalErr != nil { - return fmt.Errorf("解析 Integration Log 月度复核任务载荷失败: %w", unmarshalErr) - } - month, parseErr := parseArchiveMonth(payload.ArchiveMonth) - if parseErr != nil { - return parseErr - } - err = h.service.FinalizeIntegrationMonth(ctx, month) - } - if err != nil { - h.logger.Error("Integration Log 月度最终版本复核失败，后续清理必须阻止", zap.Error(err)) - return err - } - h.logger.Info("Integration Log 月度最终版本复核完成") - return nil -} - func parseArchiveDate(value string) (time.Time, error) { location, err := time.LoadLocation(constants.AuditArchiveTimezone) if err != nil { @@ -99,15 +67,3 @@ func parseArchiveDate(value string) (time.Time, error) { } return date, nil } - -func parseArchiveMonth(value string) (time.Time, error) { - location, err := time.LoadLocation(constants.AuditArchiveTimezone) - if err != nil { - return time.Time{}, fmt.Errorf("加载 Integration Log 归档时区失败: %w", err) - } - month, err := time.ParseInLocation("2006-01", value, location) - if err != nil { - return time.Time{}, fmt.Errorf("解析 Integration Log 归档月份失败: %w", err) - } - return month, nil -} diff --git a/openspec/changes/archive/2026-08-20-daily-audit-log-retention/.openspec.yaml b/openspec/changes/archive/2026-08-20-daily-audit-log-retention/.openspec.yaml new file mode 100644 index 0000000..f774115 --- /dev/null +++ b/openspec/changes/archive/2026-08-20-daily-audit-log-retention/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-20 diff --git a/openspec/changes/archive/2026-08-20-daily-audit-log-retention/design.md b/openspec/changes/archive/2026-08-20-daily-audit-log-retention/design.md new file mode 100644 index 0000000..4485908 --- /dev/null +++ b/openspec/changes/archive/2026-08-20-daily-audit-log-retention/design.md @@ -0,0 +1,68 @@ +## Context + +现有归档已按日生成 `tb_log_archive_run`，但 Integration Log 仅在月初终结整月 revision，物理删除任务也只接受完整自然月。`tb_log_archive_run` 已按来源和归档日保存对象键、manifest、哈希、统计数和 `cleanup_started_at` / `cleaned_at`，可作为按日留存的断点账本。审计调查的在线边界当前通过每个来源的最大已清理范围计算；若出现清理日期空洞会错误隐藏仍在线的数据。 + +## Goals / Non-Goals + +**Goals:** + +- 每天将昨天及更早连续积压的归档日推进到可恢复、可校验、已物理清理的状态。 +- 在任一日期归档、终结、对象校验或删除失败时保留在线数据，并使后续运行可安全续跑。 +- 保持归档对象和 manifest 不删除，保持现有配置开关作为物理删除总闸门。 +- 让在线查询边界只反映连续完成的日期。 + +**Non-Goals:** + +- 不降低卡观测、轮询、Gateway 或 Integration Log 的写入频率。 +- 不直接对生产表执行 SQL 删除，不绕过归档验证清理历史数据。 +- 不改变归档 JSONL 格式、对象存储提供商或调查 API 路由。 + +## Decisions + +### 以“归档日对”为最小留存单元 + +日留存以同一 Asia/Shanghai 日期的 Audit 与 Integration 两条归档账本为一个逻辑单元。任务处理上限固定为昨天，先确保两类归档存在；随后为 Integration 生成最终 revision，确认该日不存在 `pending` 记录，并复用现有 manifest、对象 metadata、哈希和在线计数复核。 + +只有两个来源均通过验证后，才开始删除该日在线数据，顺序为 Audit Resource、Audit Event、Integration Log。删除仍按既有 1000 行批次进行；每个来源独立落 `cleanup_started_at` 和 `cleaned_at`，从而在进程中断后依据剩余计数续跑。 + +不采用“定时归档完成即直接 DELETE”的方案：对象上传成功不等于可恢复，且 Integration Log 的创建日可能仍有待完成记录。 + +### 按日期从早到晚推进，不跨越失败日 + +任务枚举昨天及更早仍未完成的归档日，从最早日期开始。某日失败、缺少归档或存在 pending Integration Log 时停止，不处理更晚日期。这样在线数据的已归档边界永远是连续前缀，历史积压在补齐归档后会由同一任务自动追赶。 + +不采用并行或跳过失败日期的方案：虽然可更快释放部分容量，但会产生在线数据日期空洞；当前调查响应只表达单一 `archived_before` 边界，不能正确描述空洞。 + +### 将 Integration 最终版由月度改为逐日形成 + +保留每日普通归档任务作为预归档；日留存任务在准备删除某日数据时对该日期执行最终归档。最终归档仍要求该日没有 pending Integration Log；否则不删除并等待下次运行。原月度最终复核和月度留存调度移除，避免与日留存争夺同一账本和产生不同的清理范围。 + +不采用固定延迟天数：用户要求昨天及更早数据在可恢复后尽快清理；pending 校验是每日期间的实际安全门禁。 + +### 查询边界按连续清理前缀计算 + +留存边界查询不再取任意来源的 `MAX(range_end)`。它必须从归档账本确认连续已清理日期，且多来源时间线使用两来源共同完成的连续边界；遇到未清理日期时停止。这样部分删除、失败重试或历史补档都不会把仍在线的旧日期标记为已归档。 + +### 使用独立文件记录留存结果与失败 + +新增独立的 Lumberjack/Zap 留存日志配置，默认路径为 Worker 工作目录下的 `logs/audit-retention.log`。日留存和日归档只向该日志写入日期推进、成功数量与耗时，或带日期、来源、失败分类的安全错误摘要；运营者无需在高噪声 `app.log` 中筛选。 + +不把诊断详情写入 `tb_log_archive_run.error_summary`。账本仍只保存归档状态、对象引用和清理断点；归档或留存失败的具体错误以独立文件为准。保留日志轮转和压缩，避免长期错误积压占满 Worker 磁盘。 + +不采用为每个日期新建单独文件的方案：一个按日轮转的专用流已经能按日期字段筛选，并避免文件数量随积压日期增长。 + +## Risks / Trade-offs + +- [某日大量 pending 长期不终结，阻塞后续日期清理] → 记录日期、数量和原因；人工修复业务状态后由日任务续跑，不删除未完成数据。 +- [单日数百万记录清理超过任务超时] → 维持小批次和来源级断点，任务重试时基于剩余记录续跑；按实测调整任务超时，不增大单事务范围。 +- [归档或删除期间进程重启] → 账本的运行租约和清理标记保证归档可重做、删除可继续，且每次删除前重新验证剩余范围。 +- [历史日期未建归档账本] → 日任务在最早缺失日期停止；先通过受控补归档形成完整连续账本，再允许自动物理清理。 +- [日清理后数据库文件空间不立即下降] → 删除仅回收可复用空间；由维护者根据 PostgreSQL 运行状态决定后续 VACUUM 策略，不在 Worker 内执行高风险表重写。 +- [独立留存日志写满磁盘或轮转配置错误] → 使用已有 Lumberjack 轮转、压缩和目录初始化机制；上线前核对 `logs/audit-retention.log` 的生成和轮转。 + +## Migration Plan + +1. 发布包含日归档最终版、日留存和连续边界计算的 Worker/API 二进制，保持物理清理开关关闭。 +2. 维护者核对历史归档账本的连续性，按日期补齐缺失归档，并以只读演练确认对象、manifest 和在线计数。 +3. 维护者在低峰期启用 `JUNHONG_WORKER_AUDIT_RETENTION_CLEANUP_ENABLED=true` 并重启调度 Worker；观察 `/opt/junhong_cmp/worker/logs/audit-retention.log`、账本断点和表大小。 +4. 出现异常时关闭开关并重启调度 Worker；已删除数据通过已验证的对象存储归档恢复，未删除日期保留在线数据。 diff --git a/openspec/changes/archive/2026-08-20-daily-audit-log-retention/proposal.md b/openspec/changes/archive/2026-08-20-daily-audit-log-retention/proposal.md new file mode 100644 index 0000000..408ae0f --- /dev/null +++ b/openspec/changes/archive/2026-08-20-daily-audit-log-retention/proposal.md @@ -0,0 +1,28 @@ +## Why + +`tb_audit_event`、`tb_audit_event_resource` 与 `tb_integration_log` 每日持续写入百万级记录；现有物理留存任务仅在月初清理上一个完整自然月，在线表、索引和备份在月内持续膨胀。生产需要在确认对象存储归档完整且可恢复后，按日清理在线日志数据。 + +## What Changes + +- 将统一审计事件、审计资源快照和 Integration Log 的物理留存单位由完整自然月改为完整自然日。 +- 每次日留存任务处理昨天及更早仍未完成清理的归档日；仅在对应 Audit 与 Integration 归档均成功、完整、可读取且满足最终版要求时，分批物理删除在线表中该日的数据。 +- 将清理断点和查询留存边界改为按归档日表达，支持失败后从未完成日期续跑，且不删除未通过校验的日期。 +- 保留对象存储归档和现有物理清理开关；不降低卡观测、轮询或外部交互日志的写入频率。 + +## Capabilities + +### New Capabilities + +无。 + +### Modified Capabilities + +- `operations-audit`: 审计在线数据在逐日归档校验通过后按归档日物理留存，并向调查查询提供逐日清理边界。 +- `external-integration`: 外部交互日志在逐日最终归档校验通过后按归档日物理留存。 + +## Impact + +- Worker 定时任务、审计归档服务、月度留存处理器和留存边界查询。 +- `tb_log_archive_run` 的清理断点语义，以及 `tb_audit_event`、`tb_audit_event_resource`、`tb_integration_log` 的物理删除范围。 +- 对象存储归档读取、manifest 与哈希校验。 +- 生产需更新 Worker 二进制、`JUNHONG_WORKER_AUDIT_RETENTION_CLEANUP_ENABLED` 及独立留存日志配置；日志写入 Worker 工作目录下的 `logs/audit-retention.log`，不新增外部 API，不直接执行生产数据删除或迁移。 diff --git a/openspec/changes/archive/2026-08-20-daily-audit-log-retention/specs/external-integration/spec.md b/openspec/changes/archive/2026-08-20-daily-audit-log-retention/specs/external-integration/spec.md new file mode 100644 index 0000000..43f730b --- /dev/null +++ b/openspec/changes/archive/2026-08-20-daily-audit-log-retention/specs/external-integration/spec.md @@ -0,0 +1,19 @@ +## ADDED Requirements + +### Requirement: 外部交互日志逐日物理留存 +系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位物理留存 `tb_integration_log`。删除某日在线外部交互日志前，系统 MUST 为该日生成最终归档版本，确认不存在待完成记录，并验证归档对象、清单、日期范围、记录数量和校验摘要可作为恢复凭证。每次执行 SHALL 从最早尚未完成清理的归档日连续处理至昨天；任一日期不能形成或验证最终归档时，系统 MUST 不删除该日及更晚日期的在线外部交互日志。 + +#### Scenario: 最终归档通过后清理在线外部交互日志 +- **GIVEN** 某已结束自然日的外部交互日志均已进入终态，且该日最终归档对象和清单校验成功 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统分批删除该日的在线外部交互日志并将该归档日标记为已清理，归档对象保持可用 + +#### Scenario: 存在待完成外部交互日志 +- **GIVEN** 某归档日仍存在待完成的外部交互日志 +- **WHEN** 日留存任务尝试处理该日期 +- **THEN** 系统不删除该日及更晚日期的在线外部交互日志，并在独立日留存日志中记录该日期未形成最终归档的原因 + +#### Scenario: 历史积压按日期连续补清 +- **GIVEN** 存在多个昨天及更早日期尚未完成物理清理 +- **WHEN** 日留存任务执行且这些日期依次通过最终归档和完整性校验 +- **THEN** 系统按日期从早到晚清理全部连续合格日期 diff --git a/openspec/changes/archive/2026-08-20-daily-audit-log-retention/specs/operations-audit/spec.md b/openspec/changes/archive/2026-08-20-daily-audit-log-retention/specs/operations-audit/spec.md new file mode 100644 index 0000000..8d5ddd2 --- /dev/null +++ b/openspec/changes/archive/2026-08-20-daily-audit-log-retention/specs/operations-audit/spec.md @@ -0,0 +1,41 @@ +## ADDED Requirements + +### Requirement: 审计在线数据逐日物理留存 +系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位处理统一审计事件及其资源快照的在线留存。每次留存执行 SHALL 从最早尚未完成清理的归档日开始，连续处理至昨天；任一归档日未通过完整性校验时，系统 MUST 保留该日及其后续日期的在线审计数据，且不得将它们标记为已清理。系统 MUST 先删除该日的审计资源快照，再删除该日的审计事件，并保留对象存储归档对象及清单作为恢复凭证。 + +#### Scenario: 已验证日期完成物理清理 +- **GIVEN** 某已结束自然日的审计归档成功，归档对象和清单可读取且与在线记录范围、数量和校验摘要一致 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统分批删除该日在线审计资源快照和审计事件，并将该归档日标记为已清理 + +#### Scenario: 归档校验失败阻断连续清理 +- **GIVEN** 某尚未清理的归档日缺少归档、归档校验失败或清单与在线数据不一致 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统不删除该日或更晚日期的在线审计数据，并将失败信息写入独立日留存日志 + +#### Scenario: 失败后从未完成日期续跑 +- **GIVEN** 日留存任务在某个归档日失败，较早日期已经标记为已清理 +- **WHEN** 后续日留存任务再次执行 +- **THEN** 系统从最早未完成清理的归档日继续处理，不重复删除已完成日期的数据 + +### Requirement: 日留存独立运行日志 +系统 SHALL 将日留存的日期推进、清理成功和阻断失败写入独立轮转日志。该日志 MUST 位于 Worker 工作目录下配置的 `logs/audit-retention.log`，每条失败记录 MUST 包含归档日期、数据来源、失败分类和安全错误摘要。日留存和日归档路径 MUST NOT 将失败详情写入 `tb_log_archive_run.error_summary`。 + +#### Scenario: 留存日期被阻断 +- **GIVEN** 日留存处理某个归档日时发生缺失账本、pending 记录、归档校验或删除错误 +- **WHEN** 系统停止该日期推进 +- **THEN** `logs/audit-retention.log` 记录该日期、来源、失败分类和错误摘要 +- **AND** 对应账本记录的 `error_summary` 不写入该失败详情 + +#### Scenario: 留存日期清理成功 +- **GIVEN** 某归档日通过全部校验并完成在线数据物理删除 +- **WHEN** 系统完成该日期处理 +- **THEN** `logs/audit-retention.log` 记录归档日期、各来源清理数量和耗时 + +### Requirement: 审计调查留存边界连续 +系统 SHALL 仅将连续完成物理清理的审计归档日期间公开为已归档边界。在线审计调查接口 MUST 继续允许查询任何尚未物理清理的较早日期，且不得因某个较晚日期已归档而将仍在线的日期错误标记为已归档。 + +#### Scenario: 清理链存在未完成日期 +- **GIVEN** 某较早归档日尚未完成清理 +- **WHEN** 后续日期的归档已成功生成 +- **THEN** 审计调查接口仍将较早未清理日期视为在线可查询数据 diff --git a/openspec/changes/archive/2026-08-20-daily-audit-log-retention/tasks.md b/openspec/changes/archive/2026-08-20-daily-audit-log-retention/tasks.md new file mode 100644 index 0000000..ce55587 --- /dev/null +++ b/openspec/changes/archive/2026-08-20-daily-audit-log-retention/tasks.md @@ -0,0 +1,19 @@ +## 1. 按日归档与留存编排 + +- [x] 1.1 将 Integration Log 最终归档能力收敛为指定已结束自然日；保留 pending 记录时不产生最终 revision，并通过独立留存日志输出原因而不更新 `error_summary`。 +- [x] 1.2 将现有月度留存服务改为逐日处理：枚举昨天及更早未完成日期，按日期从早到晚确保双来源归档、最终归档和完整性校验。 +- [x] 1.3 复用现有分批删除和来源级清理断点，按 Audit Resource、Audit Event、Integration Log 的顺序物理删除单日在线数据；中断或失败时可从剩余记录续跑。 +- [x] 1.4 遇到缺失账本、归档失败、对象或 manifest 校验失败、数量不一致或 pending Integration Log 时停止日期推进；不清理该日及后续日期，且不向 `tb_log_archive_run.error_summary` 写入失败详情。 + +## 2. Worker 调度与查询边界 + +- [x] 2.1 为 Worker 新增独立留存日志配置和目录初始化，默认输出 `logs/audit-retention.log`，复用现有轮转与压缩能力，并在退出时刷新日志。 +- [x] 2.2 将 Worker 留存调度由月初任务改为每日任务，调整任务类型、唯一窗口、超时、处理器日志及归档注册，避免月度任务与日留存并发处理同一账本。 +- [x] 2.3 保留 `JUNHONG_WORKER_AUDIT_RETENTION_CLEANUP_ENABLED` 作为物理删除总闸门；关闭时执行逐日只读校验，不写清理断点、不删除在线数据。 +- [x] 2.4 改造审计和 Integration Log 在线查询留存边界，按连续已清理日期计算，并使混合时间线不将仍在线的空洞日期误报为已归档。 +- [x] 2.5 更新留存模拟 CLI 或等价可执行验证入口，以逐日范围验证归档、最终版、对象、manifest、数据库数量、清理断点及独立日志输出。 + +## 3. 运行说明与验证 + +- [x] 3.1 更新生产运行说明，明确日留存启用前的历史归档补齐、只读演练、开关启用、`/opt/junhong_cmp/worker/logs/audit-retention.log` 观察、账本断点、异常停用和归档恢复步骤。 +- [x] 3.2 执行 `gofmt`、`go build ./cmd/api ./cmd/worker`、`go run cmd/gendocs/main.go`、`openspec validate daily-audit-log-retention --strict` 与日留存模拟验证；不新增或恢复自动化测试。（隔离库模拟待维护者在具备显式 `JUNHONG_*` 配置的环境执行确认） diff --git a/openspec/specs/external-integration/spec.md b/openspec/specs/external-integration/spec.md index 8ac9e7d..c057f1e 100644 --- a/openspec/specs/external-integration/spec.md +++ b/openspec/specs/external-integration/spec.md @@ -101,6 +101,25 @@ - **WHEN** 富友主扫统一下单返回失败或请求结果未知 - **THEN** 系统返回项目稳定错误且已接入外部交互日志的调用记录脱敏结果 +### Requirement: 外部交互日志逐日物理留存 + +系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位物理留存 `tb_integration_log`。删除某日在线外部交互日志前，系统 MUST 为该日生成最终归档版本，确认不存在待完成记录，并验证归档对象、清单、日期范围、记录数量和校验摘要可作为恢复凭证。每次执行 SHALL 从最早尚未完成清理的归档日连续处理至昨天；任一日期不能形成或验证最终归档时，系统 MUST 不删除该日及更晚日期的在线外部交互日志。 + +#### Scenario: 最终归档通过后清理在线外部交互日志 +- **GIVEN** 某已结束自然日的外部交互日志均已进入终态，且该日最终归档对象和清单校验成功 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统分批删除该日的在线外部交互日志并将该归档日标记为已清理，归档对象保持可用 + +#### Scenario: 存在待完成外部交互日志 +- **GIVEN** 某归档日仍存在待完成的外部交互日志 +- **WHEN** 日留存任务尝试处理该日期 +- **THEN** 系统不删除该日及更晚日期的在线外部交互日志，并在独立日留存日志中记录该日期未形成最终归档的原因 + +#### Scenario: 历史积压按日期连续补清 +- **GIVEN** 存在多个昨天及更早日期尚未完成物理清理 +- **WHEN** 日留存任务执行且这些日期依次通过最终归档和完整性校验 +- **THEN** 系统按日期从早到晚清理全部连续合格日期 + ## 可达操作索引 本节只用于入口导航，不是行为 Requirement；业务义务以上述 Requirements 为准。 diff --git a/openspec/specs/operations-audit/spec.md b/openspec/specs/operations-audit/spec.md index be8763b..dd53ece 100644 --- a/openspec/specs/operations-audit/spec.md +++ b/openspec/specs/operations-audit/spec.md @@ -46,6 +46,49 @@ - **WHEN** 该操作的审计事件构造、校验或持久化失败 - **THEN** 系统提交或返回该业务操作原本的结果，并以请求关联标识、动作编码和资源标识记录审计失败 +### Requirement: 审计在线数据逐日物理留存 + +系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位处理统一审计事件及其资源快照的在线留存。每次留存执行 SHALL 从最早尚未完成清理的归档日开始，连续处理至昨天；任一归档日未通过完整性校验时，系统 MUST 保留该日及其后续日期的在线审计数据，且不得将它们标记为已清理。系统 MUST 先删除该日的审计资源快照，再删除该日的审计事件，并保留对象存储归档对象及清单作为恢复凭证。 + +#### Scenario: 已验证日期完成物理清理 +- **GIVEN** 某已结束自然日的审计归档成功，归档对象和清单可读取且与在线记录范围、数量和校验摘要一致 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统分批删除该日在线审计资源快照和审计事件，并将该归档日标记为已清理 + +#### Scenario: 归档校验失败阻断连续清理 +- **GIVEN** 某尚未清理的归档日缺少归档、归档校验失败或清单与在线数据不一致 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统不删除该日或更晚日期的在线审计数据，并将失败信息写入独立日留存日志 + +#### Scenario: 失败后从未完成日期续跑 +- **GIVEN** 日留存任务在某个归档日失败，较早日期已经标记为已清理 +- **WHEN** 后续日留存任务再次执行 +- **THEN** 系统从最早未完成清理的归档日继续处理，不重复删除已完成日期的数据 + +### Requirement: 日留存独立运行日志 + +系统 SHALL 将日留存的日期推进、清理成功和阻断失败写入独立轮转日志。该日志 MUST 位于 Worker 工作目录下配置的 `logs/audit-retention.log`，每条失败记录 MUST 包含归档日期、数据来源、失败分类和安全错误摘要。日留存和日归档路径 MUST NOT 将失败详情写入 `tb_log_archive_run.error_summary`。 + +#### Scenario: 留存日期被阻断 +- **GIVEN** 日留存处理某个归档日时发生缺失账本、pending 记录、归档校验或删除错误 +- **WHEN** 系统停止该日期推进 +- **THEN** `logs/audit-retention.log` 记录该日期、来源、失败分类和错误摘要 +- **AND** 对应账本记录的 `error_summary` 不写入该失败详情 + +#### Scenario: 留存日期清理成功 +- **GIVEN** 某归档日通过全部校验并完成在线数据物理删除 +- **WHEN** 系统完成该日期处理 +- **THEN** `logs/audit-retention.log` 记录归档日期、各来源清理数量和耗时 + +### Requirement: 审计调查留存边界连续 + +系统 SHALL 仅将连续完成物理清理的审计归档日期间公开为已归档边界。在线审计调查接口 MUST 继续允许查询任何尚未物理清理的较早日期，且不得因某个较晚日期已归档而将仍在线的日期错误标记为已归档。 + +#### Scenario: 清理链存在未完成日期 +- **GIVEN** 某较早归档日尚未完成清理 +- **WHEN** 后续日期的归档已成功生成 +- **THEN** 审计调查接口仍将较早未清理日期视为在线可查询数据 + ## 可达操作索引 本节只用于入口导航，不是行为 Requirement；业务义务以上述 Requirements 为准。 diff --git a/pkg/bootstrap/directories.go b/pkg/bootstrap/directories.go index 9d411a2..d73aa06 100644 --- a/pkg/bootstrap/directories.go +++ b/pkg/bootstrap/directories.go @@ -10,10 +10,11 @@ import ( ) type DirectoryResult struct { - TempDir string - AppLogDir string - AccessLogDir string - Fallbacks []string + TempDir string + AppLogDir string + AccessLogDir string + RetentionLogDir string + Fallbacks []string } func EnsureDirectories(cfg *config.Config, logger *zap.Logger) (*DirectoryResult, error) { @@ -27,6 +28,7 @@ func EnsureDirectories(cfg *config.Config, logger *zap.Logger) (*DirectoryResult {cfg.Storage.TempDir, "storage.temp_dir", &result.TempDir}, {filepath.Dir(cfg.Logging.AppLog.Filename), "logging.app_log.filename", &result.AppLogDir}, {filepath.Dir(cfg.Logging.AccessLog.Filename), "logging.access_log.filename", &result.AccessLogDir}, + {filepath.Dir(cfg.Logging.RetentionLog.Filename), "logging.retention_log.filename", &result.RetentionLogDir}, } for _, dir := range directories { diff --git a/pkg/config/config.go b/pkg/config/config.go index c4d98f6..0526678 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -79,10 +79,11 @@ type QueueConfig struct { // LoggingConfig 日志配置 type LoggingConfig struct { - Level string `mapstructure:"level"` // debug, info, warn, error - Development bool `mapstructure:"development"` // 启用开发模式（美化输出） - AppLog LogRotationConfig `mapstructure:"app_log"` // 应用日志配置 - AccessLog LogRotationConfig `mapstructure:"access_log"` // HTTP 访问日志配置 + Level string `mapstructure:"level"` // debug, info, warn, error + Development bool `mapstructure:"development"` // 启用开发模式（美化输出） + AppLog LogRotationConfig `mapstructure:"app_log"` // 应用日志配置 + AccessLog LogRotationConfig `mapstructure:"access_log"` // HTTP 访问日志配置 + RetentionLog LogRotationConfig `mapstructure:"retention_log"` // Worker 日留存日志配置 } // LogRotationConfig Lumberjack 日志轮转配置 @@ -168,7 +169,7 @@ type GatewayConfig struct { type WorkerConfig struct { Role string `mapstructure:"role"` // Worker 运行角色：all、leader、consumer InstanceName string `mapstructure:"instance_name"` // Worker 实例名称，用于多实例日志区分 - AuditRetentionCleanupEnabled bool `mapstructure:"audit_retention_cleanup_enabled"` // 是否启用审计日志月度物理清理 + AuditRetentionCleanupEnabled bool `mapstructure:"audit_retention_cleanup_enabled"` // 是否启用审计日志物理清理 } // ApprovalConfig 审批新旧入口切换配置。 @@ -291,12 +292,18 @@ func (c *Config) Validate() error { if c.Logging.AccessLog.Filename == "" { return fmt.Errorf("invalid configuration: logging.access_log.filename: must be non-empty valid file path") } + if c.Logging.RetentionLog.Filename == "" { + return fmt.Errorf("invalid configuration: logging.retention_log.filename: must be non-empty valid file path") + } if c.Logging.AppLog.MaxSize < 1 || c.Logging.AppLog.MaxSize > 1000 { return fmt.Errorf("invalid configuration: logging.app_log.max_size: size out of range (current value: %d, expected: 1-1000 MB)", c.Logging.AppLog.MaxSize) } if c.Logging.AccessLog.MaxSize < 1 || c.Logging.AccessLog.MaxSize > 1000 { return fmt.Errorf("invalid configuration: logging.access_log.max_size: size out of range (current value: %d, expected: 1-1000 MB)", c.Logging.AccessLog.MaxSize) } + if c.Logging.RetentionLog.MaxSize < 1 || c.Logging.RetentionLog.MaxSize > 1000 { + return fmt.Errorf("invalid configuration: logging.retention_log.max_size: size out of range (current value: %d, expected: 1-1000 MB)", c.Logging.RetentionLog.MaxSize) + } // 中间件验证 if c.Middleware.RateLimiter.Max <= 0 { diff --git a/pkg/config/defaults/config.yaml b/pkg/config/defaults/config.yaml index 2958809..1fc0b30 100644 --- a/pkg/config/defaults/config.yaml +++ b/pkg/config/defaults/config.yaml @@ -65,6 +65,12 @@ logging: max_backups: 3 max_age: 7 compress: true + retention_log: + filename: "logs/audit-retention.log" + max_size: 100 + max_backups: 3 + max_age: 7 + compress: true # 任务队列配置 queue: @@ -137,7 +143,7 @@ polling_auto_trigger: worker: role: "all" instance_name: "" - # 完整自然月灰度验收通过前必须保持关闭 + # 日留存只读演练验收通过前必须保持关闭 audit_retention_cleanup_enabled: false # 审批新旧入口切换配置 diff --git a/pkg/config/loader.go b/pkg/config/loader.go index f1eb3a3..9f81f4d 100644 --- a/pkg/config/loader.go +++ b/pkg/config/loader.go @@ -94,6 +94,11 @@ func bindEnvVariables(v *viper.Viper) { "logging.access_log.max_backups", "logging.access_log.max_age", "logging.access_log.compress", + "logging.retention_log.filename", + "logging.retention_log.max_size", + "logging.retention_log.max_backups", + "logging.retention_log.max_age", + "logging.retention_log.compress", "queue.concurrency", "queue.retry_max", "queue.timeout", diff --git a/pkg/constants/audit_archive.go b/pkg/constants/audit_archive.go index 8750ba1..2e64667 100644 --- a/pkg/constants/audit_archive.go +++ b/pkg/constants/audit_archive.go @@ -5,10 +5,8 @@ const ( TaskTypeAuditDailyArchive = "audit:daily:archive" // TaskTypeIntegrationDailyArchive 表示 Integration Log 每日冷归档任务。 TaskTypeIntegrationDailyArchive = "integration:daily:archive" - // TaskTypeIntegrationMonthlyFinalize 表示 Integration Log 月度最终版本复核任务。 - TaskTypeIntegrationMonthlyFinalize = "integration:monthly:finalize" - // TaskTypeAuditMonthlyRetention 表示审计日志月度物理清理任务。 - TaskTypeAuditMonthlyRetention = "audit:monthly:retention" + // TaskTypeAuditDailyRetention 表示审计日志日留存任务。 + TaskTypeAuditDailyRetention = "audit:daily:retention" // AuditArchiveSource 表示 Audit Event 与 Event Resource 归档数据源。 AuditArchiveSource = "audit" @@ -32,6 +30,6 @@ const ( // ArchiveStatusFailed 表示归档任务执行失败并等待重试。 ArchiveStatusFailed = "failed" - // AuditRetentionDeleteBatchSize 表示月度物理清理单批删除上限。 + // AuditRetentionDeleteBatchSize 表示日物理清理单批删除上限。 AuditRetentionDeleteBatchSize = 1000 ) diff --git a/pkg/constants/constants.go b/pkg/constants/constants.go index b863b01..b8ba3e0 100644 --- a/pkg/constants/constants.go +++ b/pkg/constants/constants.go @@ -298,7 +298,7 @@ func QueueForTaskType(taskType string) string { return QueueDataCleanup case TaskTypeDailyTrafficFlush: return QueueDailyTrafficFlush - case TaskTypeAuditDailyArchive, TaskTypeIntegrationDailyArchive, TaskTypeIntegrationMonthlyFinalize, TaskTypeAuditMonthlyRetention: + case TaskTypeAuditDailyArchive, TaskTypeIntegrationDailyArchive, TaskTypeAuditDailyRetention: return QueueDataCleanup case TaskTypeOutboxDeliver: return QueueOutboxDeliver diff --git a/pkg/logger/retention.go b/pkg/logger/retention.go new file mode 100644 index 0000000..a6f68c9 --- /dev/null +++ b/pkg/logger/retention.go @@ -0,0 +1,17 @@ +package logger + +import ( + "go.uber.org/zap" + "go.uber.org/zap/zapcore" +) + +// NewRetentionLogger 创建仅供 Worker 日留存使用的独立轮转日志。 +func NewRetentionLogger(level string, config LogRotationConfig) (*zap.Logger, func() error) { + encoder := zapcore.NewJSONEncoder(zapcore.EncoderConfig{ + TimeKey: "time", LevelKey: "level", MessageKey: "msg", CallerKey: "caller", + EncodeTime: zapcore.ISO8601TimeEncoder, EncodeLevel: zapcore.CapitalLevelEncoder, + EncodeCaller: zapcore.ShortCallerEncoder, + }) + logger := zap.New(zapcore.NewCore(encoder, zapcore.AddSync(newLumberjackLogger(config)), parseLevel(level)), zap.AddCaller()) + return logger, logger.Sync +}