tcjlj/OpenFlare
1
0
Fork 0
mirror of https://github.com/Rain-kl/OpenFlare.git synced 2026-09-29 14:06:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'codex/log-database-decoupling'

Browse Source 
# Conflicts:
#	docs/changelog/index.md
#	frontend/app/(main)/error-pages/page.tsx
#	internal/infra/persistence/migrator/migrator_test.go
This commit is contained in:
ryan
2026-08-09 08:37:37 +08:00
parent 7e518987de 93ec3096f3
commit 7e8aa5fa0f
95 changed files with 10697 additions and 3744 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/apps/admin/logs/routers.go
+20 -17
View File
@@ -14,10 +14,9 @@ import (
"time"
"github.com/Rain-kl/Wavelet/internal/apps/admin"
"github.com/Rain-kl/Wavelet/internal/infra/config"
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
"github.com/Rain-kl/Wavelet/internal/repository"
analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
"github.com/Rain-kl/Wavelet/internal/repository/logstore"
"github.com/Rain-kl/Wavelet/pkg/logger"
"github.com/gin-gonic/gin"
@@ -158,8 +157,8 @@ type accessLogsResponse struct {
List []accessLogItem `json:"list"`
}
func buildAccessLogFilter(ctx context.Context, c *gin.Context) (analyticsrepo.AccessLogFilter, error) {
filter := analyticsrepo.AccessLogFilter{}
func buildAccessLogFilter(ctx context.Context, c *gin.Context) (analyticsmodel.AccessLogFilter, error) {
filter := analyticsmodel.AccessLogFilter{}
username := c.Query("username")
if username != "" {
@@ -227,7 +226,7 @@ func enrichAccessLogsWithUsers(ctx context.Context, list []accessLogItem) {
// GetAccessLogs 获取 ClickHouse 异步采集的访问日志
// @Summary 获取用户访问日志
// @Description 分页并按照用户、接口路径、时间范围等维度检索 ClickHouse 用户访问日志列表（需要管理员权限，ClickHouse 未启用时报错）
// @Description 分页并按照用户、接口路径、时间范围等维度检索用户访问日志列表（需要管理员权限，日志存储未启用时报错）
// @Tags admin
// @Produce json
// @Security SessionCookie
@@ -238,14 +237,16 @@ func enrichAccessLogsWithUsers(ctx context.Context, list []accessLogItem) {
// @Param start_time query string false "起始时间（RFC3339 或 YYYY-MM-DD HH:MM:SS）"
// @Param end_time query string false "结束时间（RFC3339 或 YYYY-MM-DD HH:MM:SS）"
// @Success 200 {object} response.Any{data=logs.accessLogsResponse} "访问日志列表"
// @Failure 400 {object} response.Any "ClickHouse 未启用或参数错误"
// @Failure 400 {object} response.Any "日志存储未启用或参数错误"
// @Failure 401 {object} response.Any "未登录"
// @Failure 403 {object} response.Any "无管理员权限"
// @Router /api/v1/admin/logs/access [get]
func GetAccessLogs(c *gin.Context) {
ctx := c.Request.Context()
if !config.Config.ClickHouse.Enabled || !db.ChConnReady() {
response.AbortWithError(c, http.StatusBadRequest, "ClickHouse 存储服务未启用，无法检索访问日志")
store, err := logstore.Active(ctx)
if err != nil {
logger.ErrorF(ctx, "获取日志存储实例失败: %v", err)
response.AbortWithError(c, http.StatusBadRequest, "日志存储未启用，无法检索访问日志")
return
}
@@ -271,7 +272,7 @@ func GetAccessLogs(c *gin.Context) {
return
}
logs, total, err := analyticsrepo.ListAccessLogs(ctx, filter, page, pageSize)
logs, total, err := store.UserAccessLogs.List(ctx, filter, page, pageSize)
if err != nil {
response.AbortWithError(c, http.StatusInternalServerError, err.Error())
return
@@ -333,25 +334,27 @@ type logsAnalyticsResponse struct {
// GetLogsAnalytics 获取 ClickHouse 访问日志图表聚合指标
// @Summary 获取访问日志分析数据
// @Description 聚合统计最近 7 天的每日访问趋势、浏览器分布以及前 10 名最活跃用户排行（需要管理员权限，ClickHouse 未启用时报错）
// @Description 聚合统计最近 7 天的每日访问趋势、浏览器分布以及前 10 名最活跃用户排行（需要管理员权限，日志存储未启用时报错）
// @Tags admin
// @Produce json
// @Security SessionCookie
// @Success 200 {object} response.Any{data=logs.logsAnalyticsResponse} "分析统计数据"
// @Failure 400 {object} response.Any "ClickHouse 未启用"
// @Failure 400 {object} response.Any "日志存储未启用"
// @Failure 401 {object} response.Any "未登录"
// @Failure 403 {object} response.Any "无管理员权限"
// @Router /api/v1/admin/logs/analytics [get]
func GetLogsAnalytics(c *gin.Context) {
ctx := c.Request.Context()
if !config.Config.ClickHouse.Enabled || !db.ChConnReady() {
response.AbortWithError(c, http.StatusBadRequest, "ClickHouse 存储服务未启用，无法获取分析数据")
store, err := logstore.Active(ctx)
if err != nil {
logger.ErrorF(ctx, "获取日志存储实例失败: %v", err)
response.AbortWithError(c, http.StatusBadRequest, "日志存储未启用，无法获取分析数据")
return
}
startTime := time.Now().AddDate(0, 0, -(analyticsDays - 1)).Truncate(hoursInDay * time.Hour)
trendPoints, err := analyticsrepo.GetDailyTrend(ctx, analyticsDays)
trendPoints, err := store.UserAccessLogs.GetDailyTrend(ctx, analyticsDays)
if err != nil {
response.AbortWithError(c, http.StatusInternalServerError, "查询访问趋势失败: "+err.Error())
return
@@ -364,7 +367,7 @@ func GetLogsAnalytics(c *gin.Context) {
}
}
browserPoints, err := analyticsrepo.GetBrowserDistribution(ctx, startTime)
browserPoints, err := store.UserAccessLogs.GetBrowserDistribution(ctx, startTime)
if err != nil {
response.AbortWithError(c, http.StatusInternalServerError, "查询浏览器分布失败: "+err.Error())
return
@@ -377,7 +380,7 @@ func GetLogsAnalytics(c *gin.Context) {
}
}
topUserPoints, err := analyticsrepo.GetTopActiveUsers(ctx, startTime, topActiveLimit)
topUserPoints, err := store.UserAccessLogs.GetTopActiveUsers(ctx, startTime, topActiveLimit)
if err != nil {
response.AbortWithError(c, http.StatusInternalServerError, "查询活跃用户失败: "+err.Error())
return
internal/apps/admin/status/clickhouse.go
+99 -15
View File
@@ -4,43 +4,127 @@
package status
import (
"context"
"errors"
"net/http"
"github.com/Rain-kl/Wavelet/internal/apps/openflare/chwriter"
"github.com/Rain-kl/Wavelet/internal/apps/risk_control"
"github.com/Rain-kl/Wavelet/internal/infra/config"
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
"github.com/Rain-kl/Wavelet/internal/infra/persistence/batchwriter"
analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
"github.com/Rain-kl/Wavelet/internal/model"
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
"github.com/Rain-kl/Wavelet/internal/repository"
"github.com/Rain-kl/Wavelet/internal/repository/logstore"
"github.com/Rain-kl/Wavelet/internal/shared/response"
"github.com/Rain-kl/Wavelet/pkg/logger"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
)
// GetClickHouseStatus returns ClickHouse operational metrics for administrators.
// @Summary 获取 ClickHouse 运行指标
// @Description 返回 ClickHouse parts、mutation、async_insert 队列及进程内 batch writer 指标，需要管理员权限
// 日志库名取值（与 model 配置值、logstore provider 分支保持一致）。
const (
logDBNamePostgres = "postgres"
logDBNameSQLite = "sqlite"
logDBNameClickHouse = "clickhouse"
)
// defaultLogRetentionDays 日志保留天数配置缺失时的兜底值（与 seed 默认一致）。
const defaultLogRetentionDays = 90
// LogDatabaseStatus 日志库状态。
type LogDatabaseStatus struct {
ActiveDatabase string `json:"active_database"`
Migration string `json:"migration"` // idle | migrating
RetentionDays map[string]int `json:"retention_days"`
AvailableTargets []string `json:"available_targets"`
ClickHouse *analyticsmodel.ClickHouseOperationalStats `json:"clickhouse,omitempty"`
}
// GetLogDatabaseStatus 返回当前日志库状态。
// @Summary 获取日志数据库状态
// @Description 返回当前日志主库、迁移状态、各库保留天数与合法迁移目标，需要管理员权限
// @Tags admin
// @Produce json
// @Security SessionCookie
// @Success 200 {object} response.Any{data=analyticsrepo.ClickHouseOperationalStats} "获取成功"
// @Failure 400 {object} response.Any "ClickHouse 未启用"
// @Success 200 {object} response.Any{data=status.LogDatabaseStatus} "获取成功"
// @Failure 401 {object} response.Any "未登录"
// @Failure 403 {object} response.Any "无管理员权限"
// @Failure 500 {object} response.Any "内部错误"
// @Router /api/v1/admin/status/clickhouse [get]
func GetClickHouseStatus(c *gin.Context) {
if !config.Config.ClickHouse.Enabled || !db.ChConnReady() {
response.AbortWithError(c, http.StatusBadRequest, "ClickHouse 存储服务未启用")
// @Router /api/v1/admin/status/log-database [get]
func GetLogDatabaseStatus(c *gin.Context) {
ctx := c.Request.Context()
store, err := logstore.Active(ctx)
if err != nil {
logger.ErrorF(ctx, "获取日志存储实例失败: %v", err)
response.AbortInternal(c, "日志存储初始化失败")
return
}
stats, err := analyticsrepo.GetClickHouseOperationalStats(c.Request.Context())
// 分支判定复用同一 store 实例的 ActiveDatabase，避免与 Active 解析之间出现 TOCTOU。
activeDB, err := store.Status.ActiveDatabase(ctx)
if err != nil {
response.AbortInternal(c, "获取 ClickHouse 运行指标失败")
logger.ErrorF(ctx, "获取日志库状态失败: %v", err)
response.AbortInternal(c, "获取日志库状态失败")
return
}
stats.BatchWriters = collectBatchWriterStats()
c.JSON(http.StatusOK, response.OK(stats))
migration := "idle"
if logstore.Migrating(ctx) {
migration = "migrating"
}
out := LogDatabaseStatus{
ActiveDatabase: activeDB,
Migration: migration,
RetentionDays: map[string]int{
logDBNamePostgres: retentionOr(ctx, model.ConfigKeyLogRetentionDaysPostgres),
logDBNameSQLite: retentionOr(ctx, model.ConfigKeyLogRetentionDaysSQLite),
logDBNameClickHouse: retentionOr(ctx, model.ConfigKeyLogRetentionDaysClickHouse),
},
AvailableTargets: availableTargets(activeDB),
}
if activeDB == logDBNameClickHouse {
stats, err := store.Status.ClickHouseOperationalStats(ctx)
if err != nil {
logger.ErrorF(ctx, "获取 ClickHouse 运行指标失败: %v", err)
} else {
stats.BatchWriters = collectBatchWriterStats()
out.ClickHouse = stats
}
}
c.JSON(http.StatusOK, response.OK(out))
}
// retentionOr 读取保留天数配置，缺失或非法时返回默认值。
func retentionOr(ctx context.Context, key string) int {
v, err := repository.GetIntByKey(ctx, key)
if err != nil {
if !errors.Is(err, gorm.ErrRecordNotFound) {
logger.ErrorF(ctx, "读取日志保留天数配置失败 key=%s: %v", key, err)
}
return defaultLogRetentionDays
}
if v < 1 {
return defaultLogRetentionDays
}
return v
}
// availableTargets 返回当前日志主库的合法迁移目标（复用调用方已解析的 active）：
// 当前为 clickhouse 时目标为主库（postgres/sqlite 按启动配置）；当前为主库时目标为 clickhouse（仅 CH 启用时）。
func availableTargets(active string) []string {
if active == logDBNameClickHouse {
if config.Config.Database.Enabled {
return []string{logDBNamePostgres}
}
return []string{logDBNameSQLite}
}
if config.Config.ClickHouse.Enabled {
return []string{logDBNameClickHouse}
}
return []string{}
}
func collectBatchWriterStats() []batchwriter.Stats {
internal/apps/admin/status/clickhouse_test.go
+117
View File
@@ -0,0 +1,117 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package status
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"reflect"
"testing"
"github.com/Rain-kl/Wavelet/internal/infra/config"
"github.com/Rain-kl/Wavelet/internal/repository/logstore"
"github.com/gin-gonic/gin"
)
// restoreConfig 恢复测试中临时修改的全局配置。
func restoreConfig(t *testing.T) {
t.Helper()
dbEnabled := config.Config.Database.Enabled
chEnabled := config.Config.ClickHouse.Enabled
t.Cleanup(func() {
config.Config.Database.Enabled = dbEnabled
config.Config.ClickHouse.Enabled = chEnabled
})
}
func TestAvailableTargets(t *testing.T) {
restoreConfig(t)
logstore.ResetForTest()
t.Cleanup(logstore.ResetForTest)
// 当前 clickhouse → 主库（postgres/sqlite 按启动配置）。
logstore.SetConfigReader(func(_ context.Context, key string) (string, error) {
if key == "log_database" {
return logDBNameClickHouse, nil
}
return "", nil
})
config.Config.Database.Enabled = true
config.Config.ClickHouse.Enabled = true
if got := availableTargets(logDBNameClickHouse); !reflect.DeepEqual(got, []string{logDBNamePostgres}) {
t.Fatalf("clickhouse active + postgres main: got %v, want [postgres]", got)
}
config.Config.Database.Enabled = false
if got := availableTargets(logDBNameClickHouse); !reflect.DeepEqual(got, []string{logDBNameSQLite}) {
t.Fatalf("clickhouse active + sqlite main: got %v, want [sqlite]", got)
}
// 当前主库 → clickhouse（CH 启用时）。
logstore.SetConfigReader(func(_ context.Context, key string) (string, error) {
if key == "log_database" {
return logDBNamePostgres, nil
}
return "", nil
})
config.Config.ClickHouse.Enabled = true
if got := availableTargets(logDBNamePostgres); !reflect.DeepEqual(got, []string{logDBNameClickHouse}) {
t.Fatalf("postgres active: got %v, want [clickhouse]", got)
}
// CH 禁用时排除 clickhouse。
config.Config.ClickHouse.Enabled = false
if got := availableTargets(logDBNamePostgres); len(got) != 0 {
t.Fatalf("CH disabled: got %v, want empty", got)
}
}
// TestGetLogDatabaseStatusSmoke 覆盖 handler 的 CH 激活分支（无需 DB/CH 连接）。
func TestGetLogDatabaseStatusSmoke(t *testing.T) {
restoreConfig(t)
config.Config.Database.Enabled = false
config.Config.ClickHouse.Enabled = true
logstore.ResetForTest()
t.Cleanup(logstore.ResetForTest)
logstore.SetConfigReader(func(_ context.Context, key string) (string, error) {
if key == "log_database" {
return logDBNameClickHouse, nil
}
return "", nil
})
gin.SetMode(gin.TestMode)
w := httptest.NewRecorder()
c, _ := gin.CreateTestContext(w)
c.Request = httptest.NewRequest(http.MethodGet, "/api/v1/admin/status/log-database", nil)
GetLogDatabaseStatus(c)
if w.Code != http.StatusOK {
t.Fatalf("status code = %d, want 200; body=%s", w.Code, w.Body.String())
}
var resp struct {
Data LogDatabaseStatus `json:"data"`
}
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("unmarshal body: %v", err)
}
if resp.Data.ActiveDatabase != logDBNameClickHouse {
t.Fatalf("active_database = %q, want clickhouse", resp.Data.ActiveDatabase)
}
if resp.Data.Migration != "idle" {
t.Fatalf("migration = %q, want idle", resp.Data.Migration)
}
for _, key := range []string{logDBNamePostgres, logDBNameSQLite, logDBNameClickHouse} {
if got := resp.Data.RetentionDays[key]; got != defaultLogRetentionDays {
t.Fatalf("retention_days[%s] = %d, want default %d", key, got, defaultLogRetentionDays)
}
}
if got := resp.Data.AvailableTargets; !reflect.DeepEqual(got, []string{logDBNameSQLite}) {
t.Fatalf("available_targets = %v, want [sqlite] (test main DB disabled)", got)
}
}
internal/apps/admin/system_config/logics.go
+5
View File
@@ -17,6 +17,11 @@ import (
)
func createSystemConfig(ctx context.Context, req CreateSystemConfigRequest) error {
// 防御：受保护 key（log_database / log_db_migration）仅允许内部写入，Handler 已拦截。
if isProtectedConfigKey(req.Key) {
return errors.New(protectedConfigKeyMessage)
}
exists, err := repository.SystemConfigExists(ctx, req.Key)
if err != nil {
return err
internal/apps/admin/system_config/routers.go
+25
View File
@@ -27,6 +27,17 @@ import (
const maskedConfigValue = "******"
// protectedConfigKeyMessage 命中受保护 key 时返回给管理员的业务错误文案。
const protectedConfigKeyMessage = "该配置项由系统任务管理，禁止手动修改"
// protectedConfigKeys 仅允许内部（迁移任务/bootstrap）写入的 key。
var protectedConfigKeys = map[string]bool{
model.ConfigKeyLogDatabase: true,
model.ConfigKeyLogDBMigration: true,
}
func isProtectedConfigKey(key string) bool { return protectedConfigKeys[key] }
// CreateSystemConfigRequest 创建系统配置请求
type CreateSystemConfigRequest struct {
Key string `json:"key" binding:"required,max=64"`
@@ -64,11 +75,21 @@ func CreateSystemConfig(c *gin.Context) {
return
}
// 与 PUT 路径一致：log_database / log_db_migration 仅允许内部（迁移任务/bootstrap）写入。
if isProtectedConfigKey(req.Key) {
response.AbortBadRequest(c, protectedConfigKeyMessage)
return
}
if err := createSystemConfig(c.Request.Context(), req); err != nil {
if err.Error() == ConfigKeyExists {
response.AbortBadRequest(c, ConfigKeyExists)
return
}
if err.Error() == protectedConfigKeyMessage {
response.AbortBadRequest(c, protectedConfigKeyMessage)
return
}
response.AbortInternal(c, err.Error())
return
}
@@ -155,6 +176,10 @@ func UpdateSystemConfig(c *gin.Context) {
}
key := c.Param("key")
if isProtectedConfigKey(key) {
response.AbortBadRequest(c, protectedConfigKeyMessage)
return
}
if err := updateSystemConfig(c.Request.Context(), key, req); err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
response.AbortNotFound(c, SystemConfigNotFound)
internal/apps/admin/system_config/routers_test.go
+3 -3
View File
@@ -27,7 +27,7 @@ import (
"github.com/Rain-kl/Wavelet/internal/shared/response"
)
const expectedDefaultConfigsCount = 34
const expectedDefaultConfigsCount = 37
func setupTestRouter(authUser *model.User) *gin.Engine {
r := testhelper.NewTestGinEngine()
@@ -168,8 +168,8 @@ func TestListSystemConfigs(t *testing.T) {
var configs []model.SystemConfig
_ = json.Unmarshal(dataBytes, &configs)
if len(configs) != 5 {
t.Errorf("expected 5 business configs, got %d: %v", len(configs), configs)
if len(configs) != 8 {
t.Errorf("expected 8 business configs, got %d: %v", len(configs), configs)
}
})
}
internal/apps/openflare/async_tasks.go
+21 -62
View File
@@ -21,11 +21,6 @@ const (
// TaskTypeSSLRenew is the admin task type for SSL renewal.
TaskTypeSSLRenew = "of_ssl_renew"
// DatabaseAutoCleanupTask prunes observability tables by retention policy.
DatabaseAutoCleanupTask = "openflare:database_auto_cleanup"
// TaskTypeDatabaseAutoCleanup is the admin task type for observability cleanup.
TaskTypeDatabaseAutoCleanup = "of_database_auto_cleanup"
// WAFIPGroupSyncTask syncs due automatic/subscription WAF IP groups.
WAFIPGroupSyncTask = "openflare:waf_ip_group_sync"
// TaskTypeWAFIPGroupSync is the admin task type for WAF IP group sync.
@@ -35,6 +30,11 @@ const (
UptimeKumaSyncTask = "openflare:uptime_kuma_sync"
// TaskTypeUptimeKumaSync is the admin task type for Uptime Kuma sync.
TaskTypeUptimeKumaSync = "of_uptime_kuma_sync"
// LogDBSwitchTask 切换日志数据库任务标识。
LogDBSwitchTask = "openflare:log_db_switch"
// TaskTypeLogDBSwitch is the admin task type for log database switch.
TaskTypeLogDBSwitch = "of_log_db_switch"
)
var (
@@ -54,18 +54,6 @@ var SSLRenewMeta = task.TaskMeta{
Retryable: true,
}
// DatabaseAutoCleanupMeta describes the observability auto-cleanup task.
var DatabaseAutoCleanupMeta = task.TaskMeta{
Type: TaskTypeDatabaseAutoCleanup,
AsynqTask: DatabaseAutoCleanupTask,
Name: "OpenFlare 可观测数据自动清理",
Description: "按保留天数清理访问日志、性能快照与请求聚合数据",
SupportsTime: false,
MaxRetry: task.DefaultMaxRetry,
Queue: task.QueueDefault,
Retryable: true,
}
// WAFIPGroupSyncMeta describes the WAF IP group sync task.
var WAFIPGroupSyncMeta = task.TaskMeta{
Type: TaskTypeWAFIPGroupSync,
@@ -90,6 +78,22 @@ var UptimeKumaSyncMeta = task.TaskMeta{
Retryable: true,
}
// LogDBSwitchMeta 描述切换日志数据库任务。
var LogDBSwitchMeta = task.TaskMeta{
Type: TaskTypeLogDBSwitch,
AsynqTask: LogDBSwitchTask,
Name: "切换日志数据库",
Description: "复制迁移日志数据并在成功后切换日志主库（期间禁止日志写入）",
SupportsTime: false,
MaxRetry: task.DefaultMaxRetry,
Queue: task.QueueDefault,
Retryable: true,
Params: []task.TaskParam{
{Name: "target", Label: "目标日志库", Type: "string", Required: true,
Placeholder: "postgres|sqlite|clickhouse", Description: "迁移目标：postgres（主库为 PG 时）、sqlite（主库为 SQLite 时）或 clickhouse"},
},
}
// SSLRenewHandler renews due TLS certificates.
type SSLRenewHandler struct{}
@@ -105,51 +109,6 @@ func (h *SSLRenewHandler) Execute(ctx context.Context, _ []byte) (*task.TaskResu
return &task.TaskResult{Message: msg}, nil
}
// DatabaseAutoCleanupHandler prunes observability data when auto-cleanup is enabled.
type DatabaseAutoCleanupHandler struct{}
// Execute runs retention-based cleanup for all observability targets.
func (h *DatabaseAutoCleanupHandler) Execute(ctx context.Context, _ []byte) (*task.TaskResult, error) {
// 从 SystemConfig 读取自动清理配置
enabled, _ := repository.GetBoolByKey(ctx, model.ConfigKeyDatabaseAutoCleanupEnabled)
if !enabled {
msg := "自动清理未启用，跳过执行"
task.AppendLog(ctx, "%s", msg)
return &task.TaskResult{Message: msg}, nil
}
retentionDays, _ := repository.GetIntByKey(ctx, model.ConfigKeyDatabaseAutoCleanupRetentionDays)
if retentionDays <= 0 {
retentionDays = 30 // 默认保留 30 天
}
task.AppendLog(ctx, "开始执行可观测数据自动清理，保留天数=%d", retentionDays)
summary, err := tasks.RunDatabaseAutoCleanupOnce(ctx, time.Now())
if err != nil {
task.AppendLog(ctx, "可观测数据自动清理失败: %v", err)
return nil, err
}
if summary == nil {
msg := "自动清理未启用，跳过执行"
task.AppendLog(ctx, "%s", msg)
return &task.TaskResult{Message: msg}, nil
}
var totalDeleted int64
for _, item := range summary.Results {
totalDeleted += item.DeletedCount
task.AppendLog(ctx, "清理 %s：删除 %d 条", item.TargetLabel, item.DeletedCount)
}
msg := fmt.Sprintf(
"可观测数据自动清理完成，保留 %d 天，共删除 %d 条",
summary.RetentionDays,
totalDeleted,
)
task.AppendLog(ctx, "%s", msg)
return &task.TaskResult{Message: msg}, nil
}
// WAFIPGroupSyncHandler syncs due WAF IP groups to agents.
type WAFIPGroupSyncHandler struct{}
internal/apps/openflare/async_tasks_test.go
-58
View File
@@ -6,7 +6,6 @@ package openflare
import (
"context"
"testing"
"time"
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
"github.com/Rain-kl/Wavelet/internal/model"
@@ -17,63 +16,6 @@ import (
"gorm.io/gorm"
)
func TestDatabaseAutoCleanupHandlerSkipsWhenDisabled(t *testing.T) {
sqliteDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{
DisableForeignKeyConstraintWhenMigrating: true,
})
require.NoError(t, err)
require.NoError(t, sqliteDB.AutoMigrate(&model.SystemConfig{}))
db.SetDB(sqliteDB)
t.Cleanup(func() { db.SetDB(nil) })
ctx := context.Background()
require.NoError(t, repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyDatabaseAutoCleanupEnabled, "false"))
result, err := (&DatabaseAutoCleanupHandler{}).Execute(ctx, nil)
require.NoError(t, err)
require.NotNil(t, result)
assert.Contains(t, result.Message, "未启用")
}
func TestDatabaseAutoCleanupHandlerDeletesRowsWhenEnabled(t *testing.T) {
sqliteDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{
DisableForeignKeyConstraintWhenMigrating: true,
})
require.NoError(t, err)
require.NoError(t, sqliteDB.AutoMigrate(&model.SystemConfig{}))
db.SetDB(sqliteDB)
resetAccessLogStore := repository.SetAccessLogStoreForTest(repository.NewMemoryAccessLogStore())
resetObservabilityStore := repository.SetObservabilityStoreForTest(repository.NewMemoryObservabilityStore())
t.Cleanup(func() {
resetObservabilityStore()
resetAccessLogStore()
db.SetDB(nil)
})
ctx := context.Background()
now := time.Now().UTC()
require.NoError(t, repository.InsertOpenFlareAccessLogsBatch(ctx, []*model.OpenFlareAccessLog{{
NodeID: "node-a",
LoggedAt: now.Add(-95 * 24 * time.Hour),
RemoteAddr: "203.0.113.10",
Host: "example.com",
Path: "/access",
StatusCode: 200,
}}))
require.NoError(t, repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyDatabaseAutoCleanupEnabled, "true"))
require.NoError(t, repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyDatabaseAutoCleanupRetentionDays, "1"))
result, err := (&DatabaseAutoCleanupHandler{}).Execute(ctx, nil)
require.NoError(t, err)
require.NotNil(t, result)
assert.Contains(t, result.Message, "共删除")
rows, err := repository.ListOpenFlareAccessLogs(ctx, model.OpenFlareAccessLogQuery{Page: 0, PageSize: 10})
require.NoError(t, err)
assert.Empty(t, rows)
}
func TestUptimeKumaSyncHandlerSkipsWhenDisabled(t *testing.T) {
sqliteDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{
DisableForeignKeyConstraintWhenMigrating: true,
internal/apps/openflare/chwriter/writer.go
+102 -20
View File
@@ -11,13 +11,10 @@ import (
"sync"
"time"
"github.com/Rain-kl/Wavelet/internal/repository"
"github.com/Rain-kl/Wavelet/internal/infra/config"
"github.com/Rain-kl/Wavelet/internal/infra/persistence/batchwriter"
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
"github.com/Rain-kl/Wavelet/internal/platform/lifecycle"
analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
"github.com/Rain-kl/Wavelet/internal/repository/logstore"
"github.com/Rain-kl/Wavelet/pkg/logger"
)
@@ -56,12 +53,10 @@ var (
frpcDedup *dedupSet
)
// Init starts OpenFlare ClickHouse batch writers. Safe to call multiple times.
// Init starts OpenFlare log batch writers. Safe to call multiple times.
// Writers always initialize regardless of ClickHouse.enabled; the active log
// store is resolved via logstore at flush time (PG/SQLite when CH is not active).
func Init(ctx context.Context) {
if !config.Config.ClickHouse.Enabled {
return
}
initOnce.Do(func() {
metricSnapshotDedup = newDedupSet()
edgeHealthDedup = newDedupSet()
@@ -70,25 +65,25 @@ func Init(ctx context.Context) {
metricSnapshotWriter = mustNewObservabilityWriter(
"metric_snapshots",
withFlushRetries(analyticsrepo.BatchInsertNodeMetricSnapshots),
withFlushRetries(flushNodeMetricSnapshots),
metricSnapshotDedup,
metricSnapshotKey,
)
edgeHealthWriter = mustNewObservabilityWriter(
"edge_health",
withFlushRetries(analyticsrepo.BatchInsertNodeEdgeHealth),
withFlushRetries(flushNodeEdgeHealth),
edgeHealthDedup,
edgeHealthKey,
)
frpsWriter = mustNewObservabilityWriter(
"frps_obs",
withFlushRetries(analyticsrepo.BatchInsertNodeObsFrps),
withFlushRetries(flushNodeObsFrps),
frpsDedup,
frpsKey,
)
frpcWriter = mustNewObservabilityWriter(
"frpc_obs",
withFlushRetries(analyticsrepo.BatchInsertNodeObsFrpc),
withFlushRetries(flushNodeObsFrpc),
frpcDedup,
frpcKey,
)
@@ -129,6 +124,51 @@ func Stop(ctx context.Context) error {
return firstErr
}
// Drain 等待所有 OpenFlare 日志 writer 的在途批次落库：轮询队列 Depth 归零后
// 再保持一个最大 flush 周期（observabilityFlushEvery）持续为空才返回；
// 不停止 writer（迁移冻结后由 ensureWritable 拒绝新写入）。未初始化时直接返回 nil。
func Drain(ctx context.Context) error {
return drainWriters(ctx, WriterStats, observabilityFlushEvery)
}
// drainWriters 轮询 stats 直至所有队列 Depth=0 并持续 quietPeriod 无新积压。
func drainWriters(ctx context.Context, stats func() []batchwriter.Stats, quietPeriod time.Duration) error {
if !running() {
return nil
}
ticker := time.NewTicker(drainPollInterval)
defer ticker.Stop()
var quietSince time.Time
for {
if allDepthZero(stats()) {
if quietSince.IsZero() {
quietSince = time.Now()
} else if time.Since(quietSince) >= quietPeriod {
return nil
}
} else {
quietSince = time.Time{}
}
select {
case <-ctx.Done():
return ctx.Err()
case <-ticker.C:
}
}
}
// drainPollInterval 队列轮询间隔。
const drainPollInterval = 50 * time.Millisecond
func allDepthZero(stats []batchwriter.Stats) bool {
for _, s := range stats {
if s.Depth > 0 {
return false
}
}
return true
}
// WriterStats returns queue depth and failure counters for all OpenFlare writers.
func WriterStats() []batchwriter.Stats {
writers := []statsProvider{
@@ -237,7 +277,7 @@ func mustNewNodeAccessLogWriter() *batchwriter.Writer[analyticsmodel.NodeAccessL
}
writer, err := batchwriter.New[analyticsmodel.NodeAccessLog](
cfg,
withFlushRetries(analyticsrepo.BatchInsertNodeAccessLogs),
withFlushRetries(flushNodeAccessLogs),
batchwriter.WithDropHandler[analyticsmodel.NodeAccessLog](func(item analyticsmodel.NodeAccessLog) {
logger.WarnF(context.Background(), "[OpenFlare] node access log queue full, dropping log for node %s path %s", item.NodeID, item.Path)
}),
@@ -280,17 +320,59 @@ func withFlushRetries[T any](flush batchwriter.FlushFunc[T]) batchwriter.FlushFu
}
func wireModelInsertHooks() {
repository.SetObservabilityInsertHooks(repository.ObservabilityInsertHooks{
QueueMetricSnapshot: QueueMetricSnapshot,
QueueEdgeHealth: QueueEdgeHealth,
QueueFrpsObservation: QueueFrpsObservation,
QueueFrpcObservation: QueueFrpcObservation,
logstore.SetObservabilityHooks(logstore.ObservabilityHooks{
QueueMetricSnapshot: QueueMetricSnapshot,
QueueEdgeHealth: QueueEdgeHealth,
QueueNodeObsFrps: QueueFrpsObservation,
QueueNodeObsFrpc: QueueFrpcObservation,
})
repository.SetAccessLogInsertHooks(repository.AccessLogInsertHooks{
logstore.SetAccessLogHooks(logstore.AccessLogHooks{
QueueNodeAccessLogs: QueueNodeAccessLogs,
})
}
// 以下 flush 函数作为 batchwriter 的落库目标：激活库由 logstore 在 flush 时决定。
func flushNodeMetricSnapshots(ctx context.Context, rows []analyticsmodel.NodeMetricSnapshot) error {
s, err := logstore.Active(ctx)
if err != nil {
return err
}
return s.Observability.BatchInsertNodeMetricSnapshots(ctx, rows)
}
func flushNodeEdgeHealth(ctx context.Context, rows []analyticsmodel.NodeEdgeHealth) error {
s, err := logstore.Active(ctx)
if err != nil {
return err
}
return s.Observability.BatchInsertNodeEdgeHealth(ctx, rows)
}
func flushNodeObsFrps(ctx context.Context, rows []analyticsmodel.NodeObsFrps) error {
s, err := logstore.Active(ctx)
if err != nil {
return err
}
return s.Observability.BatchInsertNodeObsFrps(ctx, rows)
}
func flushNodeObsFrpc(ctx context.Context, rows []analyticsmodel.NodeObsFrpc) error {
s, err := logstore.Active(ctx)
if err != nil {
return err
}
return s.Observability.BatchInsertNodeObsFrpc(ctx, rows)
}
func flushNodeAccessLogs(ctx context.Context, rows []analyticsmodel.NodeAccessLog) error {
s, err := logstore.Active(ctx)
if err != nil {
return err
}
return s.AccessLogs.BatchInsertNodeAccessLogs(ctx, rows)
}
func metricSnapshotKey(snapshot analyticsmodel.NodeMetricSnapshot) string {
return fmt.Sprintf("%s|%d", snapshot.NodeID, snapshot.CapturedAt.UTC().UnixNano())
}
internal/apps/openflare/dashboard/logics_test.go
+2 -4
View File
@@ -9,6 +9,7 @@ import (
"time"
"github.com/Rain-kl/Wavelet/internal/repository"
"github.com/Rain-kl/Wavelet/internal/testhelper"
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
"github.com/Rain-kl/Wavelet/internal/model"
@@ -26,11 +27,8 @@ func setupDashboardTestDB(t *testing.T) func() {
require.NoError(t, sqliteDB.AutoMigrate(&model.OpenFlareNode{}))
db.SetDB(sqliteDB)
resetAccessLogStore := repository.SetAccessLogStoreForTest(repository.NewMemoryAccessLogStore())
resetObservabilityStore := repository.SetObservabilityStoreForTest(repository.NewMemoryObservabilityStore())
testhelper.SetupLogStoresForTest(t)
return func() {
resetObservabilityStore()
resetAccessLogStore()
db.SetDB(nil)
}
}
internal/apps/openflare/integration/agent_protocol_test.go
+1 -4
View File
@@ -40,15 +40,12 @@ func setupProtocolTestEnv(t *testing.T) (*gin.Engine, func()) {
db.SetDB(sqliteDB)
agent.ResetAuthCacheForTest()
resetAccessLogStore := repository.SetAccessLogStoreForTest(repository.NewMemoryAccessLogStore())
resetObservabilityStore := repository.SetObservabilityStoreForTest(repository.NewMemoryObservabilityStore())
testhelper.SetupLogStoresForTest(t)
engine := testhelper.NewTestGinEngine()
mountOpenFlareTestRoutes(engine)
cleanup := func() {
resetObservabilityStore()
resetAccessLogStore()
db.SetDB(nil)
agent.ResetAuthCacheForTest()
}
internal/apps/openflare/node/logics_test.go
+2 -4
View File
@@ -15,6 +15,7 @@ import (
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/repository"
"github.com/Rain-kl/Wavelet/internal/testhelper"
"github.com/glebarez/sqlite"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -41,12 +42,9 @@ func setupNodeTestDB(t *testing.T) func() {
))
db.SetDB(sqliteDB)
resetAccessLogStore := repository.SetAccessLogStoreForTest(repository.NewMemoryAccessLogStore())
resetObservabilityStore := repository.SetObservabilityStoreForTest(repository.NewMemoryObservabilityStore())
testhelper.SetupLogStoresForTest(t)
return func() {
resetObservabilityStore()
resetAccessLogStore()
db.SetDB(nil)
}
}
internal/apps/openflare/observability/access_log_logics.go
+4 -4
View File
@@ -11,7 +11,7 @@ import (
"github.com/Rain-kl/Wavelet/internal/repository"
"github.com/Rain-kl/Wavelet/internal/model"
analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
"github.com/Rain-kl/Wavelet/pkg/logger"
)
@@ -448,9 +448,9 @@ func buildAccessLogUADistributions(
for _, row := range uaRows {
ua := row.Key
count := row.Value
deviceAcc[analyticsrepo.ParseDeviceType(ua)] += count
browserAcc[analyticsrepo.ParseBrowserName(ua)] += count
osAcc[analyticsrepo.ParseOSName(ua)] += count
deviceAcc[analyticsmodel.ParseDeviceType(ua)] += count
browserAcc[analyticsmodel.ParseBrowserName(ua)] += count
osAcc[analyticsmodel.ParseOSName(ua)] += count
}
topUserAgents = make([]DistributionItem, 0, accessLogOverviewTopLimit)
internal/apps/openflare/option/logics.go
-43
View File
@@ -10,7 +10,6 @@ import (
"strings"
"github.com/Rain-kl/Wavelet/internal/apps/openflare/geoip"
oftasks "github.com/Rain-kl/Wavelet/internal/apps/openflare/tasks"
"github.com/Rain-kl/Wavelet/internal/apps/openflare/uptimekuma"
"github.com/Rain-kl/Wavelet/internal/buildinfo"
"github.com/Rain-kl/Wavelet/internal/model"
@@ -50,22 +49,6 @@ type geoIPLookupView struct {
Longitude *float64 `json:"longitude,omitempty"`
}
type databaseCleanupInput struct {
Target string `json:"target"`
RetentionDays *int `json:"retention_days"`
}
type databaseCleanupResult struct {
Target string `json:"target"`
TargetLabel string `json:"target_label"`
DeletedCount int64 `json:"deleted_count"`
EligibleCount int64 `json:"eligible_count,omitempty"`
CleanupMode string `json:"cleanup_mode,omitempty"`
TableTTLDays int `json:"table_ttl_days,omitempty"`
DeleteAll bool `json:"delete_all"`
RetentionDays *int `json:"retention_days,omitempty"`
}
type optionBatchPayload struct {
Options []model.OpenFlareOption `json:"options"`
}
@@ -183,32 +166,6 @@ func lookupGeoIP(_ context.Context, provider, rawIP string) (*geoIPLookupView, e
}, nil
}
func cleanupDatabaseObservability(ctx context.Context, input databaseCleanupInput) (*databaseCleanupResult, error) {
target := strings.TrimSpace(input.Target)
if target == "" {
return nil, errors.New(errInvalidParams)
}
result, err := oftasks.CleanupDatabaseObservability(ctx, oftasks.DatabaseCleanupInput{
Target: target,
RetentionDays: input.RetentionDays,
})
if err != nil {
return nil, err
}
return &databaseCleanupResult{
Target: result.Target,
TargetLabel: result.TargetLabel,
DeletedCount: result.DeletedCount,
EligibleCount: result.EligibleCount,
CleanupMode: result.CleanupMode,
TableTTLDays: result.TableTTLDays,
DeleteAll: result.DeleteAll,
RetentionDays: result.RetentionDays,
}, nil
}
func syncUptimeKuma(ctx context.Context) error {
return uptimekuma.SyncToUptimeKuma(ctx)
}
internal/apps/openflare/option/logics_test.go
-53
View File
@@ -6,7 +6,6 @@ package option
import (
"context"
"testing"
"time"
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
"github.com/Rain-kl/Wavelet/internal/model"
@@ -117,55 +116,3 @@ func TestLookupGeoIPDisabledProvider(t *testing.T) {
assert.Equal(t, "disabled", view.Provider)
assert.Equal(t, "8.8.8.8", view.IP)
}
func TestCleanupDatabaseObservabilityDeletesRows(t *testing.T) {
cleanup := setupOptionTestDB(t)
defer cleanup()
ctx := context.Background()
resetAccessLogStore := repository.SetAccessLogStoreForTest(repository.NewMemoryAccessLogStore())
defer resetAccessLogStore()
now := time.Now().UTC()
require.NoError(t, repository.InsertOpenFlareAccessLogsBatch(ctx, []*model.OpenFlareAccessLog{
{
NodeID: "node-a",
LoggedAt: now.Add(-10 * 24 * time.Hour),
RemoteAddr: "203.0.113.1",
Host: "example.com",
Path: "/old",
StatusCode: 200,
},
{
NodeID: "node-a",
LoggedAt: now.Add(-2 * time.Hour),
RemoteAddr: "203.0.113.2",
Host: "example.com",
Path: "/recent",
StatusCode: 200,
},
}))
// Retention shorter than table TTL (90d for access logs) must be rejected.
shortRetention := 7
_, err := cleanupDatabaseObservability(ctx, databaseCleanupInput{
Target: "node_access_logs",
RetentionDays: &shortRetention,
})
require.Error(t, err)
// Full truncate still hard-deletes all rows.
result, err := cleanupDatabaseObservability(ctx, databaseCleanupInput{
Target: "node_access_logs",
})
require.NoError(t, err)
assert.Equal(t, "node_access_logs", result.Target)
assert.Equal(t, "访问日志", result.TargetLabel)
assert.Equal(t, int64(2), result.DeletedCount)
assert.True(t, result.DeleteAll)
assert.Equal(t, "truncate", result.CleanupMode)
rows, err := repository.ListOpenFlareAccessLogs(ctx, model.OpenFlareAccessLogQuery{Page: 0, PageSize: 10})
require.NoError(t, err)
assert.Empty(t, rows)
}
internal/apps/openflare/option/routers.go
-37
View File
@@ -4,9 +4,6 @@
package option
import (
"encoding/json"
"errors"
"io"
"net/http"
"github.com/Rain-kl/Wavelet/internal/apps/openflare/apiutil"
@@ -128,33 +125,6 @@ func LookupGeoIPHandler(c *gin.Context) {
c.JSON(http.StatusOK, response.OK(view))
}
// CleanupDatabaseHandler 清理可观测性数据库数据。
// @Summary 清理可观测性数据库
// @Description 按目标与保留天数清理可观测性相关数据表，需要管理员权限
// @Tags openflare-option
// @Accept json
// @Produce json
// @Security SessionCookie
// @Param request body option.databaseCleanupInput false "清理参数"
// @Success 200 {object} response.Any{data=option.databaseCleanupResult} "清理结果"
// @Failure 400 {object} response.Any "参数错误"
// @Failure 401 {object} response.Any "未登录"
// @Failure 404 {object} response.Any "无权限或不存在"
// @Failure 500 {object} response.Any "内部错误"
// @Router /api/v1/d/option/database/cleanup [post]
func CleanupDatabaseHandler(c *gin.Context) {
var input databaseCleanupInput
if err := bindOptionalJSON(c.Request.Body, &input); err != nil {
response.AbortBadRequest(c, errInvalidParams)
return
}
result, err := cleanupDatabaseObservability(c.Request.Context(), input)
if apiutil.AbortBadRequestOnError(c, err) {
return
}
c.JSON(http.StatusOK, response.OK(result))
}
// SyncUptimeKumaHandler 同步 Uptime Kuma 监控。
// @Summary 同步 Uptime Kuma
// @Description 将 OpenFlare 节点同步到 Uptime Kuma，需要管理员权限
@@ -174,10 +144,3 @@ func SyncUptimeKumaHandler(c *gin.Context) {
}
c.JSON(http.StatusOK, response.OK("同步成功"))
}
func bindOptionalJSON(body io.Reader, target any) error {
if err := json.NewDecoder(body).Decode(target); err != nil && !errors.Is(err, io.EOF) {
return err
}
return nil
}
internal/apps/openflare/option/validate.go
+17 -5
View File
@@ -27,6 +27,17 @@ var (
const optionValueTrue = "true"
// protectedConfigKeyMessage 命中受保护 key 时返回给管理员的业务错误文案。
const protectedConfigKeyMessage = "该配置项由系统任务管理，禁止手动修改"
// protectedConfigKeys 仅允许内部（迁移任务/bootstrap）写入的 key。
var protectedConfigKeys = map[string]bool{
model.ConfigKeyLogDatabase: true,
model.ConfigKeyLogDBMigration: true,
}
func isProtectedConfigKey(key string) bool { return protectedConfigKeys[key] }
func buildOptionValidationState(ctx context.Context, options []model.OpenFlareOption) map[string]string {
// 从 SystemConfig 读取所有业务配置构建状态
configs, err := repository.ListAdminSystemConfigs(ctx, "business")
@@ -53,7 +64,7 @@ func validateOptionWithState(ctx context.Context, option model.OpenFlareOption,
if err := validateGeoIPOption(option.Key, option.Value); err != nil {
return err
}
if err := validateDatabaseCleanupOption(option.Key, option.Value); err != nil {
if err := validateLogRetentionOption(option.Key, option.Value); err != nil {
return err
}
if err := validateAgentOption(option.Key, option.Value); err != nil {
@@ -100,11 +111,9 @@ func validateGeoIPOption(key, value string) error {
return fmt.Errorf("%s 仅支持 disabled、mmdb、ip-api、geojs、ipinfo", key)
}
func validateDatabaseCleanupOption(key, value string) error {
func validateLogRetentionOption(key, value string) error {
switch key {
case model.ConfigKeyDatabaseAutoCleanupEnabled:
return validateBooleanOption(key, value)
case model.ConfigKeyDatabaseAutoCleanupRetentionDays:
case model.ConfigKeyLogRetentionDaysPostgres, model.ConfigKeyLogRetentionDaysSQLite, model.ConfigKeyLogRetentionDaysClickHouse:
intValue, err := strconv.Atoi(value)
if err != nil || intValue < 1 {
return fmt.Errorf("%s 必须为大于等于 1 的整数天", key)
@@ -221,6 +230,9 @@ func validateOptions(ctx context.Context, options []model.OpenFlareOption) error
if strings.TrimSpace(option.Key) == "" {
return errors.New(errInvalidParams)
}
if isProtectedConfigKey(option.Key) {
return errors.New(protectedConfigKeyMessage)
}
if err := validateOptionWithState(ctx, option, state); err != nil {
return err
}
internal/apps/openflare/relay/logics_test.go
+2 -2
View File
@@ -10,6 +10,7 @@ import (
"time"
"github.com/Rain-kl/Wavelet/internal/repository"
"github.com/Rain-kl/Wavelet/internal/testhelper"
"github.com/Rain-kl/Wavelet/internal/apps/openflare/agent"
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
@@ -38,10 +39,9 @@ func setupRelayTestDB(t *testing.T) func() {
db.SetDB(sqliteDB)
agent.ResetAuthCacheForTest()
resetObservabilityStore := repository.SetObservabilityStoreForTest(repository.NewMemoryObservabilityStore())
testhelper.SetupLogStoresForTest(t)
return func() {
resetObservabilityStore()
db.SetDB(nil)
agent.ResetAuthCacheForTest()
}
internal/apps/openflare/tasks/database_cleanup.go
-245
View File
@@ -1,245 +0,0 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package tasks
import (
"context"
"errors"
"fmt"
"strings"
"time"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/repository"
analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
)
const (
// DatabaseCleanupTargetAccessLogs is the API cleanup target for access logs.
DatabaseCleanupTargetAccessLogs = "node_access_logs"
// DatabaseCleanupTargetMetricSnapshots is the API cleanup target for metric snapshots.
DatabaseCleanupTargetMetricSnapshots = "node_metric_snapshots"
// DatabaseCleanupTargetEdgeHealth is the API cleanup target for OpenResty edge health (connections).
DatabaseCleanupTargetEdgeHealth = "node_edge_health"
// DatabaseCleanupTargetObsFrps is the API cleanup target for FRPS observations.
DatabaseCleanupTargetObsFrps = "node_obs_frps"
// DatabaseCleanupTargetObsFrpc is the API cleanup target for FRPC observations.
DatabaseCleanupTargetObsFrpc = "node_obs_frpc"
)
var databaseCleanupTargets = map[string]string{
DatabaseCleanupTargetAccessLogs: "访问日志",
DatabaseCleanupTargetMetricSnapshots: "性能快照",
DatabaseCleanupTargetEdgeHealth: "OpenResty 健康(连接)",
DatabaseCleanupTargetObsFrps: "FRPS 观测",
DatabaseCleanupTargetObsFrpc: "FRPC 观测",
}
// databaseCleanupTableTTLDays maps API targets to ClickHouse DDL TTL days.
var databaseCleanupTableTTLDays = map[string]int{
DatabaseCleanupTargetAccessLogs: analyticsrepo.TableTTLDaysNodeAccessLogs,
DatabaseCleanupTargetMetricSnapshots: analyticsrepo.TableTTLDaysNodeMetricSnapshots,
DatabaseCleanupTargetEdgeHealth: analyticsrepo.TableTTLDaysNodeObs,
DatabaseCleanupTargetObsFrps: analyticsrepo.TableTTLDaysNodeObs,
DatabaseCleanupTargetObsFrpc: analyticsrepo.TableTTLDaysNodeObs,
}
// DatabaseCleanupInput describes a manual observability cleanup request.
type DatabaseCleanupInput struct {
Target string `json:"target"`
RetentionDays *int `json:"retention_days"`
}
// DatabaseCleanupResult summarizes a manual observability cleanup run.
//
// Semantics:
// - delete_all / cleanup_mode=truncate: DeletedCount is hard-deleted rows (TRUNCATE).
// - retention path / cleanup_mode=ttl_materialize: DeletedCount is always 0;
// EligibleCount estimates rows past the table DDL TTL (not an arbitrary younger cutoff).
type DatabaseCleanupResult struct {
Target string `json:"target"`
TargetLabel string `json:"target_label"`
DeletedCount int64 `json:"deleted_count"`
EligibleCount int64 `json:"eligible_count,omitempty"`
CleanupMode string `json:"cleanup_mode,omitempty"`
TableTTLDays int `json:"table_ttl_days,omitempty"`
DeleteAll bool `json:"delete_all"`
RetentionDays *int `json:"retention_days,omitempty"`
Cutoff *time.Time `json:"cutoff,omitempty"`
}
// DatabaseAutoCleanupSummary summarizes a scheduled auto-cleanup run.
type DatabaseAutoCleanupSummary struct {
RetentionDays int `json:"retention_days"`
ExecutedAt time.Time `json:"executed_at"`
Results []DatabaseCleanupResult `json:"results"`
}
// TableTTLDaysForCleanupTarget returns the DDL TTL days for a cleanup target.
func TableTTLDaysForCleanupTarget(target string) (int, bool) {
days, ok := databaseCleanupTableTTLDays[strings.TrimSpace(target)]
return days, ok
}
// CleanupDatabaseObservability deletes observability rows for the given target.
//
// When RetentionDays is nil, rows are hard-deleted via TRUNCATE.
// When RetentionDays is set, ClickHouse only force-materializes the table TTL policy:
// retention_days shorter than the table TTL is rejected (do not fake success).
func CleanupDatabaseObservability(ctx context.Context, input DatabaseCleanupInput) (*DatabaseCleanupResult, error) {
target := strings.TrimSpace(input.Target)
targetLabel, ok := databaseCleanupTargets[target]
if !ok {
return nil, errors.New("unsupported cleanup target")
}
if input.RetentionDays != nil && *input.RetentionDays <= 0 {
return nil, errors.New("retention_days 必须为大于 0 的整数")
}
tableTTLDays := databaseCleanupTableTTLDays[target]
result := &DatabaseCleanupResult{
Target: target,
TargetLabel: targetLabel,
DeleteAll: input.RetentionDays == nil,
TableTTLDays: tableTTLDays,
}
if input.RetentionDays == nil {
deleted, mode, err := deleteAllObservabilityRows(ctx, target)
if err != nil {
return nil, err
}
result.DeletedCount = deleted
result.EligibleCount = deleted
result.CleanupMode = mode
return result, nil
}
retentionDays := *input.RetentionDays
if retentionDays < tableTTLDays {
return nil, fmt.Errorf(
"retention_days 不能小于表 TTL（%d 天）；ClickHouse 仅支持按表 TTL 物化过期，更短保留请使用清空全部或调整 DDL",
tableTTLDays,
)
}
// MATERIALIZE TTL only enforces DDL policy; cutoff reported is the table TTL boundary.
tableCutoff := time.Now().UTC().Add(-time.Duration(tableTTLDays) * 24 * time.Hour)
eligible, mode, err := materializeObservabilityTableTTL(ctx, target)
if err != nil {
return nil, err
}
result.DeletedCount = 0
result.EligibleCount = eligible
result.CleanupMode = mode
result.RetentionDays = &retentionDays
result.Cutoff = &tableCutoff
return result, nil
}
// RunDatabaseAutoCleanupOnce runs retention-based cleanup for all observability targets.
//
// Configured retention shorter than a target's table TTL is clamped up to the table TTL
// so the scheduled job can force-materialize each table policy without failing.
func RunDatabaseAutoCleanupOnce(ctx context.Context, now time.Time) (*DatabaseAutoCleanupSummary, error) {
enabled, err := repository.GetBoolByKey(ctx, model.ConfigKeyDatabaseAutoCleanupEnabled)
if err != nil {
return nil, fmt.Errorf("failed to read database_auto_cleanup_enabled: %w", err)
}
if !enabled {
return nil, nil
}
retentionDays, err := repository.GetIntByKey(ctx, model.ConfigKeyDatabaseAutoCleanupRetentionDays)
if err != nil || retentionDays <= 0 {
// Use default value 30 if config read fails or value is invalid
retentionDays = 30
}
results := make([]DatabaseCleanupResult, 0, len(databaseCleanupTargets))
for _, target := range []string{
DatabaseCleanupTargetAccessLogs,
DatabaseCleanupTargetMetricSnapshots,
DatabaseCleanupTargetEdgeHealth,
DatabaseCleanupTargetObsFrps,
DatabaseCleanupTargetObsFrpc,
} {
effectiveDays := retentionDays
if ttl, ok := databaseCleanupTableTTLDays[target]; ok && effectiveDays < ttl {
effectiveDays = ttl
}
result, err := CleanupDatabaseObservability(ctx, DatabaseCleanupInput{
Target: target,
RetentionDays: &effectiveDays,
})
if err != nil {
return nil, err
}
results = append(results, *result)
}
return &DatabaseAutoCleanupSummary{
RetentionDays: retentionDays,
ExecutedAt: now.UTC(),
Results: results,
}, nil
}
func deleteAllObservabilityRows(ctx context.Context, target string) (int64, string, error) {
var (
deleted int64
err error
)
switch target {
case DatabaseCleanupTargetAccessLogs:
deleted, err = repository.DeleteAllOpenFlareAccessLogs(ctx)
case DatabaseCleanupTargetMetricSnapshots:
deleted, err = repository.DeleteAllOpenFlareMetricSnapshots(ctx)
case DatabaseCleanupTargetEdgeHealth:
deleted, err = repository.DeleteAllOpenFlareEdgeHealth(ctx)
case DatabaseCleanupTargetObsFrps:
deleted, err = repository.DeleteAllOpenFlareNodeObservationFrps(ctx)
case DatabaseCleanupTargetObsFrpc:
deleted, err = repository.DeleteAllOpenFlareNodeObservationFrpc(ctx)
default:
return 0, "", errors.New("unsupported cleanup target")
}
if err != nil {
return 0, "", err
}
return deleted, analyticsrepo.CleanupModeTruncate, nil
}
// materializeObservabilityTableTTL triggers table-TTL materialize (or memory-store delete-before
// with the table TTL cutoff for tests) and returns the eligible/estimate row count.
func materializeObservabilityTableTTL(ctx context.Context, target string) (int64, string, error) {
ttlDays, ok := databaseCleanupTableTTLDays[target]
if !ok {
return 0, "", errors.New("unsupported cleanup target")
}
cutoff := time.Now().UTC().Add(-time.Duration(ttlDays) * 24 * time.Hour)
var (
eligible int64
err error
)
switch target {
case DatabaseCleanupTargetAccessLogs:
eligible, err = repository.DeleteOpenFlareAccessLogsBefore(ctx, cutoff)
case DatabaseCleanupTargetMetricSnapshots:
eligible, err = repository.DeleteOpenFlareMetricSnapshotsBefore(ctx, cutoff)
case DatabaseCleanupTargetEdgeHealth:
eligible, err = repository.DeleteOpenFlareEdgeHealthBefore(ctx, cutoff)
case DatabaseCleanupTargetObsFrps:
eligible, err = repository.DeleteOpenFlareNodeObservationFrpsBefore(ctx, cutoff)
case DatabaseCleanupTargetObsFrpc:
eligible, err = repository.DeleteOpenFlareNodeObservationFrpcBefore(ctx, cutoff)
default:
return 0, "", errors.New("unsupported cleanup target")
}
if err != nil {
return 0, "", err
}
return eligible, analyticsrepo.CleanupModeTTLMaterialize, nil
}
internal/apps/openflare/tasks/database_cleanup_test.go
-208
View File
@@ -1,208 +0,0 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package tasks
import (
"context"
"testing"
"time"
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/repository"
analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
"github.com/glebarez/sqlite"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"gorm.io/gorm"
)
func setupDatabaseCleanupTestDB(t *testing.T) context.Context {
t.Helper()
sqliteDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{
DisableForeignKeyConstraintWhenMigrating: true,
})
require.NoError(t, err)
require.NoError(t, sqliteDB.AutoMigrate(&model.SystemConfig{}))
db.SetDB(sqliteDB)
resetAccessLogStore := repository.SetAccessLogStoreForTest(repository.NewMemoryAccessLogStore())
resetObservabilityStore := repository.SetObservabilityStoreForTest(repository.NewMemoryObservabilityStore())
t.Cleanup(func() {
resetObservabilityStore()
resetAccessLogStore()
db.SetDB(nil)
})
return context.Background()
}
func TestCleanupDatabaseObservabilityRejectsRetentionShorterThanTableTTL(t *testing.T) {
ctx := setupDatabaseCleanupTestDB(t)
retentionDays := 7 // metric snapshots DDL TTL is 30 days
result, err := CleanupDatabaseObservability(ctx, DatabaseCleanupInput{
Target: DatabaseCleanupTargetMetricSnapshots,
RetentionDays: &retentionDays,
})
require.Error(t, err)
assert.Nil(t, result)
assert.Contains(t, err.Error(), "不能小于表 TTL")
assert.Contains(t, err.Error(), "30")
}
func TestCleanupDatabaseObservabilityRejectsAccessLogRetentionShorterThanTableTTL(t *testing.T) {
ctx := setupDatabaseCleanupTestDB(t)
retentionDays := 30 // access logs DDL TTL is 90 days
result, err := CleanupDatabaseObservability(ctx, DatabaseCleanupInput{
Target: DatabaseCleanupTargetAccessLogs,
RetentionDays: &retentionDays,
})
require.Error(t, err)
assert.Nil(t, result)
assert.Contains(t, err.Error(), "90")
}
func TestCleanupDatabaseObservabilityMaterializeDoesNotClaimHardDelete(t *testing.T) {
ctx := setupDatabaseCleanupTestDB(t)
now := time.Now().UTC()
// One row past metric table TTL (30d), one still inside the window.
require.NoError(t, repository.InsertOpenFlareMetricSnapshot(ctx, &model.OpenFlareMetricSnapshot{
NodeID: "node-a",
CapturedAt: now.Add(-40 * 24 * time.Hour),
CPUUsagePercent: 10,
}))
require.NoError(t, repository.InsertOpenFlareMetricSnapshot(ctx, &model.OpenFlareMetricSnapshot{
NodeID: "node-a",
CapturedAt: now.Add(-12 * time.Hour),
CPUUsagePercent: 20,
}))
retentionDays := analyticsrepo.TableTTLDaysNodeMetricSnapshots
result, err := CleanupDatabaseObservability(ctx, DatabaseCleanupInput{
Target: DatabaseCleanupTargetMetricSnapshots,
RetentionDays: &retentionDays,
})
require.NoError(t, err)
assert.False(t, result.DeleteAll)
assert.Equal(t, analyticsrepo.CleanupModeTTLMaterialize, result.CleanupMode)
assert.Equal(t, analyticsrepo.TableTTLDaysNodeMetricSnapshots, result.TableTTLDays)
// MATERIALIZE is not a counted hard delete.
assert.Equal(t, int64(0), result.DeletedCount)
assert.Equal(t, int64(1), result.EligibleCount)
require.NotNil(t, result.Cutoff)
assert.True(t, result.Cutoff.Before(now.Add(-29*24*time.Hour)))
// Memory store applies the table-TTL cutoff for tests; only the recent row remains.
rows, err := repository.ListOpenFlareMetricSnapshotsSince(ctx, "", time.Time{}, 0)
require.NoError(t, err)
require.Len(t, rows, 1)
assert.Equal(t, float64(20), rows[0].CPUUsagePercent)
}
func TestCleanupDatabaseObservabilityDeletesAllRowsWhenRetentionMissing(t *testing.T) {
ctx := setupDatabaseCleanupTestDB(t)
now := time.Now().UTC()
require.NoError(t, repository.InsertOpenFlareAccessLogsBatch(ctx, []*model.OpenFlareAccessLog{
{
NodeID: "node-a",
LoggedAt: now.Add(-3 * time.Hour),
RemoteAddr: "203.0.113.1",
Host: "example.com",
Path: "/one",
StatusCode: 200,
},
{
NodeID: "node-a",
LoggedAt: now.Add(-2 * time.Hour),
RemoteAddr: "203.0.113.2",
Host: "example.com",
Path: "/two",
StatusCode: 502,
},
}))
result, err := CleanupDatabaseObservability(ctx, DatabaseCleanupInput{
Target: DatabaseCleanupTargetAccessLogs,
})
require.NoError(t, err)
assert.True(t, result.DeleteAll)
assert.Equal(t, analyticsrepo.CleanupModeTruncate, result.CleanupMode)
assert.Equal(t, int64(2), result.DeletedCount)
assert.Equal(t, int64(2), result.EligibleCount)
rows, err := repository.ListOpenFlareAccessLogs(ctx, model.OpenFlareAccessLogQuery{Page: 0, PageSize: 10})
require.NoError(t, err)
assert.Empty(t, rows)
}
func TestRunDatabaseAutoCleanupOnceClampsRetentionToTableTTL(t *testing.T) {
ctx := setupDatabaseCleanupTestDB(t)
now := time.Now().UTC()
// Access logs TTL=90d, metrics TTL=30d. Config retention=1 must clamp, not reject.
require.NoError(t, repository.InsertOpenFlareAccessLogsBatch(ctx, []*model.OpenFlareAccessLog{{
NodeID: "node-a",
LoggedAt: now.Add(-100 * 24 * time.Hour),
RemoteAddr: "203.0.113.10",
Host: "example.com",
Path: "/access",
StatusCode: 200,
}}))
require.NoError(t, repository.InsertOpenFlareMetricSnapshot(ctx, &model.OpenFlareMetricSnapshot{
NodeID: "node-a",
CapturedAt: now.Add(-40 * 24 * time.Hour),
CPUUsagePercent: 10,
}))
require.NoError(t, repository.InsertOpenFlareEdgeHealth(ctx, &model.OpenFlareEdgeHealth{
NodeID: "node-a",
CapturedAt: now.Add(-40 * 24 * time.Hour),
Status: "healthy",
Connections: 2,
}))
require.NoError(t, repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyDatabaseAutoCleanupEnabled, "true"))
require.NoError(t, repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyDatabaseAutoCleanupRetentionDays, "1"))
summary, err := RunDatabaseAutoCleanupOnce(ctx, now)
require.NoError(t, err)
require.NotNil(t, summary)
require.Len(t, summary.Results, 5)
assert.Equal(t, 1, summary.RetentionDays)
for _, result := range summary.Results {
assert.Equal(t, analyticsrepo.CleanupModeTTLMaterialize, result.CleanupMode)
assert.Equal(t, int64(0), result.DeletedCount, "target %s must not claim hard delete", result.Target)
assert.GreaterOrEqual(t, result.TableTTLDays, 30)
require.NotNil(t, result.RetentionDays)
assert.GreaterOrEqual(t, *result.RetentionDays, result.TableTTLDays)
}
accessLogs, err := repository.ListOpenFlareAccessLogs(ctx, model.OpenFlareAccessLogQuery{Page: 0, PageSize: 10})
require.NoError(t, err)
assert.Empty(t, accessLogs)
metricSnapshots, err := repository.ListOpenFlareMetricSnapshotsSince(ctx, "", time.Time{}, 0)
require.NoError(t, err)
assert.Empty(t, metricSnapshots)
edgeHealth, err := repository.ListOpenFlareEdgeHealth(ctx, "", time.Time{}, 0)
require.NoError(t, err)
assert.Empty(t, edgeHealth)
}
func TestTableTTLDaysForCleanupTarget(t *testing.T) {
days, ok := TableTTLDaysForCleanupTarget(DatabaseCleanupTargetAccessLogs)
require.True(t, ok)
assert.Equal(t, 90, days)
days, ok = TableTTLDaysForCleanupTarget(DatabaseCleanupTargetMetricSnapshots)
require.True(t, ok)
assert.Equal(t, 30, days)
_, ok = TableTTLDaysForCleanupTarget("unknown")
assert.False(t, ok)
}
internal/apps/openflare/tasks/log_db_switch.go
+402
View File
@@ -0,0 +1,402 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package tasks
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/Rain-kl/Wavelet/internal/apps/openflare/chwriter"
"github.com/Rain-kl/Wavelet/internal/apps/risk_control"
"github.com/Rain-kl/Wavelet/internal/infra/config"
"github.com/Rain-kl/Wavelet/internal/infra/task"
"github.com/Rain-kl/Wavelet/internal/model"
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
"github.com/Rain-kl/Wavelet/internal/repository"
"github.com/Rain-kl/Wavelet/internal/repository/logstore"
"github.com/Rain-kl/Wavelet/pkg/logger"
)
const copyBatchSize = 1000
// 迁移目标库名常量（normalizeTarget 归一化后的取值）。
const (
targetPostgres = "postgres"
targetSQLite = "sqlite"
targetClickHouse = "clickhouse"
)
type logDBSwitchPayload struct {
Target string `json:"target"`
}
// LogDBSwitchHandler 切换日志数据库任务处理器。
type LogDBSwitchHandler struct{}
// ValidatePayload 校验并规范化参数。
func (h *LogDBSwitchHandler) ValidatePayload(payload []byte) ([]byte, error) {
var p logDBSwitchPayload
if err := json.Unmarshal(payload, &p); err != nil {
return nil, fmt.Errorf("参数解析失败: %w", err)
}
p.Target = normalizeTarget(p.Target)
if !validTarget(p.Target) {
return nil, fmt.Errorf("目标日志库不合法: %s", p.Target)
}
out, err := json.Marshal(p)
if err != nil {
return nil, err
}
return out, nil
}
func normalizeTarget(v string) string {
switch v {
case targetPostgres, "postgresql":
return targetPostgres
case targetSQLite, "sqlite3":
return targetSQLite
case targetClickHouse, "ch":
return targetClickHouse
}
return v
}
func validTarget(v string) bool {
return v == targetPostgres || v == targetSQLite || v == targetClickHouse
}
// Execute 执行迁移。
func (h *LogDBSwitchHandler) Execute(ctx context.Context, payload []byte) (*task.TaskResult, error) {
var p logDBSwitchPayload
if err := json.Unmarshal(payload, &p); err != nil {
return nil, fmt.Errorf("参数解析失败: %w", err)
}
p.Target = normalizeTarget(p.Target)
if err := validateSwitch(ctx, p.Target); err != nil {
return nil, err
}
source, err := currentLogDatabase(ctx)
if err != nil {
task.AppendLog(ctx, "读取日志主库失败: %v", err)
return nil, err
}
task.AppendLog(ctx, "开始切换日志数据库：%s -> %s", source, p.Target)
// 设置迁移冻结标记（置位后由 ensureWritable 拒绝新写入）。
if err := setMigrationFlag(ctx, "migrating"); err != nil {
return nil, err
}
// 失败也清除（SaveOrUpdateSystemConfig 会失效 RAM 缓存并广播），保持源库可写。
defer func() {
if err := setMigrationFlag(ctx, ""); err != nil {
logger.ErrorF(ctx, "清除日志迁移冻结标记失败: %v", err)
}
}()
// 冻结标记置位后再排空在途批次（chwriter + 用户访问日志 writer），
// 保证排空完成后不再有新批次进入源库。
if err := drainLogWriters(ctx); err != nil {
return nil, fmt.Errorf("排空日志写入队列失败: %w", err)
}
src, err := logstore.Active(ctx)
if err != nil {
return nil, err
}
dst, err := buildTargetStore(ctx, p.Target)
if err != nil {
return nil, err
}
// 清空目标库日志表（幂等重试前提）。
if err := clearTargetLogTables(ctx, dst); err != nil {
return nil, err
}
// PG 目标：按源库时间范围预建分区，避免历史数据复制报 "no partition of relation found"。
if err := ensureTargetPartitions(ctx, src, dst, p.Target); err != nil {
return nil, err
}
// 逐表复制（6 张日志表）。
if err := copyAccessLogs(ctx, src, dst); err != nil {
return nil, err
}
if err := copyUserAccessLogs(ctx, src, dst); err != nil {
return nil, err
}
if err := copyObservability(ctx, src, dst); err != nil {
return nil, err
}
// 翻转主库标记。
if err := flipLogDatabase(ctx, p.Target); err != nil {
return nil, err
}
task.AppendLog(ctx, "日志数据库已切换为 %s，写入恢复", p.Target)
return &task.TaskResult{Message: fmt.Sprintf("日志数据库已从 %s 切换为 %s", source, p.Target)}, nil
}
func validateSwitch(ctx context.Context, target string) error {
source, err := currentLogDatabase(ctx)
if err != nil {
return err
}
if source == target {
return errors.New("目标日志库与当前日志库相同，无需迁移")
}
switch target {
case "clickhouse":
if !config.Config.ClickHouse.Enabled {
return errors.New("ClickHouse 未启用，无法迁移到 ClickHouse")
}
case "postgres":
if !config.Config.Database.Enabled {
return errors.New("PostgreSQL 未启用（当前主库为 SQLite），无法迁移到 PostgreSQL")
}
case "sqlite":
if config.Config.Database.Enabled {
return errors.New("当前主库为 PostgreSQL，日志库不能设置为 SQLite")
}
}
return nil
}
func currentLogDatabase(ctx context.Context) (string, error) {
cfg, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeyLogDatabase)
if err != nil {
return "", fmt.Errorf("读取日志主库失败: %w", err)
}
if cfg.Value == "" {
return "", errors.New("日志主库配置为空")
}
return cfg.Value, nil
}
// drainLogWriters 等待 chwriter（节点访问日志 + 可观测 4 表）与 risk_control
// （用户访问日志）的在途批次全部落库。见设计 §7.2：先排空再冻结。
func drainLogWriters(ctx context.Context) error {
if err := chwriter.Drain(ctx); err != nil {
return err
}
return risk_control.DrainLogWriter(ctx)
}
// setMigrationFlag 写入迁移冻结标记。用 SaveOrUpdateSystemConfig：行缺失时 upsert，
// 并失效 RAM 缓存 + 广播其他节点，保证 logstore.Migrating/resolveDatabase 立即生效。
func setMigrationFlag(ctx context.Context, v string) error {
return repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyLogDBMigration, v)
}
// flipLogDatabase 翻转日志主库。同上用 SaveOrUpdateSystemConfig，确保各进程缓存失效后指向新库。
func flipLogDatabase(ctx context.Context, target string) error {
return repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyLogDatabase, target)
}
// buildTargetStore 构造目标库 Store（不经过 Active 缓存，直接 Build）。
// 迁移期间冻结标记已置位，目标库的清空/复制写入必须放行，故使用 BuildForMigration。
func buildTargetStore(ctx context.Context, database string) (*logstore.Store, error) {
return logstore.BuildForMigration(ctx, database)
}
func clearTargetLogTables(ctx context.Context, dst *logstore.Store) error {
// 依次清空 6 张表：AccessLogs.DeleteAll、UserAccessLogs.DeleteAll、Observability.DeleteAll*
// （SQLite/PG 用 DeleteAll；CH 用 TRUNCATE 语义）。
if _, err := dst.AccessLogs.DeleteAll(ctx); err != nil {
return fmt.Errorf("清空目标访问日志失败: %w", err)
}
if _, err := dst.UserAccessLogs.DeleteAll(ctx); err != nil {
return fmt.Errorf("清空目标用户访问日志失败: %w", err)
}
for _, fn := range []func(context.Context) (int64, error){
dst.Observability.DeleteAllMetricSnapshots,
dst.Observability.DeleteAllEdgeHealth,
dst.Observability.DeleteAllNodeObservationFrps,
dst.Observability.DeleteAllNodeObservationFrpc,
} {
if _, err := fn(ctx); err != nil {
return err
}
}
return nil
}
// ensureTargetPartitions 目标为 PG 时，按源库时间范围（两表合并）预建分区，
// 否则复制历史数据会报 "no partition of relation found"；目标非 PG 为 no-op。
func ensureTargetPartitions(ctx context.Context, src, dst *logstore.Store, target string) error {
if target != targetPostgres {
return nil
}
from, to, err := migrationRange(ctx, src)
if err != nil {
return err
}
if from.IsZero() || to.IsZero() {
task.AppendLog(ctx, "源库无日志数据，跳过分区预建")
return nil
}
if err := dst.AccessLogs.EnsurePartitions(ctx, from, to.AddDate(0, 1, 0)); err != nil {
return fmt.Errorf("预建目标 PG 分区失败: %w", err)
}
task.AppendLog(ctx, "已为目标 PG 预建分区 %s ~ %s", from.Format("2006-01"), to.Format("2006-01"))
return nil
}
// migrationRange 合并源库节点访问日志（logged_at）与用户访问日志（created_at）
// 的最小/最大时间；任一表为空时忽略该表。
func migrationRange(ctx context.Context, src *logstore.Store) (time.Time, time.Time, error) {
fromAccess, toAccess, err := src.AccessLogs.MigrationRange(ctx)
if err != nil {
return time.Time{}, time.Time{}, fmt.Errorf("读取源访问日志时间范围失败: %w", err)
}
fromUser, toUser, err := src.UserAccessLogs.MigrationRange(ctx)
if err != nil {
return time.Time{}, time.Time{}, fmt.Errorf("读取源用户访问日志时间范围失败: %w", err)
}
return minTime(fromAccess, fromUser), maxTime(toAccess, toUser), nil
}
func minTime(a, b time.Time) time.Time {
switch {
case a.IsZero():
return b
case b.IsZero():
return a
case a.Before(b):
return a
default:
return b
}
}
func maxTime(a, b time.Time) time.Time {
switch {
case a.IsZero():
return b
case b.IsZero():
return a
case a.After(b):
return a
default:
return b
}
}
// copyAccessLogs 从 src 复制节点访问日志到 dst。
func copyAccessLogs(ctx context.Context, src, dst *logstore.Store) error {
// 注意：迁移期间 src 已冻结，但复制读取不受冻结影响；每批按 id 升序扫描。
var lastID uint64
for {
rows, err := listNodeAccessLogsByID(ctx, src, lastID, copyBatchSize)
if err != nil {
return err
}
if len(rows) == 0 {
break
}
if err := dst.AccessLogs.BatchInsertNodeAccessLogs(ctx, rows); err != nil {
return fmt.Errorf("写入目标访问日志失败(批 %d): %w", lastID, err)
}
task.AppendLog(ctx, "已复制访问日志 %d 条（截至 id=%d）", len(rows), rows[len(rows)-1].ID)
lastID = rows[len(rows)-1].ID
if len(rows) < copyBatchSize {
break
}
}
return nil
}
func listNodeAccessLogsByID(ctx context.Context, src *logstore.Store, afterID uint64, limit int) ([]analyticsmodel.NodeAccessLog, error) {
return src.AccessLogs.ListForMigration(ctx, afterID, limit)
}
// copyUserAccessLogs 从 src 复制用户访问日志到 dst（按 id 升序分批）。
func copyUserAccessLogs(ctx context.Context, src, dst *logstore.Store) error {
var lastID uint64
for {
rows, err := src.UserAccessLogs.ListForMigration(ctx, lastID, copyBatchSize)
if err != nil {
return err
}
if len(rows) == 0 {
return nil
}
if err := dst.UserAccessLogs.BatchInsert(ctx, rows); err != nil {
return fmt.Errorf("写入目标用户访问日志失败(批 %d): %w", lastID, err)
}
lastID = rows[len(rows)-1].ID
task.AppendLog(ctx, "已复制用户访问日志 %d 条（截至 id=%d）", len(rows), lastID)
if len(rows) < copyBatchSize {
return nil
}
}
}
// copyObservability 复制 4 张可观测表，每张表按 id 升序分批复制，
// 以每批最后一条 id 作为下一批游标（不使用 len 近似）。
func copyObservability(ctx context.Context, src, dst *logstore.Store) error {
if err := copyObsTable(ctx, "metric_snapshots",
src.Observability.ListMetricSnapshotsForMigration,
dst.Observability.BatchInsertNodeMetricSnapshots,
lastMetricSnapshotID); err != nil {
return err
}
if err := copyObsTable(ctx, "edge_health",
src.Observability.ListEdgeHealthForMigration,
dst.Observability.BatchInsertNodeEdgeHealth,
lastEdgeHealthID); err != nil {
return err
}
if err := copyObsTable(ctx, "obs_frps",
src.Observability.ListNodeObsFrpsForMigration,
dst.Observability.BatchInsertNodeObsFrps,
lastObsFrpsID); err != nil {
return err
}
if err := copyObsTable(ctx, "obs_frpc",
src.Observability.ListNodeObsFrpcForMigration,
dst.Observability.BatchInsertNodeObsFrpc,
lastObsFrpcID); err != nil {
return err
}
return nil
}
// copyObsTable 按 id 升序分批复制单张可观测表；idOf 返回批内最后一条 id。
func copyObsTable[T any](ctx context.Context, name string,
list func(context.Context, uint64, int) ([]T, error),
insert func(context.Context, []T) error,
idOf func([]T) uint64,
) error {
var lastID uint64
for {
rows, err := list(ctx, lastID, copyBatchSize)
if err != nil {
return fmt.Errorf("复制 %s 失败: %w", name, err)
}
if len(rows) == 0 {
return nil
}
if err := insert(ctx, rows); err != nil {
return fmt.Errorf("复制 %s 失败: %w", name, err)
}
lastID = idOf(rows)
task.AppendLog(ctx, "已复制 %s %d 条（截至 id=%d）", name, len(rows), lastID)
if len(rows) < copyBatchSize {
return nil
}
}
}
func lastMetricSnapshotID(rows []analyticsmodel.NodeMetricSnapshot) uint64 {
return rows[len(rows)-1].ID
}
func lastEdgeHealthID(rows []analyticsmodel.NodeEdgeHealth) uint64 { return rows[len(rows)-1].ID }
func lastObsFrpsID(rows []analyticsmodel.NodeObsFrps) uint64 { return rows[len(rows)-1].ID }
func lastObsFrpcID(rows []analyticsmodel.NodeObsFrpc) uint64 { return rows[len(rows)-1].ID }
internal/apps/openflare/tasks/log_db_switch_test.go
+358
View File
@@ -0,0 +1,358 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package tasks
import (
"context"
"encoding/json"
"errors"
"fmt"
"sync/atomic"
"testing"
"time"
"github.com/glebarez/sqlite"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"gorm.io/gorm"
"gorm.io/gorm/logger"
"github.com/Rain-kl/Wavelet/internal/infra/config"
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
"github.com/Rain-kl/Wavelet/internal/model"
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
"github.com/Rain-kl/Wavelet/internal/repository"
"github.com/Rain-kl/Wavelet/internal/repository/logstore"
)
var logDBSwitchDBSeq int64
// newLogDBSwitchDB 构造内存 sqlite 库（含日志 5 表 + 系统配置表）。
func newLogDBSwitchDB(t *testing.T) *gorm.DB {
t.Helper()
dsn := fmt.Sprintf("file:log-db-switch-%d?mode=memory&cache=shared", atomic.AddInt64(&logDBSwitchDBSeq, 1))
gdb, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
require.NoError(t, err)
require.NoError(t, gdb.AutoMigrate(
&model.SystemConfig{},
&analyticsmodel.NodeAccessLog{},
&analyticsmodel.NodeMetricSnapshot{},
&analyticsmodel.NodeEdgeHealth{},
&analyticsmodel.NodeObsFrps{},
&analyticsmodel.NodeObsFrpc{},
&analyticsmodel.UserAccessLog{},
))
return gdb
}
// TestCopyAccessLogsPreservesIDs sqlite→sqlite 模拟：源 store 3 条，目标空库，
// copyAccessLogs 后 ID 保留、数量一致。
func TestCopyAccessLogsPreservesIDs(t *testing.T) {
oldDB, oldCH := config.Config.Database.Enabled, config.Config.ClickHouse.Enabled
config.Config.Database.Enabled, config.Config.ClickHouse.Enabled = false, false
t.Cleanup(func() {
config.Config.Database.Enabled, config.Config.ClickHouse.Enabled = oldDB, oldCH
})
logstore.ResetForTest()
defer logstore.ResetForTest()
ctx := context.Background()
srcDB := newLogDBSwitchDB(t)
dstDB := newLogDBSwitchDB(t)
db.SetDB(srcDB)
src, err := logstore.Active(ctx) // 无 reader 时按 seed 规则解析为 sqlite
require.NoError(t, err)
db.SetDB(dstDB)
dst, err := logstore.BuildForMigration(ctx, "sqlite")
require.NoError(t, err)
t.Cleanup(func() { db.SetDB(nil) })
now := time.Now().UTC()
rows := []analyticsmodel.NodeAccessLog{
{ID: 101, NodeID: "n1", LoggedAt: now, RemoteAddr: "1.1.1.1", Host: "a.example.com", Path: "/"},
{ID: 202, NodeID: "n2", LoggedAt: now, RemoteAddr: "2.2.2.2", Host: "b.example.com", Path: "/x"},
{ID: 303, NodeID: "n1", LoggedAt: now, RemoteAddr: "3.3.3.3", Host: "c.example.com", Path: "/y"},
}
require.NoError(t, src.AccessLogs.BatchInsertNodeAccessLogs(ctx, rows))
require.NoError(t, copyAccessLogs(ctx, src, dst))
var got []analyticsmodel.NodeAccessLog
require.NoError(t, dstDB.Order("id ASC").Find(&got).Error)
require.Len(t, got, 3)
for i, wantID := range []uint64{101, 202, 303} {
assert.Equal(t, wantID, got[i].ID, "row %d id preserved", i)
}
assert.Equal(t, "n1", got[0].NodeID)
assert.Equal(t, "n2", got[1].NodeID)
assert.Equal(t, "n1", got[2].NodeID)
assert.Equal(t, "1.1.1.1", got[0].RemoteAddr)
// 源库保持不变。
var srcCount int64
require.NoError(t, srcDB.Model(&analyticsmodel.NodeAccessLog{}).Count(&srcCount).Error)
assert.Equal(t, int64(3), srcCount)
}
// TestCopyUserAccessLogsPreservesIDs sqlite→sqlite 模拟：源库用户访问日志按 id 升序
// 复制到目标库，ID 保留、数量一致，且源库保持不变。
func TestCopyUserAccessLogsPreservesIDs(t *testing.T) {
oldDB, oldCH := config.Config.Database.Enabled, config.Config.ClickHouse.Enabled
config.Config.Database.Enabled, config.Config.ClickHouse.Enabled = false, false
t.Cleanup(func() {
config.Config.Database.Enabled, config.Config.ClickHouse.Enabled = oldDB, oldCH
})
logstore.ResetForTest()
defer logstore.ResetForTest()
ctx := context.Background()
srcDB := newLogDBSwitchDB(t)
dstDB := newLogDBSwitchDB(t)
db.SetDB(srcDB)
src, err := logstore.Active(ctx)
require.NoError(t, err)
db.SetDB(dstDB)
dst, err := logstore.BuildForMigration(ctx, "sqlite")
require.NoError(t, err)
t.Cleanup(func() { db.SetDB(nil) })
now := time.Now().UTC()
rows := []analyticsmodel.UserAccessLog{
{ID: 11, UserID: 1, Path: "/a", CreatedAt: now},
{ID: 22, UserID: 2, Path: "/b", CreatedAt: now.Add(time.Second)},
{ID: 33, UserID: 1, Path: "/c", CreatedAt: now.Add(2 * time.Second)},
}
require.NoError(t, src.UserAccessLogs.BatchInsert(ctx, rows))
require.NoError(t, copyUserAccessLogs(ctx, src, dst))
var got []analyticsmodel.UserAccessLog
require.NoError(t, dstDB.Order("id ASC").Find(&got).Error)
require.Len(t, got, 3)
for i, wantID := range []uint64{11, 22, 33} {
assert.Equal(t, wantID, got[i].ID, "row %d id preserved", i)
}
var srcCount int64
require.NoError(t, srcDB.Model(&analyticsmodel.UserAccessLog{}).Count(&srcCount).Error)
assert.Equal(t, int64(3), srcCount)
}
// TestClearTargetLogTablesClearsUserAccessLogs 验证清空目标包含用户访问日志表
// （6 张日志表之一），迁移「覆盖目标库已有日志」幂等前提成立。
func TestClearTargetLogTablesClearsUserAccessLogs(t *testing.T) {
oldDB, oldCH := config.Config.Database.Enabled, config.Config.ClickHouse.Enabled
config.Config.Database.Enabled, config.Config.ClickHouse.Enabled = false, false
t.Cleanup(func() {
config.Config.Database.Enabled, config.Config.ClickHouse.Enabled = oldDB, oldCH
})
logstore.ResetForTest()
defer logstore.ResetForTest()
ctx := context.Background()
dstDB := newLogDBSwitchDB(t)
db.SetDB(dstDB)
t.Cleanup(func() { db.SetDB(nil) })
dst, err := logstore.BuildForMigration(ctx, "sqlite")
require.NoError(t, err)
now := time.Now().UTC()
require.NoError(t, dst.UserAccessLogs.BatchInsert(ctx, []analyticsmodel.UserAccessLog{
{ID: 1, UserID: 1, Path: "/a", CreatedAt: now},
{ID: 2, UserID: 2, Path: "/b", CreatedAt: now},
}))
require.NoError(t, clearTargetLogTables(ctx, dst))
var count int64
require.NoError(t, dstDB.Model(&analyticsmodel.UserAccessLog{}).Count(&count).Error)
assert.Zero(t, count, "用户访问日志应被清空")
}
// TestValidateSwitch 各非法组合报错。
func TestValidateSwitch(t *testing.T) {
oldDB, oldCH := config.Config.Database.Enabled, config.Config.ClickHouse.Enabled
t.Cleanup(func() {
config.Config.Database.Enabled, config.Config.ClickHouse.Enabled = oldDB, oldCH
})
gdb := newLogDBSwitchDB(t)
db.SetDB(gdb)
t.Cleanup(func() { db.SetDB(nil) })
ctx := context.Background()
setLogDB := func(v string) {
require.NoError(t, repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyLogDatabase, v))
}
t.Run("same target rejected", func(t *testing.T) {
setLogDB("sqlite")
config.Config.Database.Enabled, config.Config.ClickHouse.Enabled = false, false
err := validateSwitch(ctx, "sqlite")
require.Error(t, err)
assert.Contains(t, err.Error(), "相同")
})
t.Run("clickhouse disabled rejected", func(t *testing.T) {
setLogDB("sqlite")
config.Config.Database.Enabled, config.Config.ClickHouse.Enabled = false, false
err := validateSwitch(ctx, "clickhouse")
require.Error(t, err)
assert.Contains(t, err.Error(), "ClickHouse 未启用")
})
t.Run("postgres requires main db enabled", func(t *testing.T) {
setLogDB("sqlite")
config.Config.Database.Enabled, config.Config.ClickHouse.Enabled = false, false
err := validateSwitch(ctx, "postgres")
require.Error(t, err)
assert.Contains(t, err.Error(), "PostgreSQL 未启用")
})
t.Run("sqlite rejected when main db is postgres", func(t *testing.T) {
setLogDB("postgres")
config.Config.Database.Enabled, config.Config.ClickHouse.Enabled = true, false
err := validateSwitch(ctx, "sqlite")
require.Error(t, err)
assert.Contains(t, err.Error(), "SQLite")
})
t.Run("valid postgres migration", func(t *testing.T) {
setLogDB("sqlite")
config.Config.Database.Enabled, config.Config.ClickHouse.Enabled = true, false
require.NoError(t, validateSwitch(ctx, "postgres"))
})
}
// TestLogDBSwitchValidatePayload 参数归一化与非法值拒绝。
func TestLogDBSwitchValidatePayload(t *testing.T) {
h := &LogDBSwitchHandler{}
cases := []struct {
name string
in string
want string
ok bool
}{
{name: "postgresql normalized", in: `{"target":"postgresql"}`, want: "postgres", ok: true},
{name: "sqlite3 normalized", in: `{"target":"sqlite3"}`, want: "sqlite", ok: true},
{name: "ch normalized", in: `{"target":"ch"}`, want: "clickhouse", ok: true},
{name: "postgres passthrough", in: `{"target":"postgres"}`, want: "postgres", ok: true},
{name: "invalid target", in: `{"target":"mysql"}`, ok: false},
{name: "malformed json", in: `not-json`, ok: false},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
out, err := h.ValidatePayload([]byte(c.in))
if !c.ok {
require.Error(t, err)
return
}
require.NoError(t, err)
var p logDBSwitchPayload
require.NoError(t, json.Unmarshal(out, &p))
assert.Equal(t, c.want, p.Target)
})
}
}
// TestExecuteFailureClearsMigrationFlag 迁移失败后 log_db_migration 冻结标记被清除。
// 在 FRESH DB（不预置 log_db_migration 行）上验证：setMigrationFlag 必须 upsert 建行，
// 且失败后经缓存路径（GetSystemConfigByKey）可观察为空。
func TestExecuteFailureClearsMigrationFlag(t *testing.T) {
oldDB := config.Config.Database.Enabled
config.Config.Database.Enabled = true
t.Cleanup(func() { config.Config.Database.Enabled = oldDB })
logstore.ResetForTest()
defer logstore.ResetForTest()
gdb := newLogDBSwitchDB(t)
db.SetDB(gdb)
t.Cleanup(func() { db.SetDB(nil) })
ctx := context.Background()
// FRESH DB：log_db_migration 行不存在（不预置），log_database 预置为 sqlite。
require.NoError(t, repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyLogDatabase, "sqlite"))
_, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeyLogDBMigration)
require.ErrorIs(t, err, gorm.ErrRecordNotFound)
// configReader 对 log_database 报错，使 logstore.Active 在冻结标记置位后失败。
logstore.SetConfigReader(func(_ context.Context, key string) (string, error) {
if key == model.ConfigKeyLogDatabase {
return "", errors.New("reader error")
}
return "", nil
})
_, err = (&LogDBSwitchHandler{}).Execute(ctx, []byte(`{"target":"postgres"}`))
require.Error(t, err)
assert.Contains(t, err.Error(), "reader error")
// 冻结标记必须被 upsert 持久化（行存在）并经缓存路径可观察为空，源库恢复可写。
cfg, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeyLogDBMigration)
require.NoError(t, err, "setMigrationFlag 应 upsert 创建 log_db_migration 行")
assert.Empty(t, cfg.Value, "失败后冻结标记必须清除，源库保持可写")
assert.False(t, logstore.Migrating(ctx))
}
// TestSetMigrationFlagObservableThroughCache 在 FRESH DB 上验证 setMigrationFlag 写入
// 经缓存路径（logstore.Migrating → repository 读取）实时反映：置位 true、清除 false。
func TestSetMigrationFlagObservableThroughCache(t *testing.T) {
logstore.ResetForTest()
defer logstore.ResetForTest()
gdb := newLogDBSwitchDB(t)
db.SetDB(gdb)
t.Cleanup(func() { db.SetDB(nil) })
ctx := context.Background()
// 按 bootstrap 同款注入 repository 读取，走 RAM 缓存路径。
logstore.SetConfigReader(func(ctx context.Context, key string) (string, error) {
cfg, err := repository.GetSystemConfigByKey(ctx, key)
if err != nil {
return "", err
}
return cfg.Value, nil
})
// FRESH DB：行缺失 → fail-open false。
assert.False(t, logstore.Migrating(ctx))
require.NoError(t, setMigrationFlag(ctx, "migrating"))
assert.True(t, logstore.Migrating(ctx), "置位后缓存路径必须立即观察到 migrating")
require.NoError(t, setMigrationFlag(ctx, ""))
assert.False(t, logstore.Migrating(ctx), "清除后缓存路径必须立即观察到非 migrating")
}
// TestFlipLogDatabaseRefreshesCachedConfig 验证翻转日志主库后缓存路径立即反映新库
// （logstore.ActiveDatabase / GetSystemConfigByKey），防止各进程继续写旧库（split-brain）。
func TestFlipLogDatabaseRefreshesCachedConfig(t *testing.T) {
logstore.ResetForTest()
defer logstore.ResetForTest()
gdb := newLogDBSwitchDB(t)
db.SetDB(gdb)
t.Cleanup(func() { db.SetDB(nil) })
ctx := context.Background()
logstore.SetConfigReader(func(ctx context.Context, key string) (string, error) {
cfg, err := repository.GetSystemConfigByKey(ctx, key)
if err != nil {
return "", err
}
return cfg.Value, nil
})
require.NoError(t, repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyLogDatabase, "sqlite"))
active, err := logstore.ActiveDatabase(ctx)
require.NoError(t, err)
assert.Equal(t, "sqlite", active) // 预热缓存
require.NoError(t, flipLogDatabase(ctx, "postgres"))
active, err = logstore.ActiveDatabase(ctx)
require.NoError(t, err)
assert.Equal(t, "postgres", active, "翻转后缓存路径必须立即反映新库")
cfg, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeyLogDatabase)
require.NoError(t, err)
assert.Equal(t, "postgres", cfg.Value)
}
internal/apps/openflare/waf/ip_group_sync_test.go
+2 -2
View File
@@ -12,6 +12,7 @@ import (
"time"
"github.com/Rain-kl/Wavelet/internal/repository"
"github.com/Rain-kl/Wavelet/internal/testhelper"
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
"github.com/Rain-kl/Wavelet/internal/model"
@@ -34,9 +35,8 @@ func setupIPGroupSyncTestDB(t *testing.T) func() {
))
db.SetDB(sqliteDB)
resetAccessLogStore := repository.SetAccessLogStoreForTest(repository.NewMemoryAccessLogStore())
testhelper.SetupLogStoresForTest(t)
return func() {
resetAccessLogStore()
db.SetDB(nil)
}
}
internal/apps/openflare/zone/logics_test.go
+2 -2
View File
@@ -9,6 +9,7 @@ import (
"time"
"github.com/Rain-kl/Wavelet/internal/repository"
"github.com/Rain-kl/Wavelet/internal/testhelper"
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
"github.com/Rain-kl/Wavelet/internal/model"
@@ -61,8 +62,7 @@ func TestLegacyImportUsesEffectiveTLDPlusOne(t *testing.T) {
func TestGetStatsAggregatesZoneHosts(t *testing.T) {
ctx := setupZoneDB(t)
reset := repository.SetAccessLogStoreForTest(repository.NewMemoryAccessLogStore())
t.Cleanup(reset)
testhelper.SetupLogStoresForTest(t)
zone, err := Create(ctx, Input{Domain: "example.com"})
require.NoError(t, err)
internal/apps/risk_control/logics.go
+43 -10
View File
@@ -8,11 +8,10 @@ import (
"sync"
"time"
"github.com/Rain-kl/Wavelet/internal/infra/config"
"github.com/Rain-kl/Wavelet/internal/infra/persistence/batchwriter"
"github.com/Rain-kl/Wavelet/internal/model/analytics"
"github.com/Rain-kl/Wavelet/internal/platform/lifecycle"
analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
"github.com/Rain-kl/Wavelet/internal/repository/logstore"
"github.com/Rain-kl/Wavelet/pkg/logger"
)
@@ -26,12 +25,10 @@ var (
logWriter *batchwriter.Writer[*analytics.UserAccessLog]
)
// InitLogWriter initializes the ClickHouse access-log batch writer.
// InitLogWriter initializes the user access-log batch writer.
// The active log store is resolved via logstore at flush time, so the writer
// runs for PG/SQLite as well as ClickHouse.
func InitLogWriter(ctx context.Context) {
if !config.Config.ClickHouse.Enabled {
return
}
logWriterMu.Lock()
defer logWriterMu.Unlock()
if logWriter != nil {
@@ -49,7 +46,11 @@ func InitLogWriter(ctx context.Context) {
}
rows = append(rows, *item)
}
return analyticsrepo.BatchInsert(ctx, rows)
s, err := logstore.Active(ctx)
if err != nil {
return err
}
return s.UserAccessLogs.BatchInsert(ctx, rows)
},
batchwriter.WithDropHandler[*analytics.UserAccessLog](func(item *analytics.UserAccessLog) {
path := ""
@@ -59,7 +60,7 @@ func InitLogWriter(ctx context.Context) {
logger.WarnF(context.Background(), "[RiskControl] Log queue full, dropping log item for path: %s", path)
}),
batchwriter.WithFlushErrorHandler[*analytics.UserAccessLog](func(ctx context.Context, items []*analytics.UserAccessLog, err error) {
logger.ErrorF(ctx, "[RiskControl] Send ClickHouse batch failed (batch=%d): %v", len(items), err)
logger.ErrorF(ctx, "[RiskControl] Send log batch failed (batch=%d): %v", len(items), err)
}),
)
if err != nil {
@@ -72,7 +73,7 @@ func InitLogWriter(ctx context.Context) {
lifecycle.OnShutdown("risk_control_log_writer", StopLogWriter)
}
// StopLogWriter stops the ClickHouse access-log batch writer and drains pending logs.
// StopLogWriter stops the user access-log batch writer and drains pending logs.
func StopLogWriter(ctx context.Context) error {
writer := currentLogWriter()
if writer == nil {
@@ -81,6 +82,38 @@ func StopLogWriter(ctx context.Context) error {
return writer.Stop(ctx)
}
// DrainLogWriter 等待用户访问日志 writer 的在途批次落库：队列 Depth 归零后
// 再保持一个 flush 周期（1s）持续为空才返回；不停止 writer（迁移冻结后由
// ensureWritable 拒绝新写入）。writer 未初始化时直接返回 nil。
func DrainLogWriter(ctx context.Context) error {
writer := currentLogWriter()
if writer == nil {
return nil
}
ticker := time.NewTicker(drainPollInterval)
defer ticker.Stop()
var quietSince time.Time
for {
if writer.Stats().Depth == 0 {
if quietSince.IsZero() {
quietSince = time.Now()
} else if time.Since(quietSince) >= batchwriter.DefaultConfig().FlushInterval {
return nil
}
} else {
quietSince = time.Time{}
}
select {
case <-ctx.Done():
return ctx.Err()
case <-ticker.C:
}
}
}
// drainPollInterval 用户访问日志队列轮询间隔。
const drainPollInterval = 50 * time.Millisecond
// IsBufferFull reports whether the access-log queue has no remaining capacity.
func IsBufferFull() bool {
writer := currentLogWriter()
internal/apps/risk_control/middleware.go
+5 -3
View File
@@ -13,11 +13,12 @@ import (
"time"
"github.com/Rain-kl/Wavelet/internal/apps/oauth"
"github.com/Rain-kl/Wavelet/internal/infra/config"
"github.com/Rain-kl/Wavelet/internal/infra/persistence/idgen"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/model/analytics"
"github.com/Rain-kl/Wavelet/internal/repository/logstore"
"github.com/Rain-kl/Wavelet/internal/shared/response"
"github.com/Rain-kl/Wavelet/pkg/logger"
"github.com/gin-gonic/gin"
)
@@ -79,8 +80,9 @@ func hashAuditLogSensitiveValue(value string) string {
// RiskControlMiddleware 全局日志采集中间件
func RiskControlMiddleware() gin.HandlerFunc {
return func(c *gin.Context) {
// 如果未启用 ClickHouse，直接放行
if !config.Config.ClickHouse.Enabled {
// 日志库迁移冻结期跳过采集，不阻断业务请求。
if logstore.Migrating(c.Request.Context()) {
logger.WarnF(c.Request.Context(), "[RiskControl] log DB migrating, skip audit log")
c.Next()
return
}
internal/apps/upload/task/cleanup.go
+13
View File
@@ -18,6 +18,7 @@ import (
"github.com/Rain-kl/Wavelet/internal/infra/task"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/repository"
"github.com/Rain-kl/Wavelet/internal/repository/logstore"
"github.com/Rain-kl/Wavelet/pkg/logger"
)
@@ -129,6 +130,18 @@ func (h *SystemCleanupHandler) Execute(ctx context.Context, _ []byte) (*task.Tas
)
}
task.AppendLog(ctx, "开始清理过期日志（按当前日志库保留天数）...")
summary, err := logstore.CleanupExpired(ctx)
switch {
case err != nil:
logger.ErrorF(ctx, "清理过期日志失败: %v", err)
task.AppendLog(ctx, "清理过期日志失败: %v", err)
case summary.Deleted == 0:
task.AppendLog(ctx, "没有需要清理的过期日志 (保留 %d 天)", summary.RetentionDays)
default:
task.AppendLog(ctx, "日志清理完成：保留 %d 天，删除 %d 条", summary.RetentionDays, summary.Deleted)
}
msg := fmt.Sprintf("系统清理完成。成功清理未使用的上传文件 %d/%d 个；清理历史推送审计日志 %d 条；清理任务执行日志 %d 条。",
totalDeleted,
totalProcessed,