break
c64f3d8b80
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 8m31s
581 lines
22 KiB
Go
581 lines
22 KiB
Go
package auditarchive
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"fmt"
|
||
"io"
|
||
"os"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/bytedance/sonic"
|
||
"gorm.io/gorm"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
)
|
||
|
||
const maxManifestBytes = 1024 * 1024
|
||
|
||
// RetentionAudit 描述月度物理清理的统一审计事实。
|
||
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
|
||
}
|
||
|
||
// RetentionResult 是月度留存清理的结构化执行结果。
|
||
type RetentionResult struct {
|
||
Month string
|
||
EventCount int64
|
||
ResourceCount int64
|
||
IntegrationCount int64
|
||
EstimatedBatches int64
|
||
ManifestKeys []string
|
||
Duration time.Duration
|
||
}
|
||
|
||
type retentionRuns struct {
|
||
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) {
|
||
if s.db == nil || s.store == nil {
|
||
return result, fmt.Errorf("日志留存演练数据库或对象存储未配置")
|
||
}
|
||
start, end, err := s.retentionMonthRange(month)
|
||
if err != nil {
|
||
return result, err
|
||
}
|
||
startedAt := time.Now()
|
||
result.Month = start.Format("2006-01")
|
||
runs, err := s.loadRetentionRuns(ctx, start, end)
|
||
if err != nil {
|
||
return result, err
|
||
}
|
||
if err := s.validateRetentionRuns(ctx, start, end, runs); err != nil {
|
||
return result, err
|
||
}
|
||
summarizeRetentionRuns(runs, &result)
|
||
result.EstimatedBatches = estimatedRetentionBatches(result)
|
||
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) {
|
||
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)
|
||
}
|
||
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)}
|
||
for index := range rows {
|
||
run := &rows[index]
|
||
switch run.Source {
|
||
case constants.AuditArchiveSource:
|
||
runs.audit = append(runs.audit, run)
|
||
case constants.IntegrationArchiveSource:
|
||
runs.integration = append(runs.integration, run)
|
||
}
|
||
}
|
||
if len(runs.audit) != days || len(runs.integration) != days {
|
||
return retentionRuns{}, fmt.Errorf("月度 Audit 或 Integration 归档账本不完整")
|
||
}
|
||
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
|
||
}
|
||
if err := s.validateAuditManifest(ctx, run); err != nil {
|
||
return err
|
||
}
|
||
events, resources, err := s.databaseCounts(ctx, run.RangeStart, run.RangeEnd)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return validateRemainingCounts(run, events, resources)
|
||
}
|
||
|
||
func (s *Service) validateIntegrationRetentionDay(ctx context.Context, date time.Time, run *model.LogArchiveRun) error {
|
||
if err := validateRunBase(run, date, constants.IntegrationArchiveSchemaVersion, true); err != nil {
|
||
return err
|
||
}
|
||
if err := s.validateIntegrationManifest(ctx, run); err != nil {
|
||
return err
|
||
}
|
||
count, err := s.integrationRecordCount(ctx, run.RangeStart, run.RangeEnd)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if run.CleanedAt != nil {
|
||
if count != 0 {
|
||
return fmt.Errorf("已标记清理完成但数据库仍有 %d 条记录", count)
|
||
}
|
||
return nil
|
||
}
|
||
if run.CleanupStartedAt != nil {
|
||
if count > run.RecordCount {
|
||
return fmt.Errorf("续跑窗口记录数超过最终归档数量")
|
||
}
|
||
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
|
||
}
|
||
|
||
func validateRunBase(run *model.LogArchiveRun, date time.Time, schema string, final bool) error {
|
||
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)) {
|
||
return fmt.Errorf("归档日期或半开时间范围不一致")
|
||
}
|
||
if final && !run.IsFinal {
|
||
return fmt.Errorf("Integration 最终 revision 尚未形成")
|
||
}
|
||
if run.ObjectKey == "" || run.ManifestKey == "" || run.SHA256 == "" {
|
||
return fmt.Errorf("归档对象、manifest 或 SHA-256 缺失")
|
||
}
|
||
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.CleanupStartedAt != nil {
|
||
if events > run.EventCount || resources > run.ResourceCount {
|
||
return fmt.Errorf("续跑窗口数量超过已归档数量")
|
||
}
|
||
return nil
|
||
}
|
||
if events != run.EventCount || resources != run.ResourceCount {
|
||
return fmt.Errorf("数据库事件或资源数量与 manifest 不一致")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *Service) validateAuditManifest(ctx context.Context, run *model.LogArchiveRun) error {
|
||
var manifest Manifest
|
||
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 {
|
||
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 {
|
||
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 {
|
||
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 {
|
||
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 {
|
||
return fmt.Errorf("读取 manifest metadata 失败: %w", err)
|
||
}
|
||
reader, err := s.store.Download(ctx, key)
|
||
if err != nil {
|
||
return fmt.Errorf("下载 manifest 失败: %w", err)
|
||
}
|
||
data, readErr := io.ReadAll(io.LimitReader(reader, maxManifestBytes+1))
|
||
closeErr := reader.Close()
|
||
if readErr != nil {
|
||
return fmt.Errorf("读取 manifest 失败: %w", readErr)
|
||
}
|
||
if closeErr != nil {
|
||
return fmt.Errorf("关闭 manifest 对象失败: %w", closeErr)
|
||
}
|
||
if len(data) > maxManifestBytes || int64(len(data)) != object.Size {
|
||
return fmt.Errorf("manifest 大小非法或不完整")
|
||
}
|
||
if err := sonic.Unmarshal(data, target); err != nil {
|
||
return fmt.Errorf("解析 manifest 失败: %w", err)
|
||
}
|
||
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),
|
||
}
|
||
if run.Source == constants.AuditArchiveSource {
|
||
metadata["event-count"] = strconv.FormatInt(run.EventCount, 10)
|
||
metadata["resource-count"] = strconv.FormatInt(run.ResourceCount, 10)
|
||
} else {
|
||
metadata["record-count"] = strconv.FormatInt(run.RecordCount, 10)
|
||
metadata["final"] = strconv.FormatBool(final)
|
||
}
|
||
if err := s.verifyObject(ctx, run.ObjectKey, run.CompressedBytes, metadata); err != nil {
|
||
return err
|
||
}
|
||
reader, err := s.store.Download(ctx, run.ObjectKey)
|
||
if err != nil {
|
||
return fmt.Errorf("下载归档对象复核 SHA-256 失败: %w", err)
|
||
}
|
||
hasher := sha256.New()
|
||
written, copyErr := io.Copy(hasher, reader)
|
||
closeErr := reader.Close()
|
||
if copyErr != nil {
|
||
return fmt.Errorf("读取归档对象复核 SHA-256 失败: %w", copyErr)
|
||
}
|
||
if closeErr != nil {
|
||
return fmt.Errorf("关闭归档对象失败: %w", closeErr)
|
||
}
|
||
if written != run.CompressedBytes || fmt.Sprintf("%x", hasher.Sum(nil)) != run.SHA256 {
|
||
return fmt.Errorf("归档对象大小或 SHA-256 复核失败")
|
||
}
|
||
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
|
||
}
|
||
|
||
func (s *Service) cleanupAuditMonth(ctx context.Context, start, end time.Time, runs []*model.LogArchiveRun) error {
|
||
if allRunsCleaned(runs) {
|
||
return nil
|
||
}
|
||
if err := s.markCleanupStarted(ctx, constants.AuditArchiveSource, start, end); err != nil {
|
||
return err
|
||
}
|
||
if err := s.deleteAuditResources(ctx, start, end); err != nil {
|
||
return err
|
||
}
|
||
if err := s.deleteAuditEvents(ctx, start, end); err != nil {
|
||
return err
|
||
}
|
||
return s.markCleaned(ctx, constants.AuditArchiveSource, start, end)
|
||
}
|
||
|
||
func (s *Service) cleanupIntegrationMonth(ctx context.Context, start, end time.Time, runs []*model.LogArchiveRun) error {
|
||
if allRunsCleaned(runs) {
|
||
return nil
|
||
}
|
||
if err := s.markCleanupStarted(ctx, constants.IntegrationArchiveSource, start, end); err != nil {
|
||
return err
|
||
}
|
||
for {
|
||
subquery := s.db.Model(&model.IntegrationLog{}).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.IntegrationLog{})
|
||
if deleted.Error != nil {
|
||
return fmt.Errorf("分批物理删除 Integration Log 失败: %w", deleted.Error)
|
||
}
|
||
if deleted.RowsAffected == 0 {
|
||
break
|
||
}
|
||
}
|
||
return s.markCleaned(ctx, constants.IntegrationArchiveSource, start, end)
|
||
}
|
||
|
||
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)
|
||
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)
|
||
}
|
||
if deleted.RowsAffected == 0 {
|
||
return nil
|
||
}
|
||
}
|
||
}
|
||
|
||
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)
|
||
deleted := s.db.WithContext(ctx).Where("id IN (?)", subquery).Delete(&model.AuditEvent{})
|
||
if deleted.Error != nil {
|
||
return fmt.Errorf("分批物理删除 Audit Event 失败: %w", deleted.Error)
|
||
}
|
||
if deleted.RowsAffected == 0 {
|
||
return nil
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *Service) markCleanupStarted(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 NULL",
|
||
source, start.Format(time.DateOnly), end.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 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, "-", "_"),
|
||
}
|
||
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)
|
||
}
|