[功能] 添加数据库观测数据清理功能，支持手动和自动清理策略

Rain-kl · Rain-kl · commit 9eeccb5fc6a2 · 2026-03-19T20:48:45.000+08:00
diff --git a/docs/app-config.md b/docs/app-config.md
@@ -57,12 +57,19 @@ go run . --port 3000 --log-dir ./logs
 | `NodeOfflineThreshold` | 节点离线阈值（毫秒） | `120000` |
 | `AgentUpdateRepo` | Agent 自更新仓库 | `Rain-kl/OpenFlare` |
 | `GeoIPProvider` | 节点/IP 归属解析方式 | `ipinfo` |
+| `DatabaseAutoCleanupEnabled` | 是否启用每日自动清理观测数据 | `false` |
+| `DatabaseAutoCleanupRetentionDays` | 自动清理保留天数（至少 1 天） | `30` |
 | `GlobalApiRateLimitNum` / `GlobalApiRateLimitDuration` | 全局 API 限流次数 / 时间窗口 | `300` / `180` |
 | `GlobalWebRateLimitNum` / `GlobalWebRateLimitDuration` | 全局 Web 限流次数 / 时间窗口 | `300` / `180` |
 | `UploadRateLimitNum` / `UploadRateLimitDuration` | 上传接口限流次数 / 时间窗口 | `50` / `60` |
 | `DownloadRateLimitNum` / `DownloadRateLimitDuration` | 下载接口限流次数 / 时间窗口 | `50` / `60` |
 | `CriticalRateLimitNum` / `CriticalRateLimitDuration` | 敏感接口限流次数 / 时间窗口 | `100` / `1200` |
 
+说明：
+
+* `DatabaseAutoCleanupEnabled` 开启后，Server 会在每天凌晨 3 点自动清理 `node_access_logs`、`node_metric_snapshots`、`node_request_reports` 三类观测数据
+* `DatabaseAutoCleanupRetentionDays` 为统一保留天数，必须大于等于 1；管理端支持手动清理时留空保留天数，以直接删除对应数据集的全部历史记录
+
 ### 1.4 OpenResty 参数
 
 OpenResty 性能参数与缓存参数继续统一保存在 `Option` 表。当前常用项包括：
diff --git a/openflare_server/common/constants.go b/openflare_server/common/constants.go
@@ -55,6 +55,8 @@ var NodeOfflineThreshold = 2 * time.Minute
 var AgentHeartbeatInterval = 10000 // milliseconds
 var AgentUpdateRepo = "Rain-kl/OpenFlare"
 var GeoIPProvider = "ipinfo"
+var DatabaseAutoCleanupEnabled = false
+var DatabaseAutoCleanupRetentionDays = 30
 
 // V5 OpenResty performance settings (hot-reloadable via Option table)
 var OpenRestyWorkerProcesses = "auto"
diff --git a/openflare_server/controller/database.go b/openflare_server/controller/database.go
@@ -0,0 +1,34 @@
+package controller
+
+import (
+	"net/http"
+	"openflare/service"
+
+	"github.com/gin-gonic/gin"
+)
+
+// CleanupDatabaseObservability godoc
+// @Summary Cleanup observability tables
+// @Tags Options
+// @Accept json
+// @Produce json
+// @Security BearerAuth
+// @Success 200 {object} map[string]interface{}
+// @Router /api/option/database/cleanup [post]
+func CleanupDatabaseObservability(c *gin.Context) {
+	var input service.DatabaseCleanupInput
+	if err := decodeOptionalJSONBody(c.Request.Body, &input); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{
+			"success": false,
+			"message": "参数错误",
+			"error":   err.Error(),
+		})
+		return
+	}
+	result, err := service.CleanupDatabaseObservability(input)
+	if err != nil {
+		respondFailure(c, err.Error())
+		return
+	}
+	respondSuccess(c, result)
+}
diff --git a/openflare_server/controller/option.go b/openflare_server/controller/option.go
@@ -73,6 +73,21 @@ func validateGeoIPOption(key string, value string) error {
 	return nil
 }
 
+func validateDatabaseCleanupOption(key string, value string) error {
+	switch key {
+	case "DatabaseAutoCleanupEnabled":
+		return validateBooleanOption(key, value)
+	case "DatabaseAutoCleanupRetentionDays":
+		intValue, err := strconv.Atoi(value)
+		if err != nil || intValue < 1 {
+			return fmt.Errorf("%s 必须为大于等于 1 的整数天", key)
+		}
+		return nil
+	default:
+		return nil
+	}
+}
+
 func validateOpenRestyOption(key string, value string) error {
 	trimmed := strings.TrimSpace(value)
 
@@ -271,6 +286,13 @@ func UpdateOption(c *gin.Context) {
 		})
 		return
 	}
+	if err = validateDatabaseCleanupOption(option.Key, option.Value); err != nil {
+		c.JSON(http.StatusOK, gin.H{
+			"success": false,
+			"message": err.Error(),
+		})
+		return
+	}
 	err = model.UpdateOption(option.Key, option.Value)
 	if err != nil {
 		c.JSON(http.StatusOK, gin.H{
diff --git a/openflare_server/main.go b/openflare_server/main.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"embed"
 	"fmt"
 	"github.com/gin-contrib/sessions"
@@ -13,6 +14,7 @@ import (
 	"openflare/middleware"
 	"openflare/model"
 	"openflare/router"
+	"openflare/service"
 	"openflare/utils/geoip"
 	"os"
 	"strconv"
@@ -67,6 +69,9 @@ func main() {
 	// Initialize options
 	model.InitOptionMap()
 	geoip.InitGeoIP()
+	backgroundCtx, cancelBackgroundTasks := context.WithCancel(context.Background())
+	defer cancelBackgroundTasks()
+	service.StartDatabaseAutoCleanupScheduler(backgroundCtx)
 
 	// Initialize HTTP server
 	server := gin.Default()
diff --git a/openflare_server/model/node_access_log.go b/openflare_server/model/node_access_log.go
@@ -240,14 +240,13 @@ func ListNodeAccessLogIPTrend(query NodeAccessLogIPTrendQuery) (items []*NodeAcc
 }
 
 func DeleteNodeAccessLogsBefore(before time.Time) (deleted int64, err error) {
-	for _, table := range observabilityShardTables("node_access_logs") {
-		result := DB.Table(table).Where("logged_at < ?", before).Delete(&NodeAccessLog{})
-		if result.Error != nil {
-			return deleted, result.Error
-		}
-		deleted += result.RowsAffected
-	}
-	return deleted, nil
+	return deleteAcrossShards(DB, "node_access_logs", &NodeAccessLog{}, func(tx *gorm.DB) *gorm.DB {
+		return tx.Where("logged_at < ?", before)
+	})
+}
+
+func DeleteAllNodeAccessLogs(db *gorm.DB) (deleted int64, err error) {
+	return deleteAcrossShards(db, "node_access_logs", &NodeAccessLog{}, nil)
 }
 
 func NodeAccessLogExists(db *gorm.DB, record *NodeAccessLog) (bool, error) {
@@ -279,15 +278,9 @@ func NodeAccessLogExists(db *gorm.DB, record *NodeAccessLog) (bool, error) {
 }
 
 func DeleteNodeAccessLogsByNodeBefore(db *gorm.DB, nodeID string, before time.Time) (deleted int64, err error) {
-	db = normalizeShardedDB(db)
-	for _, table := range observabilityShardTables("node_access_logs") {
-		result := db.Table(table).Where("node_id = ? AND logged_at < ?", nodeID, before).Delete(&NodeAccessLog{})
-		if result.Error != nil {
-			return deleted, result.Error
-		}
-		deleted += result.RowsAffected
-	}
-	return deleted, nil
+	return deleteAcrossShards(db, "node_access_logs", &NodeAccessLog{}, func(tx *gorm.DB) *gorm.DB {
+		return tx.Where("node_id = ? AND logged_at < ?", nodeID, before)
+	})
 }
 
 func buildNodeAccessLogQuery(db *gorm.DB, query NodeAccessLogQuery) *gorm.DB {
diff --git a/openflare_server/model/node_metric_snapshot.go b/openflare_server/model/node_metric_snapshot.go
@@ -104,3 +104,13 @@ func NodeMetricSnapshotExists(db *gorm.DB, nodeID string, capturedAt time.Time)
 	}
 	return false, nil
 }
+
+func DeleteNodeMetricSnapshotsBefore(db *gorm.DB, before time.Time) (int64, error) {
+	return deleteAcrossShards(db, "node_metric_snapshots", &NodeMetricSnapshot{}, func(tx *gorm.DB) *gorm.DB {
+		return tx.Where("captured_at < ?", before)
+	})
+}
+
+func DeleteAllNodeMetricSnapshots(db *gorm.DB) (int64, error) {
+	return deleteAcrossShards(db, "node_metric_snapshots", &NodeMetricSnapshot{}, nil)
+}
diff --git a/openflare_server/model/node_request_report.go b/openflare_server/model/node_request_report.go
@@ -99,3 +99,13 @@ func NodeRequestReportExists(db *gorm.DB, nodeID string, windowStartedAt time.Ti
 	}
 	return false, nil
 }
+
+func DeleteNodeRequestReportsBefore(db *gorm.DB, before time.Time) (int64, error) {
+	return deleteAcrossShards(db, "node_request_reports", &NodeRequestReport{}, func(tx *gorm.DB) *gorm.DB {
+		return tx.Where("window_ended_at < ?", before)
+	})
+}
+
+func DeleteAllNodeRequestReports(db *gorm.DB) (int64, error) {
+	return deleteAcrossShards(db, "node_request_reports", &NodeRequestReport{}, nil)
+}
diff --git a/openflare_server/model/option.go b/openflare_server/model/option.go
@@ -56,6 +56,8 @@ func InitOptionMap() {
 	common.OptionMap["NodeOfflineThreshold"] = strconv.Itoa(int(common.NodeOfflineThreshold.Milliseconds()))
 	common.OptionMap["AgentUpdateRepo"] = common.AgentUpdateRepo
 	common.OptionMap["GeoIPProvider"] = common.GeoIPProvider
+	common.OptionMap["DatabaseAutoCleanupEnabled"] = strconv.FormatBool(common.DatabaseAutoCleanupEnabled)
+	common.OptionMap["DatabaseAutoCleanupRetentionDays"] = strconv.Itoa(common.DatabaseAutoCleanupRetentionDays)
 	common.OptionMap["OpenRestyWorkerProcesses"] = common.OpenRestyWorkerProcesses
 	common.OptionMap["OpenRestyWorkerConnections"] = strconv.Itoa(common.OpenRestyWorkerConnections)
 	common.OptionMap["OpenRestyWorkerRlimitNofile"] = strconv.Itoa(common.OpenRestyWorkerRlimitNofile)
@@ -219,6 +221,12 @@ func updateOptionMap(key string, value string) {
 			common.GeoIPProvider = value
 			shouldRefreshGeoIP = true
 		}
+	case "DatabaseAutoCleanupEnabled":
+		common.DatabaseAutoCleanupEnabled = value == "true"
+	case "DatabaseAutoCleanupRetentionDays":
+		if v, err := strconv.Atoi(value); err == nil && v >= 1 {
+			common.DatabaseAutoCleanupRetentionDays = v
+		}
 	case "OpenRestyWorkerProcesses":
 		if strings.TrimSpace(value) != "" {
 			common.OpenRestyWorkerProcesses = value
diff --git a/openflare_server/model/sharding.go b/openflare_server/model/sharding.go
@@ -205,6 +205,25 @@ func queryAcrossShardsWithDB[T any](db *gorm.DB, baseTable string, query func(tx
 	return items, nil
 }
 
+func deleteAcrossShards(db *gorm.DB, baseTable string, model any, apply func(tx *gorm.DB) *gorm.DB) (int64, error) {
+	db = normalizeShardedDB(db)
+	var deleted int64
+	for _, table := range observabilityShardTables(baseTable) {
+		tx := db.Table(table)
+		if apply != nil {
+			tx = apply(tx)
+		} else {
+			tx = tx.Session(&gorm.Session{AllowGlobalUpdate: true})
+		}
+		result := tx.Delete(model)
+		if result.Error != nil {
+			return deleted, result.Error
+		}
+		deleted += result.RowsAffected
+	}
+	return deleted, nil
+}
+
 func sortShardRows[T any](items []T, less func(left T, right T) bool) {
 	sort.Slice(items, func(i int, j int) bool {
 		return less(items[i], items[j])
diff --git a/openflare_server/router/api-router.go b/openflare_server/router/api-router.go
@@ -55,6 +55,7 @@ func SetApiRouter(router *gin.Engine) {
 			optionRoute.GET("/", controller.GetOptions)
 			optionRoute.POST("/update", controller.UpdateOption)
 			optionRoute.POST("/geoip/lookup", controller.LookupGeoIP)
+			optionRoute.POST("/database/cleanup", controller.CleanupDatabaseObservability)
 		}
 		updateRoute := apiRouter.Group("/update")
 		updateRoute.Use(middleware.RootAuth(), middleware.NoTokenAuth())
diff --git a/openflare_server/service/database_maintenance.go b/openflare_server/service/database_maintenance.go
diff --git a/openflare_server/service/database_maintenance_test.go b/openflare_server/service/database_maintenance_test.go
diff --git a/openflare_server/web/features/settings/api/settings.ts b/openflare_server/web/features/settings/api/settings.ts
diff --git a/openflare_server/web/features/settings/components/settings-page.tsx b/openflare_server/web/features/settings/components/settings-page.tsx
diff --git a/openflare_server/web/features/settings/types.ts b/openflare_server/web/features/settings/types.ts

Original file line number	Diff line number	Diff line change
`@@ -104,3 +104,13 @@ func NodeMetricSnapshotExists(db *gorm.DB, nodeID string, capturedAt time.Time)`
`104`	`104`	`}`
`105`	`105`	`return false, nil`
`106`	`106`	`}`
	`107`	`+`
	`108`	`+func DeleteNodeMetricSnapshotsBefore(db *gorm.DB, before time.Time) (int64, error) {`
	`109`	`+ return deleteAcrossShards(db, "node_metric_snapshots", &NodeMetricSnapshot{}, func(tx gorm.DB) gorm.DB {`
	`110`	`+ return tx.Where("captured_at < ?", before)`
	`111`	`+ })`
	`112`	`+}`
	`113`	`+`
	`114`	`+func DeleteAllNodeMetricSnapshots(db *gorm.DB) (int64, error) {`
	`115`	`+ return deleteAcrossShards(db, "node_metric_snapshots", &NodeMetricSnapshot{}, nil)`
	`116`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -99,3 +99,13 @@ func NodeRequestReportExists(db *gorm.DB, nodeID string, windowStartedAt time.Ti`
`99`	`99`	`}`
`100`	`100`	`return false, nil`
`101`	`101`	`}`
	`102`	`+`
	`103`	`+func DeleteNodeRequestReportsBefore(db *gorm.DB, before time.Time) (int64, error) {`
	`104`	`+ return deleteAcrossShards(db, "node_request_reports", &NodeRequestReport{}, func(tx gorm.DB) gorm.DB {`
	`105`	`+ return tx.Where("window_ended_at < ?", before)`
	`106`	`+ })`
	`107`	`+}`
	`108`	`+`
	`109`	`+func DeleteAllNodeRequestReports(db *gorm.DB) (int64, error) {`
	`110`	`+ return deleteAcrossShards(db, "node_request_reports", &NodeRequestReport{}, nil)`
	`111`	`+}`