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

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 {

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)
+}

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)
+}

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

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])