From fa52a76fe14c513aa9d2da2daa1738c0579a23cb Mon Sep 17 00:00:00 2001 From: samwaf Date: Fri, 26 Jun 2026 11:33:10 +0800 Subject: [PATCH] fix: enforce db_file_size via 5-min sharding check and archive cleanup #838 --- cmd/samwaf/main.go | 1 + enums/task_enum.go | 3 +- go.sum | 1 + service/waf_service/waf_sharedb.go | 5 ++ wafdb/migrations_task.go | 38 +++++++++++++++ waftask/task_clean_archive.go | 74 ++++++++++++++++++++++++++++++ waftask/task_db_sharding.go | 19 ++++++-- 7 files changed, 136 insertions(+), 5 deletions(-) create mode 100644 waftask/task_clean_archive.go diff --git a/cmd/samwaf/main.go b/cmd/samwaf/main.go index f76eb39..07d37b8 100644 --- a/cmd/samwaf/main.go +++ b/cmd/samwaf/main.go @@ -475,6 +475,7 @@ func (m *wafSystenService) run() { globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_RUNTIME_QPS_CLEAN, waftask.TaskLogQpsClean) globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_HOST_QPS_CLEAN, waftask.TaskHostQpsClean) globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_SHARE_DB, waftask.TaskShareDbInfo) + globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_SHARE_DB_CHECK, waftask.TaskShareDbInfo) globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_COUNTER, waftask.TaskCounter) globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_DELAY_INFO, waftask.TaskDelayInfo) globalobj.GWAF_RUNTIME_OBJ_WAF_TaskRegistry.RegisterTask(enums.TASK_LOAD_CONFIG, waftask.TaskLoadSettingCron) diff --git a/enums/task_enum.go b/enums/task_enum.go index 45334d9..dfcacf9 100644 --- a/enums/task_enum.go +++ b/enums/task_enum.go @@ -3,7 +3,8 @@ package enums const ( TASK_RUNTIME_QPS_CLEAN = "task_runtime_qps_clean" //清空运行QPS TASK_HOST_QPS_CLEAN = "task_host_qps_clean" //清空主机QPS - TASK_SHARE_DB = "task_share_db" //分库检测 + TASK_SHARE_DB = "task_share_db" //分库检测(每天兜底) + TASK_SHARE_DB_CHECK = "task_share_db_check" //分库高频检测(每5分钟按大小切库) TASK_COUNTER = "task_counter" //统计 TASK_DELAY_INFO = "task_delay_info" //延迟统计信息 TASK_LOAD_CONFIG = "task_load_config" //获取配置信息 diff --git a/go.sum b/go.sum index 1b27ea9..9070e59 100644 --- a/go.sum +++ b/go.sum @@ -354,6 +354,7 @@ github.com/samwafgo/cap_go_server v1.0.0 h1:1LnITCKtYaw0X3KM4FGT9mh+3RDsyMGHefwX github.com/samwafgo/cap_go_server v1.0.0/go.mod h1:PbMuR/mvf5D2vqj6nZDIFuBhYWCQCq4VnKPQstZS2gc= github.com/samwafgo/go-wxsqlite3 v1.0.1 h1:Jv3IH1zmp2kyFWFRzz6eZGXSWWCGESPneCAgeYqZAlk= github.com/samwafgo/go-wxsqlite3 v1.0.1/go.mod h1:Q72tT7m3Twc1yNqJHmwl3nKR26mgFlguzSnxSAFiLJU= +github.com/samwafgo/go-wxsqlite3 v1.0.2 h1:ipwEiHq2+jL1qAO01BrPVAXiV6Z+vKBinbKF6rGpIyM= github.com/samwafgo/go-wxsqlite3 v1.0.2/go.mod h1:Q72tT7m3Twc1yNqJHmwl3nKR26mgFlguzSnxSAFiLJU= github.com/samwafgo/sqlitedriver v1.0.2 h1:QGQrK+r9z9izEpPAW5MSjhK3vg5H45STLVrAiFT0QJU= github.com/samwafgo/sqlitedriver v1.0.2/go.mod h1:OtH66Yn4WxfnFA+1VhT2iMYdbKJQgZDExqsDy/1Nz5w= diff --git a/service/waf_service/waf_sharedb.go b/service/waf_service/waf_sharedb.go index baa6483..49ab8a9 100644 --- a/service/waf_service/waf_sharedb.go +++ b/service/waf_service/waf_sharedb.go @@ -16,6 +16,11 @@ func (receiver *WafShareDbService) AddApi(shareDb model.ShareDb) error { return nil } +// DeleteById 删除一条分库归档记录（按主键 Id）。归档物理文件由调用方负责删除。 +func (receiver *WafShareDbService) DeleteById(id string) error { + return global.GWAF_LOCAL_DB.Where("id = ?", id).Delete(&model.ShareDb{}).Error +} + func (receiver *WafShareDbService) GetListApi(req request.WafShareDbReq) ([]model.ShareDb, int64, error) { var list []model.ShareDb var total int64 = 0 diff --git a/wafdb/migrations_task.go b/wafdb/migrations_task.go index 1bb4738..3f39f8c 100644 --- a/wafdb/migrations_task.go +++ b/wafdb/migrations_task.go @@ -428,6 +428,44 @@ func RunTaskInitMigrations(db *gorm.DB) error { return tx.Where("task_method = ?", enums.TASK_STATS_DATA_CLEANUP).Delete(&model.Task{}).Error }, }, + // 迁移3: 新增高频分库检测任务（每5分钟按文件大小切库，兼容老用户） + { + ID: "202606260004_add_share_db_check_task", + Migrate: func(tx *gorm.DB) error { + zlog.Info("迁移 202606260004: 新增高频分库检测任务") + + var count int64 + tx.Model(&model.Task{}).Where("task_method = ?", enums.TASK_SHARE_DB_CHECK).Count(&count) + if count > 0 { + zlog.Info("高频分库检测任务已存在，跳过", "task_method", enums.TASK_SHARE_DB_CHECK) + return nil + } + + task := model.Task{ + BaseOrm: baseorm.BaseOrm{ + Id: uuid.GenUUID(), + USER_CODE: global.GWAF_USER_CODE, + Tenant_ID: global.GWAF_TENANT_ID, + CREATE_TIME: customtype.JsonTime(time.Now()), + UPDATE_TIME: customtype.JsonTime(time.Now()), + }, + TaskName: "每5分钟按文件大小检测分库", + TaskUnit: enums.TASK_MIN, + TaskValue: 5, + TaskAt: "", + TaskMethod: enums.TASK_SHARE_DB_CHECK, + } + if err := tx.Create(&task).Error; err != nil { + return fmt.Errorf("创建高频分库检测任务失败: %w", err) + } + zlog.Info("高频分库检测任务创建成功") + return nil + }, + Rollback: func(tx *gorm.DB) error { + zlog.Info("回滚 202606260004: 删除高频分库检测任务") + return tx.Where("task_method = ?", enums.TASK_SHARE_DB_CHECK).Delete(&model.Task{}).Error + }, + }, }) // 执行迁移 diff --git a/waftask/task_clean_archive.go b/waftask/task_clean_archive.go new file mode 100644 index 0000000..cd8ec8a --- /dev/null +++ b/waftask/task_clean_archive.go @@ -0,0 +1,74 @@ +package waftask + +import ( + "SamWaf/common/zlog" + "SamWaf/enums" + "SamWaf/global" + "SamWaf/service/waf_service" + "SamWaf/utils" + "SamWaf/wafdb" + "SamWaf/wafdb/dialect" + "os" + "time" +) + +// CleanExpiredArchiveShard 清理超过保留期(GDATA_DELETE_INTERVAL 天)的归档日志分片文件。 +// +// 背景：高频切库(见 TaskShareDbInfo)后会产生大量 local_log_.db 归档文件， +// 而原有的删历史逻辑只对 live 库做 DELETE，从不清理归档文件，会很快堆满磁盘。 +// 真正的 N 天保留改由本函数按分片 EndTime 删除整个过期归档文件实现。 +// +// 仅 SQLite(文件型驱动)生效；MySQL 的归档是表，由其各自策略处理。 +func CleanExpiredArchiveShard() { + innerLogName := "CleanExpiredArchiveShard" + if !dialect.Get().IsFileBased() { + return + } + + cutoff := time.Now().AddDate(0, 0, -int(global.GDATA_DELETE_INTERVAL)) + currentDir := utils.GetCurrentDir() + + shards, err := waf_service.WafShareDbServiceApp.GetAllShareDbApi() + if err != nil { + zlog.Error(innerLogName, "获取分库列表失败", err) + return + } + + removedFiles := 0 + for _, shard := range shards { + // 只处理日志分片；跳过 live 库本身 + if shard.DbLogicType != "log" { + continue + } + if shard.FileName == "" || shard.FileName == enums.DB_LOG || shard.FileName == "local_log.db" { + continue + } + // 整个分片(截止 EndTime)早于保留期才删除 + if !time.Time(shard.EndTime).Before(cutoff) { + continue + } + + // 删除前先关闭可能已按需打开的连接，避免删正在查询的文件 + wafdb.CloseManualLogDb(shard.FileName) + + dbPath := currentDir + "/data/" + shard.FileName + for _, p := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} { + if err := os.Remove(p); err != nil && !os.IsNotExist(err) { + zlog.Warn(innerLogName, "删除归档文件失败", "file", p, "error", err.Error()) + } + } + removedFiles++ + + // 删除归档元数据记录 + if err := waf_service.WafShareDbServiceApp.DeleteById(shard.Id); err != nil { + zlog.Warn(innerLogName, "删除归档记录失败", "file", shard.FileName, "error", err.Error()) + } else { + zlog.Info(innerLogName, "已清理过期归档分片", "file", shard.FileName, + "end_time", time.Time(shard.EndTime).Format("2006-01-02 15:04:05")) + } + } + + if removedFiles > 0 { + zlog.Info(innerLogName, "归档清理完成", "removed", removedFiles, "cutoff", cutoff.Format("2006-01-02")) + } +} diff --git a/waftask/task_db_sharding.go b/waftask/task_db_sharding.go index 6fafb8e..86ab290 100644 --- a/waftask/task_db_sharding.go +++ b/waftask/task_db_sharding.go @@ -54,10 +54,15 @@ func TaskShareDbInfo() { if dialect.Get().IsFileBased() { fileInfo, err := os.Stat(dbFilePath) if err == nil { - fileSizeMB := fileInfo.Size() / (1024 * 1024) // 转换为MB + totalBytes := fileInfo.Size() + // 把 -wal 大小一并计入：WAL 模式下数据可能暂存在 wal 文件，主库文件显小会导致漏判 + if walInfo, werr := os.Stat(dbFilePath + "-wal"); werr == nil { + totalBytes += walInfo.Size() + } + fileSizeMB := totalBytes / (1024 * 1024) // 转换为MB if fileSizeMB > global.GDATA_SHARE_DB_FILE_SIZE { needSharding = true - shardingReason = fmt.Sprintf("文件大小(%dMB)超过限制(%dMB)", fileSizeMB, global.GDATA_SHARE_DB_FILE_SIZE) + shardingReason = fmt.Sprintf("文件大小(%dMB,含WAL)超过限制(%dMB)", fileSizeMB, global.GDATA_SHARE_DB_FILE_SIZE) } } else { zlog.Error(innerLogName, "获取数据库文件大小失败:", err) @@ -122,11 +127,17 @@ func TaskShareDbInfo() { zlog.Error(innerLogName, "切换关闭时候错误", err) } } + // 等待连接彻底关闭（Count 报错即连接已关闭，可安全重命名文件）。 + // 加最大重试上限，避免连接异常未报错时无限循环卡住切库（高频切库后该段执行更频繁）。 var testTotal int64 - for { + for attempt := 0; attempt < 10; attempt++ { testError := global.GWAF_LOCAL_LOG_DB.Model(&innerbean.WebLog{}).Count(&testTotal).Error if testError != nil { - zlog.Error(innerLogName, "检测数据", testError) + zlog.Debug(innerLogName, "连接已关闭，可切库", testError) + break + } + if attempt == 9 { + zlog.Warn(innerLogName, "等待日志库连接关闭超时，强制继续切库") break } time.Sleep(1 * time.Second)