From 4065bc3834b1df707924856400ad287498761b4a Mon Sep 17 00:00:00 2001 From: samwaf Date: Fri, 26 Jun 2026 11:32:43 +0800 Subject: [PATCH] perf: tune SQLite pragmas and reclaim log space after deletion #838 --- wafdb/localdb.go | 64 +++++++--- wafdb/sqlite_perf_test.go | 243 ++++++++++++++++++++++++++++++++++++++ waftask/task_history.go | 15 +++ 3 files changed, 306 insertions(+), 16 deletions(-) create mode 100644 wafdb/sqlite_perf_test.go diff --git a/wafdb/localdb.go b/wafdb/localdb.go index e2959cc..c04ee6c 100644 --- a/wafdb/localdb.go +++ b/wafdb/localdb.go @@ -30,6 +30,28 @@ import ( "gorm.io/gorm" ) +// applyPerfPragmas 统一设置 SQLite 性能/可靠性 pragma。 +// relaxedSync=true 使用 synchronous=NORMAL（WAL 下安全，仅在 checkpoint 时 fsync，大幅减少磁盘同步、 +// 提升高频写入吞吐），适用于日志/统计库（可容忍断电丢失最后若干条未 checkpoint 记录）； +// 核心库传 false 保持默认 FULL 以保证可靠性。 +func applyPerfPragmas(db *gorm.DB, relaxedSync bool) { + // 启用 WAL 模式（读写并发） + _ = db.Exec("PRAGMA journal_mode=WAL;") + // 升级重叠期可能新旧 Worker 并存写库，busy_timeout 让写操作在锁竞争时重试等待，避免立即报 database is locked + _ = db.Exec("PRAGMA busy_timeout=5000;") + if relaxedSync { + _ = db.Exec("PRAGMA synchronous=NORMAL;") + } + // 64MB page cache，减少磁盘随机读 + _ = db.Exec("PRAGMA cache_size=-65536;") + // 排序/临时表走内存，加速 ORDER BY / 临时表 + _ = db.Exec("PRAGMA temp_store=MEMORY;") + // 注意：本项目所有库均带 _db_key 加密，禁止开启 PRAGMA mmap_size。 + // mmap 直接读裸文件、绕过加密 codec，会导致读出密文(integrity_check 报 SQLITE_CORRUPT、Count 归零)。 + // 每累计约 1000 页触发自动 checkpoint，配合主动 checkpoint 控制 WAL 增长 + _ = db.Exec("PRAGMA wal_autocheckpoint=1000;") +} + func InitCoreDb(currentDir string) (bool, error) { if dialect.Get().Name() == "mysql" { return InitCoreDbMySQL() @@ -69,10 +91,8 @@ func InitCoreDb(currentDir string) (bool, error) { if err != nil { panic("failed to connect database") } - // 启用 WAL 模式 - _ = db.Exec("PRAGMA journal_mode=WAL;") - // 升级重叠期可能新旧 Worker 并存写库，busy_timeout 让写操作在锁竞争时重试等待，避免立即报 database is locked - _ = db.Exec("PRAGMA busy_timeout=5000;") + // 核心库保持 synchronous=FULL 保证可靠性，其余性能 pragma 统一设置 + applyPerfPragmas(db, false) // 创建自定义日志记录器 gormLogger := NewGormZLogger() @@ -148,10 +168,8 @@ func InitLogDb(currentDir string) (bool, error) { if err != nil { panic("failed to connect database") } - // 启用 WAL 模式 - _ = db.Exec("PRAGMA journal_mode=WAL;") - // 升级重叠期可能新旧 Worker 并存写库，busy_timeout 让写操作在锁竞争时重试等待，避免立即报 database is locked - _ = db.Exec("PRAGMA busy_timeout=5000;") + // 日志/统计库使用 synchronous=NORMAL 提升高频写入吞吐，其余性能 pragma 统一设置 + applyPerfPragmas(db, true) // 创建自定义日志记录器 gormLogger := NewGormZLogger() if global.GWAF_LOG_DEBUG_DB_ENABLE == true { @@ -228,10 +246,8 @@ func InitManaulLogDb(currentDir string, custFileName string) { if err != nil { panic("failed to connect database") } - // 启用 WAL 模式 - _ = db.Exec("PRAGMA journal_mode=WAL;") - // 升级重叠期可能新旧 Worker 并存写库，busy_timeout 让写操作在锁竞争时重试等待，避免立即报 database is locked - _ = db.Exec("PRAGMA busy_timeout=5000;") + // 日志/统计库使用 synchronous=NORMAL 提升高频写入吞吐，其余性能 pragma 统一设置 + applyPerfPragmas(db, true) // 创建自定义日志记录器 gormLogger := NewGormZLogger() if global.GWAF_LOG_DEBUG_DB_ENABLE == true { @@ -264,6 +280,24 @@ func InitManaulLogDb(currentDir string, custFileName string) { } } +// CloseManualLogDb 关闭并移除一个按需打开的归档日志分片连接（若存在）， +// 供归档清理删除文件前调用，避免删正在被打开查询的 .db 文件。 +func CloseManualLogDb(custFileName string) { + if global.GDATA_CURRENT_LOG_DB_MAP == nil { + return + } + db := global.GDATA_CURRENT_LOG_DB_MAP[custFileName] + if db == nil { + return + } + if sqlDB, err := db.DB(); err == nil { + if cerr := sqlDB.Close(); cerr != nil { + zlog.Warn("关闭归档分片连接失败", "file", custFileName, "error", cerr.Error()) + } + } + delete(global.GDATA_CURRENT_LOG_DB_MAP, custFileName) +} + func InitStatsDb(currentDir string) (bool, error) { if dialect.Get().Name() == "mysql" { return InitStatsDbMySQL() @@ -285,10 +319,8 @@ func InitStatsDb(currentDir string) (bool, error) { if err != nil { panic("failed to connect database") } - // 启用 WAL 模式 - _ = db.Exec("PRAGMA journal_mode=WAL;") - // 升级重叠期可能新旧 Worker 并存写库，busy_timeout 让写操作在锁竞争时重试等待，避免立即报 database is locked - _ = db.Exec("PRAGMA busy_timeout=5000;") + // 日志/统计库使用 synchronous=NORMAL 提升高频写入吞吐，其余性能 pragma 统一设置 + applyPerfPragmas(db, true) // 创建自定义日志记录器 gormLogger := NewGormZLogger() if global.GWAF_LOG_DEBUG_DB_ENABLE == true { diff --git a/wafdb/sqlite_perf_test.go b/wafdb/sqlite_perf_test.go new file mode 100644 index 0000000..72b3480 --- /dev/null +++ b/wafdb/sqlite_perf_test.go @@ -0,0 +1,243 @@ +package wafdb + +import ( + "SamWaf/innerbean" + "fmt" + "net/url" + "os" + "sort" + "strings" + "testing" + "time" + + sqlite "github.com/samwafgo/sqlitedriver" + "gorm.io/gorm" + "gorm.io/gorm/logger" +) + +// 这些用例验证 Issue #838 的两类目标： +// - 响应速度：T1 高频写入吞吐(FULL vs NORMAL)、T2 不同数据量下查询延迟 +// - 文件完整性：T3 多分片 integrity_check + 零丢数据、T4 WAL 受控 +// +// 注意：SQLite 驱动需要 CGO，运行需 `CGO_ENABLED=1` + C 编译器(mingw)。 +// 用临时目录隔离，不污染真实 data 目录。 + +// openPerfTestDB 打开一个临时加密日志库并应用性能 pragma + 迁移 web_logs 表。 +func openPerfTestDB(t *testing.T, path string, relaxedSync bool) *gorm.DB { + t.Helper() + key := url.QueryEscape("perf_test_key_3Y)(27Et") + dns := fmt.Sprintf("%s?_db_key=%s", path, key) + db, err := gorm.Open(sqlite.Open(dns), &gorm.Config{ + Logger: logger.Default.LogMode(logger.Silent), + }) + if err != nil { + t.Fatalf("打开测试库失败: %v", err) + } + applyPerfPragmas(db, relaxedSync) + if err := db.AutoMigrate(&innerbean.WebLog{}); err != nil { + t.Fatalf("迁移 web_logs 失败: %v", err) + } + // 关闭连接，否则 Windows 下 t.TempDir 清理会因文件被占用而失败 + t.Cleanup(func() { + if sqlDB, err := db.DB(); err == nil { + _ = sqlDB.Close() + } + }) + return db +} + +// genWebLogs 构造 n 条贴近真实的访问日志（含 BODY/POST_FORM 大字段）。 +func genWebLogs(n int) []*innerbean.WebLog { + body := strings.Repeat("a=1&b=2&payload=", 64) // ~1KB + postForm := strings.Repeat("user=test&token=abc&", 32) // ~640B + now := time.Now() + logs := make([]*innerbean.WebLog, 0, n) + for i := 0; i < n; i++ { + ts := now.Add(time.Duration(i) * time.Millisecond) + logs = append(logs, &innerbean.WebLog{ + HOST: "site.example.com", + URL: fmt.Sprintf("/api/path/%d?x=%d", i, i), + METHOD: "POST", + SRC_IP: fmt.Sprintf("10.0.%d.%d", (i/256)%256, i%256), + BODY: body, + POST_FORM: postForm, + HEADER: "Content-Type: application/x-www-form-urlencoded", + STATUS_CODE: 200, + USER_CODE: "test_user", + TenantId: "test_user", + HOST_CODE: "host_01", + CREATE_TIME: ts.Format("2006-01-02 15:04:05"), + UNIX_ADD_TIME: ts.UnixNano() / 1e6, + Day: ts.Year()*10000 + int(ts.Month())*100 + ts.Day(), + }) + } + return logs +} + +func pctile(d []time.Duration, p float64) time.Duration { + if len(d) == 0 { + return 0 + } + cp := append([]time.Duration(nil), d...) + sort.Slice(cp, func(i, j int) bool { return cp[i] < cp[j] }) + idx := int(float64(len(cp)-1) * p) + return cp[idx] +} + +func batchInsert(t *testing.T, db *gorm.DB, logs []*innerbean.WebLog, batchSize int) (time.Duration, []time.Duration) { + t.Helper() + var perBatch []time.Duration + start := time.Now() + for i := 0; i < len(logs); i += batchSize { + end := i + batchSize + if end > len(logs) { + end = len(logs) + } + bs := time.Now() + if err := db.CreateInBatches(logs[i:end], batchSize).Error; err != nil { + t.Fatalf("批量写入失败: %v", err) + } + perBatch = append(perBatch, time.Since(bs)) + } + return time.Since(start), perBatch +} + +// T1：高频写入吞吐与延迟，对比 synchronous=FULL 与 NORMAL。 +func TestSqliteHighFreqWrite(t *testing.T) { + if testing.Short() { + t.Skip("short 模式跳过写入压测") + } + const n = 50000 + const batch = 100 + dir := t.TempDir() + + fullDB := openPerfTestDB(t, dir+"/full.db", false) + normalDB := openPerfTestDB(t, dir+"/normal.db", true) + + fullElapsed, fullBatches := batchInsert(t, fullDB, genWebLogs(n), batch) + normalElapsed, normalBatches := batchInsert(t, normalDB, genWebLogs(n), batch) + + fullQPS := float64(n) / fullElapsed.Seconds() + normalQPS := float64(n) / normalElapsed.Seconds() + t.Logf("写入 %d 条：FULL=%v (%.0f QPS, 批P95=%v) NORMAL=%v (%.0f QPS, 批P95=%v) 提速=%.2fx", + n, fullElapsed, fullQPS, pctile(fullBatches, 0.95), + normalElapsed, normalQPS, pctile(normalBatches, 0.95), + fullElapsed.Seconds()/normalElapsed.Seconds()) + + var cnt int64 + normalDB.Model(&innerbean.WebLog{}).Count(&cnt) + if cnt != int64(n) { + t.Fatalf("NORMAL 库写入条数不符: got=%d want=%d", cnt, n) + } +} + +// T2：不同数据量下分页查询延迟（模拟 GetListApi：时间范围 + 倒序 + Count）。 +func TestSqliteQueryLatencyBySize(t *testing.T) { + if testing.Short() { + t.Skip("short 模式跳过查询压测") + } + dir := t.TempDir() + for _, size := range []int{2000, 20000} { + db := openPerfTestDB(t, fmt.Sprintf("%s/q_%d.db", dir, size), true) + if _, _, err := func() (time.Duration, []time.Duration, error) { + el, pb := batchInsert(t, db, genWebLogs(size), 100) + return el, pb, nil + }(); err != nil { + t.Fatal(err) + } + + nowMs := time.Now().UnixNano() / 1e6 + var lat []time.Duration + for page := 1; page <= 5; page++ { + st := time.Now() + var rows []innerbean.WebLog + db.Table("web_logs"). + Where("unix_add_time >= ? and unix_add_time <= ?", int64(0), nowMs+86400000). + Order("unix_add_time desc"). + Limit(20).Offset(20 * (page - 1)). + Find(&rows) + var total int64 + db.Table("web_logs"). + Where("unix_add_time >= ? and unix_add_time <= ?", int64(0), nowMs+86400000). + Count(&total) + lat = append(lat, time.Since(st)) + } + t.Logf("表 %d 行：查询 P50=%v P95=%v", size, pctile(lat, 0.5), pctile(lat, 0.95)) + } +} + +// T3：多分片完整性 + 零丢数据。模拟切库后存在多个 .db 文件，逐个 integrity_check，并校验各分片条数之和。 +func TestSqliteIntegrityAcrossShards(t *testing.T) { + dir := t.TempDir() + counts := []int{300, 500, 200} + total := 0 + for i, c := range counts { + path := fmt.Sprintf("%s/local_log_shard%d.db", dir, i) + db := openPerfTestDB(t, path, true) + if err := db.CreateInBatches(genWebLogs(c), 100).Error; err != nil { + t.Fatalf("写入分片失败: %v", err) + } + // checkpoint 让数据落主库，贴近切库后归档文件状态 + db.Exec("PRAGMA wal_checkpoint(TRUNCATE);") + + var res string + db.Raw("PRAGMA integrity_check;").Scan(&res) + if res != "ok" { + t.Fatalf("分片 %s integrity_check 未通过: %q", path, res) + } + var n int64 + db.Model(&innerbean.WebLog{}).Count(&n) + if n != int64(c) { + t.Fatalf("分片 %d 条数不符: got=%d want=%d", i, n, c) + } + total += int(n) + + if sqlDB, err := db.DB(); err == nil { + sqlDB.Close() + } + } + want := 0 + for _, c := range counts { + want += c + } + if total != want { + t.Fatalf("各分片条数之和不符(疑似丢数据): got=%d want=%d", total, want) + } + t.Logf("跨 %d 个分片完整性校验通过，总条数=%d", len(counts), total) +} + +// T4：WAL 受控。关闭自动 checkpoint 写入后 WAL 增大，手动 wal_checkpoint(TRUNCATE) 后回落。 +func TestSqliteWalBounded(t *testing.T) { + if testing.Short() { + t.Skip("short 模式跳过 WAL 压测") + } + dir := t.TempDir() + path := dir + "/wal.db" + db := openPerfTestDB(t, path, true) + // 关闭自动 checkpoint 以便观察 WAL 累积 + db.Exec("PRAGMA wal_autocheckpoint=0;") + + if err := db.CreateInBatches(genWebLogs(20000), 100).Error; err != nil { + t.Fatalf("写入失败: %v", err) + } + walBefore := fileSize(path + "-wal") + if walBefore == 0 { + t.Skip("未观察到 WAL 文件（驱动可能即时 checkpoint），跳过断言") + } + + if err := db.Exec("PRAGMA wal_checkpoint(TRUNCATE);").Error; err != nil { + t.Fatalf("checkpoint 失败: %v", err) + } + walAfter := fileSize(path + "-wal") + t.Logf("WAL: checkpoint 前=%dKB 后=%dKB", walBefore/1024, walAfter/1024) + if walAfter >= walBefore { + t.Fatalf("checkpoint(TRUNCATE) 后 WAL 未回落: before=%d after=%d", walBefore, walAfter) + } +} + +func fileSize(p string) int64 { + if fi, err := os.Stat(p); err == nil { + return fi.Size() + } + return 0 +} diff --git a/waftask/task_history.go b/waftask/task_history.go index 2b1a83e..447b5e9 100644 --- a/waftask/task_history.go +++ b/waftask/task_history.go @@ -5,6 +5,7 @@ import ( "SamWaf/global" "SamWaf/service/waf_service" "SamWaf/utils" + "SamWaf/wafdb/dialect" "os" "time" ) @@ -14,6 +15,20 @@ func TaskDeleteHistoryInfo() { zlog.Debug("TaskDeleteHistoryInfo") deleteBeforeDay := time.Now().AddDate(0, 0, -int(global.GDATA_DELETE_INTERVAL)).Format("2006-01-02 15:04") waf_service.WafLogServiceApp.DeleteHistory(deleteBeforeDay) + + // 清理过期的归档分片文件（高频切库后 live 库只存最近数据，真正的保留期回收靠删归档文件） + CleanExpiredArchiveShard() + + // 仅 SQLite：DELETE 不会收缩文件，主动 checkpoint 截断 WAL 并 VACUUM 回收空间。 + // 本任务在低峰(05:00)执行，且 live 库已被高频切库控制在阈值内，VACUUM 成本可控。 + if dialect.Get().IsFileBased() && global.GWAF_LOCAL_LOG_DB != nil { + if err := global.GWAF_LOCAL_LOG_DB.Exec("PRAGMA wal_checkpoint(TRUNCATE);").Error; err != nil { + zlog.Warn("TaskDeleteHistoryInfo wal_checkpoint 失败", "error", err.Error()) + } + if err := global.GWAF_LOCAL_LOG_DB.Exec("VACUUM;").Error; err != nil { + zlog.Warn("TaskDeleteHistoryInfo VACUUM 失败", "error", err.Error()) + } + } } // TaskHistoryDownload 删除老旧数据