mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-06 15:46:37 +08:00
feat(log): decouple log storage from ClickHouse with switchable logstore
- New internal/repository/logstore abstraction: exported domain interfaces (AccessLogStore/ObservabilityStore/UserAccessLogStore/StatusStore), config-driven provider (Active/Build/Migrating/SetConfigReader), GORM implementation for PostgreSQL/SQLite (incl. hourly rollups computed in real time, migration listers, PG partition maintenance), and a ClickHouse wrapper preserving the native batch path; repository facade delegates to logstore; import-lint test enforces apps never import analyticsrepo. - ClickHouse is now optional: the log DB is either the main DB (postgres when database.enabled, else sqlite) or clickhouse; boot validation + first-run seed; log_database / log_db_migration are protected keys. - New user task 切换日志数据库 (of_log_db_switch): freeze log writes, drain batch writers, copy all 6 raw log tables by id (preserving IDs) with target-partition pre-creation for PG, flip log_database on success, clear the freeze flag on failure. - Per-store retention (log_retention_days_*) with expiry cleanup folded into the daily system_cleanup task; legacy database_auto_cleanup_* and of_database_auto_cleanup decommissioned. - goose migrations: 6 log tables in PG (2 monthly-partitioned) + SQLite, retention config seeds, schedule cleanup; GET /api/v1/admin/status/log-database endpoint; frontend retention settings, switch-task UI and status badge; changelog and docs updated. docs(plan): log database decoupling implementation plan docs(design): log database decoupling design (ClickHouse optional)
This commit is contained in:
ryan
@@ -0,0 +1,125 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/model"
|
||||
)
|
||||
|
||||
// CleanupSummary 汇总本次清理结果。
|
||||
type CleanupSummary struct {
|
||||
ActiveDatabase string `json:"active_database"`
|
||||
RetentionDays int `json:"retention_days"`
|
||||
Deleted int64 `json:"deleted"`
|
||||
// Tables 记录本次清理的物理表简写名（去掉 of_ 前缀，如 node_access_logs 对应
|
||||
// of_node_access_logs；CH 侧物理表名相同，简写仅便于状态展示）。
|
||||
Tables []string `json:"tables"`
|
||||
}
|
||||
|
||||
// defaultLogRetentionDays 默认日志保留天数（配置缺失/非法时回退）。
|
||||
const defaultLogRetentionDays = 90
|
||||
|
||||
// partitionLeadMonths 清理时确保「当前月 + 未来 2 个月」分区持续存在。
|
||||
const partitionLeadMonths = 2
|
||||
|
||||
// retentionDaysForActive 按当前激活库读取保留天数（默认 90）。
|
||||
func retentionDaysForActive(ctx context.Context) int {
|
||||
key := model.ConfigKeyLogRetentionDaysPostgres
|
||||
if dbName, _ := resolveDatabase(ctx); dbName == dbNameSQLite {
|
||||
key = model.ConfigKeyLogRetentionDaysSQLite
|
||||
} else if dbName == dbNameClickHouse {
|
||||
key = model.ConfigKeyLogRetentionDaysClickHouse
|
||||
}
|
||||
v, err := getConfig(ctx, key)
|
||||
if err != nil {
|
||||
return defaultLogRetentionDays
|
||||
}
|
||||
days, perr := strconv.Atoi(v)
|
||||
if perr != nil || days <= 0 {
|
||||
return defaultLogRetentionDays
|
||||
}
|
||||
return days
|
||||
}
|
||||
|
||||
// CleanupExpired 按当前激活库保留天数清理过期日志（每日由 system_cleanup 调用）。
|
||||
func CleanupExpired(ctx context.Context) (*CleanupSummary, error) {
|
||||
s, err := Active(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
days := retentionDaysForActive(ctx)
|
||||
cutoff := time.Now().AddDate(0, 0, -days)
|
||||
summary := &CleanupSummary{RetentionDays: days, Tables: []string{}}
|
||||
summary.ActiveDatabase, _ = resolveDatabase(ctx)
|
||||
|
||||
// PG 分区表仅在迁移时预建「当前+2 月」分区，此处确保分区持续存在，
|
||||
// 否则跨月后新写入会报 "no partition of relation found"（SQLite/CH 为 no-op）。
|
||||
now := time.Now().UTC()
|
||||
if err := s.AccessLogs.EnsurePartitions(ctx, now, now.AddDate(0, partitionLeadMonths, 0)); err != nil {
|
||||
return nil, fmt.Errorf("ensure partitions: %w", err)
|
||||
}
|
||||
|
||||
if err := cleanupTable("node_access_logs", func() (int64, error) {
|
||||
return s.AccessLogs.DeleteBefore(ctx, cutoff)
|
||||
}, summary); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := cleanupTable("metric_snapshots", func() (int64, error) {
|
||||
return s.Observability.DeleteMetricSnapshotsBefore(ctx, cutoff)
|
||||
}, summary); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := cleanupTable("edge_health", func() (int64, error) {
|
||||
return s.Observability.DeleteEdgeHealthBefore(ctx, cutoff)
|
||||
}, summary); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := cleanupTable("obs_frps", func() (int64, error) {
|
||||
return s.Observability.DeleteNodeObservationFrpsBefore(ctx, cutoff)
|
||||
}, summary); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := cleanupTable("obs_frpc", func() (int64, error) {
|
||||
return s.Observability.DeleteNodeObservationFrpcBefore(ctx, cutoff)
|
||||
}, summary); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return summary, nil
|
||||
}
|
||||
|
||||
func cleanupTable(name string, fn func() (int64, error), summary *CleanupSummary) error {
|
||||
n, err := fn()
|
||||
if err != nil {
|
||||
return fmt.Errorf("cleanup %s: %w", name, err)
|
||||
}
|
||||
summary.Deleted += n
|
||||
summary.Tables = append(summary.Tables, name)
|
||||
return nil
|
||||
}
|
||||
|
||||
// partitionStatementsRange 生成覆盖 [from, to] 全部月份的两表分区 DDL，
|
||||
// 幂等 CREATE TABLE IF NOT EXISTS ... PARTITION OF ... FOR VALUES FROM ... TO ...。
|
||||
// 入参为任意时间点：按各自所在月份生成，含 from 月与 to 月（to 常用 max+1 月兜底）。
|
||||
func partitionStatementsRange(from, to time.Time) []string {
|
||||
var out []string
|
||||
start := time.Date(from.Year(), from.Month(), 1, 0, 0, 0, 0, time.UTC)
|
||||
end := time.Date(to.Year(), to.Month(), 1, 0, 0, 0, 0, time.UTC).AddDate(0, 1, 0)
|
||||
for ; start.Before(end); start = start.AddDate(0, 1, 0) {
|
||||
monthEnd := start.AddDate(0, 1, 0)
|
||||
suffix := start.Format("200601")
|
||||
fromDay := start.Format("2006-01-02")
|
||||
toDay := monthEnd.Format("2006-01-02")
|
||||
for _, table := range []string{"of_node_access_logs", "w_user_access_logs"} {
|
||||
out = append(out, fmt.Sprintf(
|
||||
"CREATE TABLE IF NOT EXISTS %s_%s PARTITION OF %s FOR VALUES FROM ('%s') TO ('%s')",
|
||||
table, suffix, table, fromDay, toDay))
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,223 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/glebarez/sqlite"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/logger"
|
||||
|
||||
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"
|
||||
)
|
||||
|
||||
// cleanupTestModels 清理涉及的 5 张日志/可观测表。
|
||||
func cleanupTestModels() []any {
|
||||
return []any{
|
||||
&analyticsmodel.NodeAccessLog{},
|
||||
&analyticsmodel.NodeMetricSnapshot{},
|
||||
&analyticsmodel.NodeEdgeHealth{},
|
||||
&analyticsmodel.NodeObsFrps{},
|
||||
&analyticsmodel.NodeObsFrpc{},
|
||||
}
|
||||
}
|
||||
|
||||
// newCleanupTestDB 构造内存 sqlite 库并注入 db.DB（CleanupExpired 经 Active → buildStore 使用）。
|
||||
func newCleanupTestDB(t *testing.T) *gorm.DB {
|
||||
t.Helper()
|
||||
dsn := fmt.Sprintf("file:logstore-cleanup-%d?mode=memory&cache=shared", atomic.AddInt64(&testGormStoreSeq, 1))
|
||||
gdb, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
|
||||
if err != nil {
|
||||
t.Fatalf("open sqlite: %v", err)
|
||||
}
|
||||
if err := gdb.AutoMigrate(cleanupTestModels()...); err != nil {
|
||||
t.Fatalf("automigrate: %v", err)
|
||||
}
|
||||
db.SetDB(gdb)
|
||||
t.Cleanup(func() { db.SetDB(nil) })
|
||||
return gdb
|
||||
}
|
||||
|
||||
// TestCleanupExpiredSQLite 验证 sqlite 激活库的过期日志清理：
|
||||
// 注入 log_retention_days_sqlite=30，40 天前的 5 表记录被删、昨天的保留。
|
||||
func TestCleanupExpiredSQLite(t *testing.T) {
|
||||
ResetForTest()
|
||||
SetConfigReader(func(_ context.Context, key string) (string, error) {
|
||||
switch key {
|
||||
case logDatabaseKey:
|
||||
return "sqlite", nil
|
||||
case model.ConfigKeyLogRetentionDaysSQLite:
|
||||
return "30", nil
|
||||
}
|
||||
return "", nil
|
||||
})
|
||||
defer ResetForTest()
|
||||
|
||||
gdb := newCleanupTestDB(t)
|
||||
ctx := context.Background()
|
||||
old := time.Now().AddDate(0, 0, -40).UTC()
|
||||
recent := time.Now().AddDate(0, 0, -1).UTC()
|
||||
|
||||
if err := gdb.Create([]analyticsmodel.NodeAccessLog{
|
||||
{ID: 1, NodeID: "n1", LoggedAt: old, RemoteAddr: "1.1.1.1"},
|
||||
{ID: 2, NodeID: "n1", LoggedAt: recent, RemoteAddr: "2.2.2.2"},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("seed node access logs: %v", err)
|
||||
}
|
||||
if err := gdb.Create([]analyticsmodel.NodeMetricSnapshot{
|
||||
{ID: 1, NodeID: "n1", CapturedAt: old},
|
||||
{ID: 2, NodeID: "n1", CapturedAt: recent},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("seed metric snapshots: %v", err)
|
||||
}
|
||||
if err := gdb.Create([]analyticsmodel.NodeEdgeHealth{
|
||||
{ID: 1, NodeID: "n1", CapturedAt: old},
|
||||
{ID: 2, NodeID: "n1", CapturedAt: recent},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("seed edge health: %v", err)
|
||||
}
|
||||
if err := gdb.Create([]analyticsmodel.NodeObsFrps{
|
||||
{ID: 1, NodeID: "n1", CapturedAt: old},
|
||||
{ID: 2, NodeID: "n1", CapturedAt: recent},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("seed obs frps: %v", err)
|
||||
}
|
||||
if err := gdb.Create([]analyticsmodel.NodeObsFrpc{
|
||||
{ID: 1, NodeID: "n1", CapturedAt: old},
|
||||
{ID: 2, NodeID: "n1", CapturedAt: recent},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("seed obs frpc: %v", err)
|
||||
}
|
||||
|
||||
summary, err := CleanupExpired(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("CleanupExpired: %v", err)
|
||||
}
|
||||
if summary.ActiveDatabase != "sqlite" {
|
||||
t.Fatalf("ActiveDatabase = %q, want sqlite", summary.ActiveDatabase)
|
||||
}
|
||||
if summary.RetentionDays != 30 {
|
||||
t.Fatalf("RetentionDays = %d, want 30", summary.RetentionDays)
|
||||
}
|
||||
if summary.Deleted != 5 {
|
||||
t.Fatalf("Deleted = %d, want 5", summary.Deleted)
|
||||
}
|
||||
if len(summary.Tables) != 5 {
|
||||
t.Fatalf("Tables = %v, want 5 tables", summary.Tables)
|
||||
}
|
||||
|
||||
assertCount := func(m any, want int64, label string) {
|
||||
t.Helper()
|
||||
var n int64
|
||||
if err := gdb.Model(m).Count(&n).Error; err != nil {
|
||||
t.Fatalf("count %s: %v", label, err)
|
||||
}
|
||||
if n != want {
|
||||
t.Fatalf("%s count = %d, want %d", label, n, want)
|
||||
}
|
||||
}
|
||||
assertCount(&analyticsmodel.NodeAccessLog{}, 1, "node_access_logs")
|
||||
assertCount(&analyticsmodel.NodeMetricSnapshot{}, 1, "metric_snapshots")
|
||||
assertCount(&analyticsmodel.NodeEdgeHealth{}, 1, "edge_health")
|
||||
assertCount(&analyticsmodel.NodeObsFrps{}, 1, "obs_frps")
|
||||
assertCount(&analyticsmodel.NodeObsFrpc{}, 1, "obs_frpc")
|
||||
|
||||
var kept analyticsmodel.NodeAccessLog
|
||||
if err := gdb.First(&kept).Error; err != nil {
|
||||
t.Fatalf("recent node access log missing: %v", err)
|
||||
}
|
||||
if kept.ID != 2 {
|
||||
t.Fatalf("kept log ID = %d, want 2 (recent)", kept.ID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRetentionDaysForActive 覆盖保留天数读取：按激活库选 key、非法值回退默认 90。
|
||||
func TestRetentionDaysForActive(t *testing.T) {
|
||||
ResetForTest()
|
||||
SetConfigReader(func(_ context.Context, key string) (string, error) {
|
||||
switch key {
|
||||
case logDatabaseKey:
|
||||
return "sqlite", nil
|
||||
case model.ConfigKeyLogRetentionDaysSQLite:
|
||||
return "30", nil
|
||||
}
|
||||
return "", nil
|
||||
})
|
||||
if got := retentionDaysForActive(context.Background()); got != 30 {
|
||||
t.Fatalf("retentionDaysForActive = %d, want 30", got)
|
||||
}
|
||||
|
||||
// 非法值（非数字/<=0）回退默认 90。
|
||||
SetConfigReader(func(_ context.Context, key string) (string, error) {
|
||||
switch key {
|
||||
case logDatabaseKey:
|
||||
return "postgres", nil
|
||||
case model.ConfigKeyLogRetentionDaysPostgres:
|
||||
return "abc", nil
|
||||
}
|
||||
return "", nil
|
||||
})
|
||||
if got := retentionDaysForActive(context.Background()); got != 90 {
|
||||
t.Fatalf("retentionDaysForActive invalid value = %d, want 90", got)
|
||||
}
|
||||
|
||||
// reader 报错回退默认 90。
|
||||
SetConfigReader(func(_ context.Context, _ string) (string, error) {
|
||||
return "", fmt.Errorf("boom")
|
||||
})
|
||||
if got := retentionDaysForActive(context.Background()); got != 90 {
|
||||
t.Fatalf("retentionDaysForActive reader error = %d, want 90", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPartitionStatements 验证 PG 分区 DDL 生成：当前月 + 未来 2 个月 × 2 表，
|
||||
// 幂等 PARTITION OF 语句与迁移 SQL 命名一致（含跨年）。
|
||||
func TestPartitionStatements(t *testing.T) {
|
||||
now := time.Date(2026, 8, 15, 10, 0, 0, 0, time.UTC)
|
||||
stmts := partitionStatementsRange(now, now.AddDate(0, 2, 0))
|
||||
if len(stmts) != 6 {
|
||||
t.Fatalf("partitionStatements len = %d, want 6", len(stmts))
|
||||
}
|
||||
want := []string{
|
||||
"CREATE TABLE IF NOT EXISTS of_node_access_logs_202608 PARTITION OF of_node_access_logs FOR VALUES FROM ('2026-08-01') TO ('2026-09-01')",
|
||||
"CREATE TABLE IF NOT EXISTS w_user_access_logs_202608 PARTITION OF w_user_access_logs FOR VALUES FROM ('2026-08-01') TO ('2026-09-01')",
|
||||
"CREATE TABLE IF NOT EXISTS of_node_access_logs_202609 PARTITION OF of_node_access_logs FOR VALUES FROM ('2026-09-01') TO ('2026-10-01')",
|
||||
"CREATE TABLE IF NOT EXISTS w_user_access_logs_202609 PARTITION OF w_user_access_logs FOR VALUES FROM ('2026-09-01') TO ('2026-10-01')",
|
||||
"CREATE TABLE IF NOT EXISTS of_node_access_logs_202610 PARTITION OF of_node_access_logs FOR VALUES FROM ('2026-10-01') TO ('2026-11-01')",
|
||||
"CREATE TABLE IF NOT EXISTS w_user_access_logs_202610 PARTITION OF w_user_access_logs FOR VALUES FROM ('2026-10-01') TO ('2026-11-01')",
|
||||
}
|
||||
for i, w := range want {
|
||||
if stmts[i] != w {
|
||||
t.Fatalf("stmt[%d] = %q, want %q", i, stmts[i], w)
|
||||
}
|
||||
}
|
||||
|
||||
// 跨年：2026-11 → 202611, 202612, 202701。
|
||||
nov := time.Date(2026, 11, 1, 0, 0, 0, 0, time.UTC)
|
||||
suffixes := []string{"202611", "202612", "202701"}
|
||||
for _, stmt := range partitionStatementsRange(nov, nov.AddDate(0, 2, 0)) {
|
||||
if !hasAnySuffix(stmt, suffixes) {
|
||||
t.Fatalf("statement lacks expected month suffix: %s", stmt)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func hasAnySuffix(stmt string, suffixes []string) bool {
|
||||
for _, table := range []string{"of_node_access_logs", "w_user_access_logs"} {
|
||||
for _, suf := range suffixes {
|
||||
if strings.Contains(stmt, table+"_"+suf) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,731 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"math"
|
||||
"time"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
|
||||
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"
|
||||
analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
|
||||
)
|
||||
|
||||
// clickhouseLogStore 实现 AccessLogStore / ObservabilityStore / StatusStore，
|
||||
// 逐方法委托 analyticsrepo（CH 原生 batch 写入，零性能损耗）。
|
||||
// UserAccessLogStore 由 clickhouseUserAccessLogStore 实现（List/Count 方法名已被
|
||||
// AccessLogStore 占用，Go 不允许同名不同签名方法）。
|
||||
type clickhouseLogStore struct {
|
||||
// skipFreeze 为 true 时跳过迁移冻结检查（仅迁移目标 store 使用）。
|
||||
skipFreeze bool
|
||||
}
|
||||
|
||||
func newClickHouseStore() *clickhouseLogStore { return &clickhouseLogStore{} }
|
||||
|
||||
// 编译期断言。
|
||||
var (
|
||||
_ AccessLogStore = (*clickhouseLogStore)(nil)
|
||||
_ ObservabilityStore = (*clickhouseLogStore)(nil)
|
||||
_ StatusStore = (*clickhouseLogStore)(nil)
|
||||
_ UserAccessLogStore = (*clickhouseUserAccessLogStore)(nil)
|
||||
)
|
||||
|
||||
func chConnErr() error {
|
||||
if !db.ChConnReady() {
|
||||
return errors.New("clickhouse connection is not initialized")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ensureWritable 迁移冻结期拒绝写入。
|
||||
func (s *clickhouseLogStore) ensureWritable(ctx context.Context) error {
|
||||
if !s.skipFreeze && Migrating(ctx) {
|
||||
return ErrMigrating
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ---- AccessLogStore ----
|
||||
|
||||
// InsertBatch 节点访问日志写入入口：冻结检查后经 hook 入队（异步），不直接落库。
|
||||
func (s *clickhouseLogStore) InsertBatch(ctx context.Context, records []*model.OpenFlareAccessLog) error {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
rows := make([]analyticsmodel.NodeAccessLog, 0, len(records))
|
||||
for _, r := range records {
|
||||
if r == nil {
|
||||
continue
|
||||
}
|
||||
rows = append(rows, toAnalyticsNodeAccessLog(r))
|
||||
}
|
||||
if h := currentAccessLogHooks().QueueNodeAccessLogs; h != nil {
|
||||
h(rows)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// BatchInsertNodeAccessLogs 是 batchwriter flush 目标：CH 原生批量写入。
|
||||
func (s *clickhouseLogStore) BatchInsertNodeAccessLogs(ctx context.Context, rows []analyticsmodel.NodeAccessLog) error {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
return analyticsrepo.BatchInsertNodeAccessLogs(ctx, rows)
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) List(ctx context.Context, query model.OpenFlareAccessLogQuery) ([]*model.OpenFlareAccessLog, error) {
|
||||
rows, err := analyticsrepo.ListNodeAccessLogs(ctx, toNodeAccessLogFilter(query))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return fromAnalyticsNodeAccessLogs(rows), nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) Count(ctx context.Context, query model.OpenFlareAccessLogQuery) (int64, int64, int64, error) {
|
||||
return analyticsrepo.CountNodeAccessLogs(ctx, toNodeAccessLogFilter(query))
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) RegionCounts(ctx context.Context, nodeID string, since time.Time, limit int) ([]*model.OpenFlareAccessLogRegionCount, error) {
|
||||
rows, err := analyticsrepo.RegionCountsNodeAccessLogs(ctx, nodeID, since, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]*model.OpenFlareAccessLogRegionCount, len(rows))
|
||||
for i, r := range rows {
|
||||
out[i] = &model.OpenFlareAccessLogRegionCount{Region: r.Region, Count: r.Count}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) BucketAggregates(ctx context.Context, query model.OpenFlareAccessLogQuery, bucketSeconds int64) ([]analyticsmodel.NodeAccessLogBucketAggregate, error) {
|
||||
return analyticsrepo.BucketAggregatesNodeAccessLogs(ctx, toNodeAccessLogFilter(query), bucketSeconds)
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) CountBuckets(ctx context.Context, query model.OpenFlareAccessLogQuery, bucketSeconds int64) (int64, error) {
|
||||
return analyticsrepo.CountBucketAggregatesNodeAccessLogs(ctx, toNodeAccessLogFilter(query), bucketSeconds)
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) BucketDimensions(ctx context.Context, query model.OpenFlareAccessLogQuery, column string, bucketSeconds int64) ([]analyticsmodel.NodeAccessLogBucketDimension, error) {
|
||||
return analyticsrepo.BucketDimensionsNodeAccessLogs(ctx, toNodeAccessLogFilter(query), column, bucketSeconds)
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) IPAggregates(ctx context.Context, query model.OpenFlareAccessLogQuery, exactRemoteAddr bool) ([]analyticsmodel.NodeAccessLogIPAggregate, error) {
|
||||
return analyticsrepo.IPAggregatesNodeAccessLogs(ctx, toNodeAccessLogFilter(query), exactRemoteAddr)
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) IPSummaries(ctx context.Context, query model.OpenFlareAccessLogQuery, recentSince time.Time) ([]analyticsmodel.NodeAccessLogIPSummary, error) {
|
||||
return analyticsrepo.IPSummariesNodeAccessLogs(ctx, toNodeAccessLogFilter(query), recentSince)
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) CountIPSummaries(ctx context.Context, query model.OpenFlareAccessLogQuery) (int64, error) {
|
||||
return analyticsrepo.CountIPSummaryNodeAccessLogs(ctx, toNodeAccessLogFilter(query))
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) WAFIPAggregates(ctx context.Context, query model.OpenFlareAccessLogQuery) ([]analyticsmodel.NodeAccessLogWAFIPAggregate, error) {
|
||||
return analyticsrepo.IPAggregatesForWAFNodeAccessLogs(ctx, toNodeAccessLogFilter(query))
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) IPTrend(ctx context.Context, query model.OpenFlareAccessLogQuery, bucketSeconds int64) ([]analyticsmodel.NodeAccessLogIPTrend, error) {
|
||||
return analyticsrepo.IPTrendNodeAccessLogs(ctx, toNodeAccessLogFilter(query), bucketSeconds)
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) TrafficSummary(ctx context.Context, query model.OpenFlareAccessLogQuery) (model.OpenFlareAccessLogTrafficSummary, error) {
|
||||
row, err := analyticsrepo.TrafficSummaryNodeAccessLogs(ctx, toNodeAccessLogFilter(query))
|
||||
if err != nil {
|
||||
return model.OpenFlareAccessLogTrafficSummary{}, err
|
||||
}
|
||||
return model.OpenFlareAccessLogTrafficSummary{
|
||||
RequestCount: row.RequestCount,
|
||||
ErrorCount: row.ErrorCount,
|
||||
UniqueIPCount: row.UniqueIPCount,
|
||||
BytesSent: row.BytesSent,
|
||||
RequestLength: row.RequestLength,
|
||||
NodeCount: row.NodeCount,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) ValueCounts(ctx context.Context, query model.OpenFlareAccessLogQuery, column string, limit int) ([]model.OpenFlareAccessLogValueCount, error) {
|
||||
rows, err := analyticsrepo.ValueCountsNodeAccessLogs(ctx, toNodeAccessLogFilter(query), column, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]model.OpenFlareAccessLogValueCount, len(rows))
|
||||
for i, r := range rows {
|
||||
out[i] = model.OpenFlareAccessLogValueCount{Value: r.Value, Count: r.Count}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) NodeAggregates(ctx context.Context, query model.OpenFlareAccessLogQuery) ([]model.OpenFlareAccessLogNodeAggregate, error) {
|
||||
rows, err := analyticsrepo.NodeAggregatesNodeAccessLogs(ctx, toNodeAccessLogFilter(query))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]model.OpenFlareAccessLogNodeAggregate, len(rows))
|
||||
for i, r := range rows {
|
||||
out[i] = model.OpenFlareAccessLogNodeAggregate{NodeID: r.NodeID, RequestCount: r.RequestCount, ErrorCount: r.ErrorCount, UniqueIPCount: r.UniqueIPCount}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) DeleteAll(ctx context.Context) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return analyticsrepo.DeleteAllNodeAccessLogs(ctx)
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return analyticsrepo.DeleteNodeAccessLogsBefore(ctx, cutoff)
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) DeleteByNodeBefore(ctx context.Context, nodeID string, before time.Time) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return analyticsrepo.DeleteNodeAccessLogsByNodeBefore(ctx, nodeID, before)
|
||||
}
|
||||
|
||||
// ListForMigration 按 id 升序分页读取（迁移复制用）：直接查询 CH 原生表。
|
||||
func (s *clickhouseLogStore) ListForMigration(ctx context.Context, afterID uint64, limit int) ([]analyticsmodel.NodeAccessLog, error) {
|
||||
if err := chConnErr(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rows, err := db.ChConn.Query(ctx, `
|
||||
SELECT `+analyticsmodel.NodeAccessLog{}.InsertColumns()+`
|
||||
FROM `+analyticsmodel.NodeAccessLog{}.TableName()+`
|
||||
WHERE id > ?
|
||||
ORDER BY id ASC
|
||||
LIMIT ?`, afterID, limitOr(limit, migrationPageSize))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list node access logs for migration: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
var result []analyticsmodel.NodeAccessLog
|
||||
for rows.Next() {
|
||||
var item analyticsmodel.NodeAccessLog
|
||||
if err := rows.Scan(
|
||||
&item.ID,
|
||||
&item.NodeID,
|
||||
&item.LoggedAt,
|
||||
&item.RemoteAddr,
|
||||
&item.Region,
|
||||
&item.Host,
|
||||
&item.Path,
|
||||
&item.UserAgent,
|
||||
&item.CacheStatus,
|
||||
&item.StatusCode,
|
||||
&item.BytesSent,
|
||||
&item.RequestLength,
|
||||
&item.RequestTimeMs,
|
||||
&item.CreatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("scan node access log row: %w", err)
|
||||
}
|
||||
item.LoggedAt = item.LoggedAt.UTC()
|
||||
item.CreatedAt = item.CreatedAt.UTC()
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// ---- ObservabilityStore ----
|
||||
|
||||
// InsertMetricSnapshot 写入入口：冻结检查 + 经 hook 入队（异步），不直接落库。
|
||||
func (s *clickhouseLogStore) InsertMetricSnapshot(ctx context.Context, record *model.OpenFlareMetricSnapshot) error {
|
||||
if record == nil {
|
||||
return nil
|
||||
}
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if h := currentObservabilityHooks().QueueMetricSnapshot; h != nil {
|
||||
h(toAnalyticsNodeMetricSnapshot(record))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) ListMetricSnapshots(ctx context.Context, nodeID string, since time.Time, limit int) ([]*model.OpenFlareMetricSnapshot, error) {
|
||||
rows, err := analyticsrepo.ListNodeMetricSnapshots(ctx, toNodeObservabilityFilter(nodeID, since, limit))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return fromAnalyticsNodeMetricSnapshots(rows), nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) DeleteAllMetricSnapshots(ctx context.Context) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return analyticsrepo.DeleteAllNodeMetricSnapshots(ctx)
|
||||
}
|
||||
|
||||
// ListTrafficHourly 委托 analyticsrepo 读 of_access_log_hourly rollup（M5 口径，UV 恒 0）。
|
||||
func (s *clickhouseLogStore) ListTrafficHourly(ctx context.Context, nodeID string, since time.Time) ([]analyticsmodel.NodeTrafficHourly, error) {
|
||||
return analyticsrepo.ListNodeTrafficHourly(ctx, toNodeObservabilitySince(nodeID, since))
|
||||
}
|
||||
|
||||
// ListAccessLogHourly 委托 analyticsrepo 读 of_access_log_hourly rollup。
|
||||
func (s *clickhouseLogStore) ListAccessLogHourly(ctx context.Context, nodeID string, since time.Time) ([]analyticsmodel.AccessLogHourly, error) {
|
||||
return analyticsrepo.ListAccessLogHourly(ctx, toNodeObservabilitySince(nodeID, since))
|
||||
}
|
||||
|
||||
// ListMetricHourly 委托 analyticsrepo ListNodeMetricHourly：rollup 覆盖窗口时读
|
||||
// of_node_metric_capacity_hourly，否则按 mergeNodeMetricHourlyPreferRollup 合并 raw 兜底。
|
||||
func (s *clickhouseLogStore) ListMetricHourly(ctx context.Context, nodeID string, since time.Time) ([]analyticsmodel.NodeMetricHourly, error) {
|
||||
return analyticsrepo.ListNodeMetricHourly(ctx, toNodeObservabilitySince(nodeID, since))
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) DeleteMetricSnapshotsBefore(ctx context.Context, cutoff time.Time) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return analyticsrepo.DeleteNodeMetricSnapshotsBefore(ctx, cutoff)
|
||||
}
|
||||
|
||||
// BatchInsertNodeMetricSnapshots 是 batchwriter flush 目标：CH 原生批量写入。
|
||||
func (s *clickhouseLogStore) BatchInsertNodeMetricSnapshots(ctx context.Context, rows []analyticsmodel.NodeMetricSnapshot) error {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
return analyticsrepo.BatchInsertNodeMetricSnapshots(ctx, rows)
|
||||
}
|
||||
|
||||
// InsertEdgeHealth 写入入口：冻结检查 + 经 hook 入队（异步），不直接落库。
|
||||
func (s *clickhouseLogStore) InsertEdgeHealth(ctx context.Context, record *model.OpenFlareEdgeHealth) error {
|
||||
if record == nil {
|
||||
return nil
|
||||
}
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if h := currentObservabilityHooks().QueueEdgeHealth; h != nil {
|
||||
h(toAnalyticsNodeEdgeHealth(record))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) ListEdgeHealth(ctx context.Context, nodeID string, since time.Time, limit int) ([]*model.OpenFlareEdgeHealth, error) {
|
||||
rows, err := analyticsrepo.ListNodeEdgeHealth(ctx, toNodeObservabilityFilter(nodeID, since, limit))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return fromAnalyticsNodeEdgeHealths(rows), nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) DeleteAllEdgeHealth(ctx context.Context) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return analyticsrepo.DeleteAllNodeEdgeHealth(ctx)
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) DeleteEdgeHealthBefore(ctx context.Context, cutoff time.Time) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return analyticsrepo.DeleteNodeEdgeHealthBefore(ctx, cutoff)
|
||||
}
|
||||
|
||||
// BatchInsertNodeEdgeHealth 是 batchwriter flush 目标：CH 原生批量写入。
|
||||
func (s *clickhouseLogStore) BatchInsertNodeEdgeHealth(ctx context.Context, rows []analyticsmodel.NodeEdgeHealth) error {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
return analyticsrepo.BatchInsertNodeEdgeHealth(ctx, rows)
|
||||
}
|
||||
|
||||
// InsertNodeObservationFrps 写入入口：冻结检查 + 经 hook 入队（异步），不直接落库。
|
||||
func (s *clickhouseLogStore) InsertNodeObservationFrps(ctx context.Context, record *model.OpenFlareNodeObservationFrps) error {
|
||||
if record == nil {
|
||||
return nil
|
||||
}
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if h := currentObservabilityHooks().QueueNodeObsFrps; h != nil {
|
||||
h(toAnalyticsNodeObsFrps(record))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) ListNodeObservationFrps(ctx context.Context, nodeID string, since time.Time, limit int) ([]*model.OpenFlareNodeObservationFrps, error) {
|
||||
rows, err := analyticsrepo.ListNodeObsFrps(ctx, toNodeObservabilityFilter(nodeID, since, limit))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return fromAnalyticsNodeObsFrps(rows), nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) DeleteAllNodeObservationFrps(ctx context.Context) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return analyticsrepo.DeleteAllNodeObsFrps(ctx)
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) DeleteNodeObservationFrpsBefore(ctx context.Context, cutoff time.Time) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return analyticsrepo.DeleteNodeObsFrpsBefore(ctx, cutoff)
|
||||
}
|
||||
|
||||
// BatchInsertNodeObsFrps 是 batchwriter flush 目标：CH 原生批量写入。
|
||||
func (s *clickhouseLogStore) BatchInsertNodeObsFrps(ctx context.Context, rows []analyticsmodel.NodeObsFrps) error {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
return analyticsrepo.BatchInsertNodeObsFrps(ctx, rows)
|
||||
}
|
||||
|
||||
// InsertNodeObservationFrpc 写入入口：冻结检查 + 经 hook 入队（异步），不直接落库。
|
||||
func (s *clickhouseLogStore) InsertNodeObservationFrpc(ctx context.Context, record *model.OpenFlareNodeObservationFrpc) error {
|
||||
if record == nil {
|
||||
return nil
|
||||
}
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if h := currentObservabilityHooks().QueueNodeObsFrpc; h != nil {
|
||||
h(toAnalyticsNodeObsFrpc(record))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) ListNodeObservationFrpc(ctx context.Context, nodeID string, since time.Time, limit int) ([]*model.OpenFlareNodeObservationFrpc, error) {
|
||||
rows, err := analyticsrepo.ListNodeObsFrpc(ctx, toNodeObservabilityFilter(nodeID, since, limit))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return fromAnalyticsNodeObsFrpc(rows), nil
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) DeleteAllNodeObservationFrpc(ctx context.Context) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return analyticsrepo.DeleteAllNodeObsFrpc(ctx)
|
||||
}
|
||||
|
||||
func (s *clickhouseLogStore) DeleteNodeObservationFrpcBefore(ctx context.Context, cutoff time.Time) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return analyticsrepo.DeleteNodeObsFrpcBefore(ctx, cutoff)
|
||||
}
|
||||
|
||||
// MigrationRange 返回 of_node_access_logs.logged_at 的最小/最大值（空表返回零值）。
|
||||
func (s *clickhouseLogStore) MigrationRange(ctx context.Context) (time.Time, time.Time, error) {
|
||||
return chMigrationRange(ctx, analyticsmodel.NodeAccessLog{}.TableName(), "logged_at")
|
||||
}
|
||||
|
||||
// EnsurePartitions 是 CH 分支 no-op（CH 无 PG 式分区）。
|
||||
func (s *clickhouseLogStore) EnsurePartitions(_ context.Context, _, _ time.Time) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// chMigrationRange 查询 CH 表时间列 MIN/MAX；空表（NULL）返回零值。
|
||||
func chMigrationRange(ctx context.Context, table, column string) (time.Time, time.Time, error) {
|
||||
if err := chConnErr(); err != nil {
|
||||
return time.Time{}, time.Time{}, err
|
||||
}
|
||||
var minTime, maxTime *time.Time
|
||||
if err := db.ChConn.QueryRow(ctx,
|
||||
"SELECT min("+column+"), max("+column+") FROM "+table,
|
||||
).Scan(&minTime, &maxTime); err != nil {
|
||||
return time.Time{}, time.Time{}, fmt.Errorf("query migration range %s: %w", table, err)
|
||||
}
|
||||
if minTime == nil || maxTime == nil {
|
||||
return time.Time{}, time.Time{}, nil
|
||||
}
|
||||
return minTime.UTC(), maxTime.UTC(), nil
|
||||
}
|
||||
|
||||
// BatchInsertNodeObsFrpc 是 batchwriter flush 目标：CH 原生批量写入。
|
||||
func (s *clickhouseLogStore) BatchInsertNodeObsFrpc(ctx context.Context, rows []analyticsmodel.NodeObsFrpc) error {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
return analyticsrepo.BatchInsertNodeObsFrpc(ctx, rows)
|
||||
}
|
||||
|
||||
// ListMetricSnapshotsForMigration 按 id 升序分页读取（迁移复制用）。
|
||||
func (s *clickhouseLogStore) ListMetricSnapshotsForMigration(ctx context.Context, afterID uint64, limit int) ([]analyticsmodel.NodeMetricSnapshot, error) {
|
||||
return chListForMigration(ctx, afterID, limit,
|
||||
analyticsmodel.NodeMetricSnapshot{}.TableName(),
|
||||
analyticsmodel.NodeMetricSnapshot{}.InsertColumns(),
|
||||
func(rows driver.Rows) ([]analyticsmodel.NodeMetricSnapshot, error) {
|
||||
var result []analyticsmodel.NodeMetricSnapshot
|
||||
for rows.Next() {
|
||||
var item analyticsmodel.NodeMetricSnapshot
|
||||
if err := rows.Scan(
|
||||
&item.ID,
|
||||
&item.NodeID,
|
||||
&item.CapturedAt,
|
||||
&item.CPUUsagePercent,
|
||||
&item.MemoryUsedBytes,
|
||||
&item.MemoryTotalBytes,
|
||||
&item.StorageUsedBytes,
|
||||
&item.StorageTotalBytes,
|
||||
&item.DiskReadBytes,
|
||||
&item.DiskWriteBytes,
|
||||
&item.NetworkRxBytes,
|
||||
&item.NetworkTxBytes,
|
||||
&item.CreatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("scan node metric snapshot row: %w", err)
|
||||
}
|
||||
item.CapturedAt = item.CapturedAt.UTC()
|
||||
item.CreatedAt = item.CreatedAt.UTC()
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
}
|
||||
|
||||
// chObsRow 迁移读取共用的双字段观测行（字符串状态 + 数值计数）：
|
||||
// edge_health（status/connections）与 obs_frpc（tunnel_status/connected_relays_count）同形状。
|
||||
type chObsRow struct {
|
||||
ID uint64
|
||||
NodeID string
|
||||
CapturedAt time.Time
|
||||
Status string
|
||||
Count int64
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
// countToInt32 将观测计数转为 int32（防御溢出；观测计数远小于 int32 上限）。
|
||||
func countToInt32(v int64) int32 {
|
||||
if v > math.MaxInt32 {
|
||||
return math.MaxInt32
|
||||
}
|
||||
if v < math.MinInt32 {
|
||||
return math.MinInt32
|
||||
}
|
||||
return int32(v)
|
||||
}
|
||||
|
||||
// scanChObsRow 扫描 chObsRow（含 UTC 归一化）。
|
||||
func scanChObsRow(rows driver.Rows) ([]chObsRow, error) {
|
||||
var result []chObsRow
|
||||
for rows.Next() {
|
||||
var item chObsRow
|
||||
if err := rows.Scan(
|
||||
&item.ID,
|
||||
&item.NodeID,
|
||||
&item.CapturedAt,
|
||||
&item.Status,
|
||||
&item.Count,
|
||||
&item.CreatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("scan observation row: %w", err)
|
||||
}
|
||||
item.CapturedAt = item.CapturedAt.UTC()
|
||||
item.CreatedAt = item.CreatedAt.UTC()
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// ListEdgeHealthForMigration 按 id 升序分页读取（迁移复制用）。
|
||||
func (s *clickhouseLogStore) ListEdgeHealthForMigration(ctx context.Context, afterID uint64, limit int) ([]analyticsmodel.NodeEdgeHealth, error) {
|
||||
rows, err := chListForMigration(ctx, afterID, limit,
|
||||
analyticsmodel.NodeEdgeHealth{}.TableName(),
|
||||
analyticsmodel.NodeEdgeHealth{}.InsertColumns(),
|
||||
scanChObsRow)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]analyticsmodel.NodeEdgeHealth, len(rows))
|
||||
for i, r := range rows {
|
||||
out[i] = analyticsmodel.NodeEdgeHealth{ID: r.ID, NodeID: r.NodeID, CapturedAt: r.CapturedAt, Status: r.Status, Connections: r.Count, CreatedAt: r.CreatedAt}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ListNodeObsFrpsForMigration 按 id 升序分页读取（迁移复制用）。
|
||||
func (s *clickhouseLogStore) ListNodeObsFrpsForMigration(ctx context.Context, afterID uint64, limit int) ([]analyticsmodel.NodeObsFrps, error) {
|
||||
return chListForMigration(ctx, afterID, limit,
|
||||
analyticsmodel.NodeObsFrps{}.TableName(),
|
||||
analyticsmodel.NodeObsFrps{}.InsertColumns(),
|
||||
func(rows driver.Rows) ([]analyticsmodel.NodeObsFrps, error) {
|
||||
var result []analyticsmodel.NodeObsFrps
|
||||
for rows.Next() {
|
||||
var item analyticsmodel.NodeObsFrps
|
||||
if err := rows.Scan(
|
||||
&item.ID,
|
||||
&item.NodeID,
|
||||
&item.CapturedAt,
|
||||
&item.FrpsConnections,
|
||||
&item.FrpsProxyCount,
|
||||
&item.FrpsClientCount,
|
||||
&item.FrpsProxies,
|
||||
&item.CreatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("scan node frps observation row: %w", err)
|
||||
}
|
||||
item.CapturedAt = item.CapturedAt.UTC()
|
||||
item.CreatedAt = item.CreatedAt.UTC()
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
}
|
||||
|
||||
// ListNodeObsFrpcForMigration 按 id 升序分页读取（迁移复制用）。
|
||||
func (s *clickhouseLogStore) ListNodeObsFrpcForMigration(ctx context.Context, afterID uint64, limit int) ([]analyticsmodel.NodeObsFrpc, error) {
|
||||
rows, err := chListForMigration(ctx, afterID, limit,
|
||||
analyticsmodel.NodeObsFrpc{}.TableName(),
|
||||
analyticsmodel.NodeObsFrpc{}.InsertColumns(),
|
||||
scanChObsRow)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]analyticsmodel.NodeObsFrpc, len(rows))
|
||||
for i, r := range rows {
|
||||
out[i] = analyticsmodel.NodeObsFrpc{ID: r.ID, NodeID: r.NodeID, CapturedAt: r.CapturedAt, TunnelStatus: r.Status, ConnectedRelaysCount: countToInt32(r.Count), CreatedAt: r.CreatedAt}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// chListForMigration 执行按 id 升序分页的 CH 原生表查询，并交给 scanner 扫描。
|
||||
func chListForMigration[T any](ctx context.Context, afterID uint64, limit int, table, columns string, scanner func(driver.Rows) ([]T, error)) ([]T, error) {
|
||||
if err := chConnErr(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rows, err := db.ChConn.Query(ctx, `
|
||||
SELECT `+columns+`
|
||||
FROM `+table+`
|
||||
WHERE id > ?
|
||||
ORDER BY id ASC
|
||||
LIMIT ?`, afterID, limitOr(limit, migrationPageSize))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list %s for migration: %w", table, err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
return scanner(rows)
|
||||
}
|
||||
|
||||
// ---- StatusStore ----
|
||||
|
||||
// ActiveDatabase 返回当前日志主库名（CH 分支固定 clickhouse）。
|
||||
func (s *clickhouseLogStore) ActiveDatabase(_ context.Context) (string, error) {
|
||||
return dbNameClickHouse, nil
|
||||
}
|
||||
|
||||
// ClickHouseOperationalStats 委托 analyticsrepo 汇总 CH 运行状态。
|
||||
func (s *clickhouseLogStore) ClickHouseOperationalStats(ctx context.Context) (*analyticsmodel.ClickHouseOperationalStats, error) {
|
||||
return analyticsrepo.GetClickHouseOperationalStats(ctx)
|
||||
}
|
||||
|
||||
// ---- UserAccessLogStore ----
|
||||
|
||||
// clickhouseUserAccessLogStore 实现 UserAccessLogStore。clickhouseLogStore 已占用
|
||||
// List/Count 方法名（AccessLogStore 接口），Go 不允许同名不同签名方法，故用户访问日志
|
||||
// 用独立类型嵌入同一 clickhouseLogStore（与 userAccessLogGormStore 同构），复用 ensureWritable。
|
||||
type clickhouseUserAccessLogStore struct {
|
||||
*clickhouseLogStore
|
||||
}
|
||||
|
||||
func newClickHouseUserAccessLogStore() *clickhouseUserAccessLogStore {
|
||||
return &clickhouseUserAccessLogStore{clickhouseLogStore: newClickHouseStore()}
|
||||
}
|
||||
|
||||
// BatchInsert 是 batchwriter flush 目标：CH 原生批量写入；冻结期拒绝写入，空批次直接返回。
|
||||
func (s *clickhouseUserAccessLogStore) BatchInsert(ctx context.Context, logs []analyticsmodel.UserAccessLog) error {
|
||||
if len(logs) == 0 {
|
||||
return nil
|
||||
}
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
return analyticsrepo.BatchInsert(ctx, logs)
|
||||
}
|
||||
|
||||
// DeleteAll 清空全部用户访问日志（TRUNCATE 语义，迁移「覆盖目标库已有日志」幂等前提用）。
|
||||
func (s *clickhouseUserAccessLogStore) DeleteAll(ctx context.Context) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return analyticsrepo.DeleteAllUserAccessLogs(ctx)
|
||||
}
|
||||
|
||||
// ListForMigration 按 id 升序分页读取（迁移复制用）。
|
||||
func (s *clickhouseUserAccessLogStore) ListForMigration(ctx context.Context, afterID uint64, limit int) ([]analyticsmodel.UserAccessLog, error) {
|
||||
return chListForMigration(ctx, afterID, limit,
|
||||
analyticsmodel.UserAccessLog{}.TableName(),
|
||||
analyticsmodel.UserAccessLog{}.InsertColumns(),
|
||||
func(rows driver.Rows) ([]analyticsmodel.UserAccessLog, error) {
|
||||
var result []analyticsmodel.UserAccessLog
|
||||
for rows.Next() {
|
||||
var item analyticsmodel.UserAccessLog
|
||||
if err := rows.Scan(
|
||||
&item.ID,
|
||||
&item.UserID,
|
||||
&item.Path,
|
||||
&item.Method,
|
||||
&item.IP,
|
||||
&item.UserAgent,
|
||||
&item.Headers,
|
||||
&item.Status,
|
||||
&item.Latency,
|
||||
&item.CreatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("scan user access log row: %w", err)
|
||||
}
|
||||
item.CreatedAt = item.CreatedAt.UTC()
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
}
|
||||
|
||||
// MigrationRange 返回 w_user_access_logs.created_at 的最小/最大值（空表返回零值）。
|
||||
func (s *clickhouseUserAccessLogStore) MigrationRange(ctx context.Context) (time.Time, time.Time, error) {
|
||||
return chMigrationRange(ctx, analyticsmodel.UserAccessLog{}.TableName(), "created_at")
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) Count(ctx context.Context, filter analyticsmodel.AccessLogFilter) (uint64, error) {
|
||||
return analyticsrepo.CountAccessLogs(ctx, filter)
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) List(ctx context.Context, filter analyticsmodel.AccessLogFilter, page, pageSize int) ([]analyticsmodel.UserAccessLog, uint64, error) {
|
||||
return analyticsrepo.ListAccessLogs(ctx, filter, page, pageSize)
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) GetDailyTrend(ctx context.Context, days int) ([]analyticsmodel.DailyTrend, error) {
|
||||
return analyticsrepo.GetDailyTrend(ctx, days)
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) GetBrowserDistribution(ctx context.Context, startTime time.Time) ([]analyticsmodel.BrowserShare, error) {
|
||||
return analyticsrepo.GetBrowserDistribution(ctx, startTime)
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) GetTopActiveUsers(ctx context.Context, startTime time.Time, limit int) ([]analyticsmodel.TopUser, error) {
|
||||
return analyticsrepo.GetTopActiveUsers(ctx, startTime, limit)
|
||||
}
|
||||
|
||||
// toNodeObservabilityFilter 构造 CH 可观测查询过滤器（limit<=0 表示不限制）。
|
||||
func toNodeObservabilityFilter(nodeID string, since time.Time, limit int) analyticsmodel.NodeObservabilityFilter {
|
||||
return analyticsmodel.NodeObservabilityFilter{
|
||||
NodeID: nodeID,
|
||||
Since: since,
|
||||
Limit: limit,
|
||||
}
|
||||
}
|
||||
|
||||
// toNodeObservabilitySince 构造不带 limit 的可观测查询过滤器
|
||||
// （小时级聚合读无需分页，避免传无意义的 0）。
|
||||
func toNodeObservabilitySince(nodeID string, since time.Time) analyticsmodel.NodeObservabilityFilter {
|
||||
return analyticsmodel.NodeObservabilityFilter{NodeID: nodeID, Since: since}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
|
||||
)
|
||||
|
||||
// TestClickHouseHourlyDelegationRegression 验证 CH 后端小时级聚合读委托 analyticsrepo：
|
||||
// 未初始化 CH 连接时返回 analyticsrepo 的 "clickhouse connection is not initialized" 错误
|
||||
// （而非未实现/panic），证明 3 个方法都路由到 CH 原生查询。
|
||||
func TestClickHouseHourlyDelegationRegression(t *testing.T) {
|
||||
if db.ChConnReady() {
|
||||
t.Skip("clickhouse connection initialized; skipping delegation regression")
|
||||
}
|
||||
s := newClickHouseStore()
|
||||
ctx := context.Background()
|
||||
now := time.Now()
|
||||
check := func(name string, err error) {
|
||||
t.Helper()
|
||||
if err == nil {
|
||||
t.Fatalf("%s: want clickhouse-not-initialized error, got nil", name)
|
||||
}
|
||||
if !strings.Contains(err.Error(), "clickhouse connection is not initialized") {
|
||||
t.Fatalf("%s: unexpected error %v", name, err)
|
||||
}
|
||||
}
|
||||
_, err := s.ListTrafficHourly(ctx, "n1", now)
|
||||
check("ListTrafficHourly", err)
|
||||
_, err = s.ListAccessLogHourly(ctx, "n1", now)
|
||||
check("ListAccessLogHourly", err)
|
||||
_, err = s.ListMetricHourly(ctx, "n1", now)
|
||||
check("ListMetricHourly", err)
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"strconv"
|
||||
)
|
||||
|
||||
// timeBucketSQLPostgres 返回 PG 时间分桶表达式（epoch 秒 -> 分桶起点，int64）。
|
||||
func timeBucketSQLPostgres(column string, bucketSeconds int64) string {
|
||||
return "(floor(extract(epoch from " + column + ")/" + strconv.FormatInt(bucketSeconds, 10) + ")*" + strconv.FormatInt(bucketSeconds, 10) + ")::bigint"
|
||||
}
|
||||
|
||||
// dailyTrendDateSQLPostgres 返回 PG 按日聚合的日期表达式。
|
||||
func dailyTrendDateSQLPostgres() string {
|
||||
return "to_char(created_at, 'YYYY-MM-DD')"
|
||||
}
|
||||
|
||||
// epochSQLPostgres 返回 PG epoch 秒表达式（int64）。
|
||||
func epochSQLPostgres(column string) string {
|
||||
return "extract(epoch from " + column + ")::bigint"
|
||||
}
|
||||
|
||||
// textCastSQLPostgres 返回 PG 数值列转文本表达式。
|
||||
func textCastSQLPostgres(column string) string {
|
||||
return column + "::text"
|
||||
}
|
||||
|
||||
// distinctNonEmptyCountSQLPostgres 返回 PG 排除空串的 distinct 计数表达式。
|
||||
func distinctNonEmptyCountSQLPostgres(column string) string {
|
||||
return "COUNT(DISTINCT " + column + ") FILTER (WHERE " + column + " <> '')"
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"strconv"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// isPostgresDialect 判断 gorm 句柄是否为 PostgreSQL 方言（否则按 SQLite 处理）。
|
||||
// Dialector 经 gorm.Config 内嵌提升，Name() 可直接在 DB 上调用。
|
||||
func isPostgresDialect(db *gorm.DB) bool {
|
||||
return db != nil && db.Dialector != nil && db.Name() == "postgres"
|
||||
}
|
||||
|
||||
// timeBucketSQLSQLite 返回 SQLite 时间分桶表达式（epoch 秒 -> 分桶起点）。
|
||||
func timeBucketSQLSQLite(column string, bucketSeconds int64) string {
|
||||
return "(floor(unixepoch(" + column + ")/" + strconv.FormatInt(bucketSeconds, 10) + ")*" + strconv.FormatInt(bucketSeconds, 10) + ")"
|
||||
}
|
||||
|
||||
// dailyTrendDateSQLSQLite 返回 SQLite 按日聚合的日期表达式。
|
||||
func dailyTrendDateSQLSQLite() string {
|
||||
return "strftime('%Y-%m-%d', created_at)"
|
||||
}
|
||||
|
||||
// epochSQLSQLite 返回 SQLite epoch 秒表达式（unixepoch 整数秒）。
|
||||
func epochSQLSQLite(column string) string {
|
||||
return "unixepoch(" + column + ")"
|
||||
}
|
||||
|
||||
// textCastSQLSQLite 返回 SQLite 数值列转文本表达式。
|
||||
func textCastSQLSQLite(column string) string {
|
||||
return "CAST(" + column + " AS TEXT)"
|
||||
}
|
||||
|
||||
// distinctNonEmptyCountSQLSQLite 返回 SQLite 排除空串的 distinct 计数表达式
|
||||
// （SQLite 无 FILTER 语法，用 CASE 等价实现）。
|
||||
func distinctNonEmptyCountSQLSQLite(column string) string {
|
||||
return "COUNT(DISTINCT CASE WHEN " + column + " <> '' THEN " + column + " END)"
|
||||
}
|
||||
|
||||
// distinctNonEmptyCountSQL 按当前方言返回排除空串的 distinct 计数表达式
|
||||
// （运行时按 Dialector 分发，默认 SQLite）。
|
||||
func distinctNonEmptyCountSQL(db *gorm.DB, column string) string {
|
||||
if isPostgresDialect(db) {
|
||||
return distinctNonEmptyCountSQLPostgres(column)
|
||||
}
|
||||
return distinctNonEmptyCountSQLSQLite(column)
|
||||
}
|
||||
|
||||
// dailyTrendDateSQL 按当前方言返回按日聚合的日期表达式（运行时按 Dialector 分发，默认 SQLite）。
|
||||
func dailyTrendDateSQL(db *gorm.DB) string {
|
||||
if isPostgresDialect(db) {
|
||||
return dailyTrendDateSQLPostgres()
|
||||
}
|
||||
return dailyTrendDateSQLSQLite()
|
||||
}
|
||||
|
||||
// epochSQL 按当前方言返回 epoch 秒表达式（运行时按 Dialector 分发，默认 SQLite）。
|
||||
func epochSQL(db *gorm.DB, column string) string {
|
||||
if isPostgresDialect(db) {
|
||||
return epochSQLPostgres(column)
|
||||
}
|
||||
return epochSQLSQLite(column)
|
||||
}
|
||||
|
||||
// textCastSQL 按当前方言返回数值列转文本表达式（运行时按 Dialector 分发，默认 SQLite）。
|
||||
func textCastSQL(db *gorm.DB, column string) string {
|
||||
if isPostgresDialect(db) {
|
||||
return textCastSQLPostgres(column)
|
||||
}
|
||||
return textCastSQLSQLite(column)
|
||||
}
|
||||
|
||||
// timeBucketSQL 按当前方言返回时间分桶表达式。
|
||||
// brief 将 PG/SQLite 两版写为同名函数，同包无法共存；log_database 为运行时配置，
|
||||
// 不能使用编译期 build tag，故按 db.Dialector.Name() 运行时分发（默认 SQLite）。
|
||||
func timeBucketSQL(db *gorm.DB, column string, bucketSeconds int64) string {
|
||||
if isPostgresDialect(db) {
|
||||
return timeBucketSQLPostgres(column, bucketSeconds)
|
||||
}
|
||||
return timeBucketSQLSQLite(column, bucketSeconds)
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
|
||||
)
|
||||
|
||||
// AccessLogHooks 节点访问日志异步入队回调（由 chwriter 装配）。
|
||||
type AccessLogHooks struct {
|
||||
QueueNodeAccessLogs func(logs []analyticsmodel.NodeAccessLog)
|
||||
}
|
||||
|
||||
// ObservabilityHooks 可观测异步入队回调（由 chwriter 装配）。
|
||||
type ObservabilityHooks struct {
|
||||
QueueMetricSnapshot func(record analyticsmodel.NodeMetricSnapshot)
|
||||
QueueEdgeHealth func(record analyticsmodel.NodeEdgeHealth)
|
||||
QueueNodeObsFrps func(record analyticsmodel.NodeObsFrps)
|
||||
QueueNodeObsFrpc func(record analyticsmodel.NodeObsFrpc)
|
||||
}
|
||||
|
||||
var (
|
||||
hooksMu sync.RWMutex
|
||||
accessLogHooks AccessLogHooks
|
||||
observabilityHooks ObservabilityHooks
|
||||
)
|
||||
|
||||
// SetAccessLogHooks 注册节点访问日志异步入队回调。
|
||||
func SetAccessLogHooks(h AccessLogHooks) {
|
||||
hooksMu.Lock()
|
||||
accessLogHooks = h
|
||||
hooksMu.Unlock()
|
||||
}
|
||||
|
||||
// SetObservabilityHooks 注册可观测异步入队回调。
|
||||
func SetObservabilityHooks(h ObservabilityHooks) {
|
||||
hooksMu.Lock()
|
||||
observabilityHooks = h
|
||||
hooksMu.Unlock()
|
||||
}
|
||||
|
||||
// currentAccessLogHooks 返回当前 hooks 快照（未注册时为 zero value，调用方判空跳过）。
|
||||
func currentAccessLogHooks() AccessLogHooks {
|
||||
hooksMu.RLock()
|
||||
defer hooksMu.RUnlock()
|
||||
return accessLogHooks
|
||||
}
|
||||
|
||||
// currentObservabilityHooks 返回当前 hooks 快照（未注册时为 zero value，调用方判空跳过）。
|
||||
func currentObservabilityHooks() ObservabilityHooks {
|
||||
hooksMu.RLock()
|
||||
defer hooksMu.RUnlock()
|
||||
return observabilityHooks
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"os/exec"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// forbiddenImports 上层应用禁止直接触碰的底层日志实现。
|
||||
var forbiddenImports = []string{
|
||||
"github.com/Rain-kl/Wavelet/internal/repository/analytics",
|
||||
}
|
||||
|
||||
// allowedAnalyticsDelegation 允许直接依赖 analyticsrepo 的委托层：
|
||||
// - internal/repository：持久化门面，ListOpenFlareLatestMetricSnapshotsSince 的
|
||||
// CH 快速路径仍直连 analyticsrepo（LIMIT 1 BY node_id）；小时级聚合读已改走 logstore；
|
||||
// - internal/repository/logstore：CH 后端实现按设计委托 analyticsrepo。
|
||||
//
|
||||
// 除此之外，依赖闭包内任何包都禁止引入 analyticsrepo。
|
||||
var allowedAnalyticsDelegation = map[string]bool{
|
||||
"github.com/Rain-kl/Wavelet/internal/repository": true,
|
||||
"github.com/Rain-kl/Wavelet/internal/repository/logstore": true,
|
||||
}
|
||||
|
||||
// allowedInfraPersistence 允许 apps 引入的 infra/persistence 子包。
|
||||
var allowedInfraPersistence = []string{
|
||||
"github.com/Rain-kl/Wavelet/internal/infra/persistence/batchwriter", // batchwriter 统计类型
|
||||
"github.com/Rain-kl/Wavelet/internal/infra/persistence/idgen", // 雪花 ID 生成（无日志依赖）
|
||||
}
|
||||
|
||||
func TestAppsMustNotImportLogBackendDirectly(t *testing.T) {
|
||||
t.Chdir("../../..") // module root，保证 ./internal/apps/... 可解析
|
||||
// -test 同时列出测试二进制依赖，覆盖仅测试文件引入的底层日志实现。
|
||||
out, err := exec.Command("go", "list", "-test", "-deps", "-f", `{{.ImportPath}} {{join .Imports " "}}`, "./internal/apps/...").Output()
|
||||
if err != nil {
|
||||
t.Fatalf("go list: %v", err)
|
||||
}
|
||||
for _, line := range strings.Split(string(out), "\n") {
|
||||
fields := strings.Fields(line)
|
||||
if len(fields) == 0 {
|
||||
continue
|
||||
}
|
||||
pkg := fields[0]
|
||||
for _, forbidden := range forbiddenImports {
|
||||
for _, imp := range fields[1:] {
|
||||
if imp == forbidden && !allowedAnalyticsDelegation[pkg] {
|
||||
t.Errorf("internal/apps must not import %s (via %s)", forbidden, pkg)
|
||||
}
|
||||
}
|
||||
}
|
||||
if strings.HasPrefix(pkg, "github.com/Rain-kl/Wavelet/internal/infra/persistence/") {
|
||||
allowed := false
|
||||
for _, a := range allowedInfraPersistence {
|
||||
if pkg == a || strings.HasPrefix(pkg, a+"/") {
|
||||
allowed = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !allowed {
|
||||
t.Errorf("internal/apps must not import infra/persistence subpackage: %s", pkg)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/model"
|
||||
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
|
||||
)
|
||||
|
||||
// TestGormAccessLogInsertBatchHooks 覆盖访问日志写入入口：
|
||||
// 冻结检查、hook 入队、不直接落库、flush 后可见（行为与旧 repository clickhouse 包装一致）。
|
||||
func TestGormAccessLogInsertBatchHooks(t *testing.T) {
|
||||
ResetForTest()
|
||||
SetConfigReader(func(_ context.Context, key string) (string, error) {
|
||||
return "", nil
|
||||
})
|
||||
defer ResetForTest()
|
||||
|
||||
s := newTestGormStore(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
|
||||
var hooked []analyticsmodel.NodeAccessLog
|
||||
SetAccessLogHooks(AccessLogHooks{
|
||||
QueueNodeAccessLogs: func(logs []analyticsmodel.NodeAccessLog) {
|
||||
hooked = append(hooked, logs...)
|
||||
},
|
||||
})
|
||||
defer SetAccessLogHooks(AccessLogHooks{})
|
||||
|
||||
records := []*model.OpenFlareAccessLog{
|
||||
{NodeID: "n1", LoggedAt: now, RemoteAddr: "1.1.1.1", StatusCode: 200, BytesSent: 100},
|
||||
{NodeID: "n1", LoggedAt: now, RemoteAddr: "2.2.2.2", StatusCode: 404},
|
||||
}
|
||||
if err := s.InsertBatch(ctx, records); err != nil {
|
||||
t.Fatalf("insert batch: %v", err)
|
||||
}
|
||||
if len(hooked) != 2 || hooked[0].RemoteAddr != "1.1.1.1" || hooked[0].BytesSent != 100 || hooked[1].StatusCode != 404 {
|
||||
t.Fatalf("hook rows mismatch: %+v", hooked)
|
||||
}
|
||||
// 写入入口只入队、不直接落库。
|
||||
rows, err := s.List(ctx, model.OpenFlareAccessLogQuery{NodeID: "n1"})
|
||||
if err != nil {
|
||||
t.Fatalf("list: %v", err)
|
||||
}
|
||||
if len(rows) != 0 {
|
||||
t.Fatalf("entry insert must not write rows, got %d", len(rows))
|
||||
}
|
||||
// flush 后可见。
|
||||
if err := s.BatchInsertNodeAccessLogs(ctx, hooked); err != nil {
|
||||
t.Fatalf("flush: %v", err)
|
||||
}
|
||||
rows, err = s.List(ctx, model.OpenFlareAccessLogQuery{NodeID: "n1"})
|
||||
if err != nil {
|
||||
t.Fatalf("list after flush: %v", err)
|
||||
}
|
||||
if len(rows) != 2 {
|
||||
t.Fatalf("list after flush want 2, got %d", len(rows))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
// Package logstore 提供日志/分析存储抽象：上层只面向本包接口，
|
||||
// 禁止直接 import internal/repository/analytics 或触碰 db.ChConn/db.ChDB。
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/model"
|
||||
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
|
||||
)
|
||||
|
||||
// ErrMigrating 表示日志数据库正在迁移，当前禁止写入。
|
||||
var ErrMigrating = errors.New("log database is migrating, writes are disabled")
|
||||
|
||||
// AccessLogStore 节点访问日志（of_node_access_logs）。
|
||||
type AccessLogStore interface {
|
||||
// InsertBatch 为写入入口：冻结检查 + 经 hook 入队（异步），不直接落库。
|
||||
InsertBatch(ctx context.Context, records []*model.OpenFlareAccessLog) error
|
||||
// BatchInsertNodeAccessLogs 为 batchwriter flush 目标：直接批量写入当前存储。
|
||||
BatchInsertNodeAccessLogs(ctx context.Context, rows []analyticsmodel.NodeAccessLog) error
|
||||
|
||||
List(ctx context.Context, query model.OpenFlareAccessLogQuery) ([]*model.OpenFlareAccessLog, error)
|
||||
Count(ctx context.Context, query model.OpenFlareAccessLogQuery) (int64, int64, int64, error)
|
||||
RegionCounts(ctx context.Context, nodeID string, since time.Time, limit int) ([]*model.OpenFlareAccessLogRegionCount, error)
|
||||
BucketAggregates(ctx context.Context, filter model.OpenFlareAccessLogQuery, bucketSeconds int64) ([]analyticsmodel.NodeAccessLogBucketAggregate, error)
|
||||
CountBuckets(ctx context.Context, filter model.OpenFlareAccessLogQuery, bucketSeconds int64) (int64, error)
|
||||
BucketDimensions(ctx context.Context, filter model.OpenFlareAccessLogQuery, column string, bucketSeconds int64) ([]analyticsmodel.NodeAccessLogBucketDimension, error)
|
||||
IPAggregates(ctx context.Context, filter model.OpenFlareAccessLogQuery, exactRemoteAddr bool) ([]analyticsmodel.NodeAccessLogIPAggregate, error)
|
||||
IPSummaries(ctx context.Context, filter model.OpenFlareAccessLogQuery, recentSince time.Time) ([]analyticsmodel.NodeAccessLogIPSummary, error)
|
||||
CountIPSummaries(ctx context.Context, filter model.OpenFlareAccessLogQuery) (int64, error)
|
||||
WAFIPAggregates(ctx context.Context, filter model.OpenFlareAccessLogQuery) ([]analyticsmodel.NodeAccessLogWAFIPAggregate, error)
|
||||
IPTrend(ctx context.Context, filter model.OpenFlareAccessLogQuery, bucketSeconds int64) ([]analyticsmodel.NodeAccessLogIPTrend, error)
|
||||
TrafficSummary(ctx context.Context, filter model.OpenFlareAccessLogQuery) (model.OpenFlareAccessLogTrafficSummary, error)
|
||||
ValueCounts(ctx context.Context, filter model.OpenFlareAccessLogQuery, column string, limit int) ([]model.OpenFlareAccessLogValueCount, error)
|
||||
NodeAggregates(ctx context.Context, filter model.OpenFlareAccessLogQuery) ([]model.OpenFlareAccessLogNodeAggregate, error)
|
||||
DeleteAll(ctx context.Context) (int64, error)
|
||||
DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error)
|
||||
DeleteByNodeBefore(ctx context.Context, nodeID string, before time.Time) (int64, error)
|
||||
// ListForMigration 按 id 升序分页读取（迁移复制用）。
|
||||
ListForMigration(ctx context.Context, afterID uint64, limit int) ([]analyticsmodel.NodeAccessLog, error)
|
||||
// MigrationRange 返回源表 logged_at 的最小/最大值（空表返回零值），迁移预建分区用。
|
||||
MigrationRange(ctx context.Context) (from, to time.Time, err error)
|
||||
// EnsurePartitions 幂等预建 PG 分区（按月），覆盖 [from, to] 月份；CH/SQLite 为 no-op。
|
||||
// 目标为 PG 的迁移在复制前调用，避免历史数据写入报 "no partition of relation found"。
|
||||
EnsurePartitions(ctx context.Context, from, to time.Time) error
|
||||
}
|
||||
|
||||
// ObservabilityStore 可观测 4 表（metric snapshots / edge health / frps / frpc）。
|
||||
type ObservabilityStore interface {
|
||||
InsertMetricSnapshot(ctx context.Context, record *model.OpenFlareMetricSnapshot) error
|
||||
ListMetricSnapshots(ctx context.Context, nodeID string, since time.Time, limit int) ([]*model.OpenFlareMetricSnapshot, error)
|
||||
DeleteAllMetricSnapshots(ctx context.Context) (int64, error)
|
||||
DeleteMetricSnapshotsBefore(ctx context.Context, cutoff time.Time) (int64, error)
|
||||
BatchInsertNodeMetricSnapshots(ctx context.Context, rows []analyticsmodel.NodeMetricSnapshot) error
|
||||
|
||||
// ListTrafficHourly 返回小时级流量汇总（按 node/hour 聚合，unique_visitor_count 恒 0）。
|
||||
// CH 后端读 of_access_log_hourly rollup；PG/SQLite 从 of_node_access_logs 实时聚合。
|
||||
ListTrafficHourly(ctx context.Context, nodeID string, since time.Time) ([]analyticsmodel.NodeTrafficHourly, error)
|
||||
// ListAccessLogHourly 返回按 node/hour/host 的小时级访问日志汇总。
|
||||
ListAccessLogHourly(ctx context.Context, nodeID string, since time.Time) ([]analyticsmodel.AccessLogHourly, error)
|
||||
// ListMetricHourly 返回小时级指标聚合（avg cpu/memory + 计数器增量，reported_nodes 去重节点数）。
|
||||
ListMetricHourly(ctx context.Context, nodeID string, since time.Time) ([]analyticsmodel.NodeMetricHourly, error)
|
||||
|
||||
InsertEdgeHealth(ctx context.Context, record *model.OpenFlareEdgeHealth) error
|
||||
ListEdgeHealth(ctx context.Context, nodeID string, since time.Time, limit int) ([]*model.OpenFlareEdgeHealth, error)
|
||||
DeleteAllEdgeHealth(ctx context.Context) (int64, error)
|
||||
DeleteEdgeHealthBefore(ctx context.Context, cutoff time.Time) (int64, error)
|
||||
BatchInsertNodeEdgeHealth(ctx context.Context, rows []analyticsmodel.NodeEdgeHealth) error
|
||||
|
||||
InsertNodeObservationFrps(ctx context.Context, record *model.OpenFlareNodeObservationFrps) error
|
||||
ListNodeObservationFrps(ctx context.Context, nodeID string, since time.Time, limit int) ([]*model.OpenFlareNodeObservationFrps, error)
|
||||
DeleteAllNodeObservationFrps(ctx context.Context) (int64, error)
|
||||
DeleteNodeObservationFrpsBefore(ctx context.Context, cutoff time.Time) (int64, error)
|
||||
BatchInsertNodeObsFrps(ctx context.Context, rows []analyticsmodel.NodeObsFrps) error
|
||||
|
||||
InsertNodeObservationFrpc(ctx context.Context, record *model.OpenFlareNodeObservationFrpc) error
|
||||
ListNodeObservationFrpc(ctx context.Context, nodeID string, since time.Time, limit int) ([]*model.OpenFlareNodeObservationFrpc, error)
|
||||
DeleteAllNodeObservationFrpc(ctx context.Context) (int64, error)
|
||||
DeleteNodeObservationFrpcBefore(ctx context.Context, cutoff time.Time) (int64, error)
|
||||
BatchInsertNodeObsFrpc(ctx context.Context, rows []analyticsmodel.NodeObsFrpc) error
|
||||
|
||||
// 迁移复制用：按 id 升序分页读取。
|
||||
ListMetricSnapshotsForMigration(ctx context.Context, afterID uint64, limit int) ([]analyticsmodel.NodeMetricSnapshot, error)
|
||||
ListEdgeHealthForMigration(ctx context.Context, afterID uint64, limit int) ([]analyticsmodel.NodeEdgeHealth, error)
|
||||
ListNodeObsFrpsForMigration(ctx context.Context, afterID uint64, limit int) ([]analyticsmodel.NodeObsFrps, error)
|
||||
ListNodeObsFrpcForMigration(ctx context.Context, afterID uint64, limit int) ([]analyticsmodel.NodeObsFrpc, error)
|
||||
}
|
||||
|
||||
// UserAccessLogStore 用户访问日志（w_user_access_logs）。
|
||||
type UserAccessLogStore interface {
|
||||
BatchInsert(ctx context.Context, logs []analyticsmodel.UserAccessLog) error
|
||||
// DeleteAll 清空全部用户访问日志（迁移「覆盖目标库已有日志」幂等前提用）。
|
||||
DeleteAll(ctx context.Context) (int64, error)
|
||||
Count(ctx context.Context, filter analyticsmodel.AccessLogFilter) (uint64, error)
|
||||
List(ctx context.Context, filter analyticsmodel.AccessLogFilter, page, pageSize int) ([]analyticsmodel.UserAccessLog, uint64, error)
|
||||
GetDailyTrend(ctx context.Context, days int) ([]analyticsmodel.DailyTrend, error)
|
||||
GetBrowserDistribution(ctx context.Context, startTime time.Time) ([]analyticsmodel.BrowserShare, error)
|
||||
GetTopActiveUsers(ctx context.Context, startTime time.Time, limit int) ([]analyticsmodel.TopUser, error)
|
||||
// ListForMigration 按 id 升序分页读取（迁移复制用）。
|
||||
ListForMigration(ctx context.Context, afterID uint64, limit int) ([]analyticsmodel.UserAccessLog, error)
|
||||
// MigrationRange 返回源表 created_at 的最小/最大值（空表返回零值），迁移预建分区用。
|
||||
MigrationRange(ctx context.Context) (from, to time.Time, err error)
|
||||
}
|
||||
|
||||
// StatusStore 日志库状态（供管理端状态端点）。
|
||||
type StatusStore interface {
|
||||
ActiveDatabase(ctx context.Context) (string, error)
|
||||
ClickHouseOperationalStats(ctx context.Context) (*analyticsmodel.ClickHouseOperationalStats, error) // 仅 CH 激活时非 nil
|
||||
}
|
||||
|
||||
// Store 聚合当前生效日志库的全部域存储。
|
||||
type Store struct {
|
||||
AccessLogs AccessLogStore
|
||||
Observability ObservabilityStore
|
||||
UserAccessLogs UserAccessLogStore
|
||||
Status StatusStore
|
||||
}
|
||||
@@ -0,0 +1,183 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"regexp"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gorm.io/driver/postgres"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/logger"
|
||||
|
||||
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
|
||||
)
|
||||
|
||||
// TestEnsurePartitionsPostgresInsertAcrossMonths 需要 TEST_POSTGRES_DSN（未设置时跳过）：
|
||||
// 验证 EnsurePartitions 预建任意月份范围分区后，跨月历史数据可写入 PG 分区表
|
||||
// （对应迁移任务从 CH/SQLite 复制历史日志到 PG 时先预建分区的场景）。
|
||||
func TestEnsurePartitionsPostgresInsertAcrossMonths(t *testing.T) {
|
||||
dsn := strings.TrimSpace(os.Getenv("TEST_POSTGRES_DSN"))
|
||||
if dsn == "" {
|
||||
t.Skip("TEST_POSTGRES_DSN is not set")
|
||||
}
|
||||
|
||||
gdb, err := gorm.Open(postgres.Open(dsn), &gorm.Config{
|
||||
DisableForeignKeyConstraintWhenMigrating: true,
|
||||
Logger: logger.Default.LogMode(logger.Silent),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("open postgres: %v", err)
|
||||
}
|
||||
sqlDB, err := gdb.DB()
|
||||
if err != nil {
|
||||
t.Fatalf("sql db: %v", err)
|
||||
}
|
||||
sqlDB.SetMaxOpenConns(1)
|
||||
|
||||
schema := fmt.Sprintf("logstore_partition_%d", time.Now().UnixNano())
|
||||
if !regexp.MustCompile(`^[a-z0-9_]+$`).MatchString(schema) {
|
||||
t.Fatalf("invalid schema: %s", schema)
|
||||
}
|
||||
if err := gdb.Exec(`CREATE SCHEMA "` + schema + `"`).Error; err != nil {
|
||||
t.Fatalf("create schema: %v", err)
|
||||
}
|
||||
if err := gdb.Exec(`SET search_path TO "` + schema + `"`).Error; err != nil {
|
||||
t.Fatalf("set search_path: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
_ = gdb.Exec("SET search_path TO public").Error
|
||||
_ = gdb.Exec(`DROP SCHEMA IF EXISTS "` + schema + `" CASCADE`).Error
|
||||
_ = sqlDB.Close()
|
||||
})
|
||||
|
||||
// 与 goose/postgres/202608080001_create_log_tables.sql 保持一致的分区父表 DDL。
|
||||
for _, ddl := range []string{postgresNodeAccessLogsDDL, postgresUserAccessLogsDDL} {
|
||||
if err := gdb.Exec(ddl).Error; err != nil {
|
||||
t.Fatalf("create partitioned table: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
ResetForTest()
|
||||
SetConfigReader(func(_ context.Context, _ string) (string, error) { return "", nil })
|
||||
defer ResetForTest()
|
||||
|
||||
ctx := context.Background()
|
||||
store := newGormStore(gdb)
|
||||
ua := newUserAccessLogGormStore(gdb)
|
||||
|
||||
// 源范围跨 3 个月：2026-01-10 ~ 2026-03-20；to+1 月兜底生成 202601..202604 分区。
|
||||
from := time.Date(2026, 1, 10, 8, 0, 0, 0, time.UTC)
|
||||
max := time.Date(2026, 3, 20, 9, 30, 0, 0, time.UTC)
|
||||
if err := store.EnsurePartitions(ctx, from, max.AddDate(0, 1, 0)); err != nil {
|
||||
t.Fatalf("EnsurePartitions: %v", err)
|
||||
}
|
||||
|
||||
// 幂等：重复调用不报错（CREATE TABLE IF NOT EXISTS ... PARTITION OF）。
|
||||
if err := store.EnsurePartitions(ctx, from, max.AddDate(0, 1, 0)); err != nil {
|
||||
t.Fatalf("EnsurePartitions idempotent: %v", err)
|
||||
}
|
||||
|
||||
var partitionCount int64
|
||||
if err := gdb.Raw(
|
||||
"SELECT count(*) FROM pg_inherits WHERE inhrelid = to_regclass('of_node_access_logs')",
|
||||
).Scan(&partitionCount).Error; err != nil {
|
||||
t.Fatalf("count partitions: %v", err)
|
||||
}
|
||||
if partitionCount != 4 {
|
||||
t.Fatalf("of_node_access_logs partitions = %d, want 4", partitionCount)
|
||||
}
|
||||
|
||||
// 跨月插入：1/2/3 月各 2 条节点访问日志 + 2 条用户访问日志，均应命中已有分区。
|
||||
nodeRows := []analyticsmodel.NodeAccessLog{
|
||||
{ID: 1, NodeID: "n1", LoggedAt: time.Date(2026, 1, 15, 0, 0, 0, 0, time.UTC), RemoteAddr: "1.1.1.1"},
|
||||
{ID: 2, NodeID: "n1", LoggedAt: time.Date(2026, 1, 20, 0, 0, 0, 0, time.UTC), RemoteAddr: "1.1.1.2"},
|
||||
{ID: 3, NodeID: "n2", LoggedAt: time.Date(2026, 2, 10, 0, 0, 0, 0, time.UTC), RemoteAddr: "2.2.2.2"},
|
||||
{ID: 4, NodeID: "n2", LoggedAt: time.Date(2026, 2, 12, 0, 0, 0, 0, time.UTC), RemoteAddr: "2.2.2.3"},
|
||||
{ID: 5, NodeID: "n1", LoggedAt: time.Date(2026, 3, 5, 0, 0, 0, 0, time.UTC), RemoteAddr: "3.3.3.3"},
|
||||
{ID: 6, NodeID: "n1", LoggedAt: time.Date(2026, 3, 18, 0, 0, 0, 0, time.UTC), RemoteAddr: "3.3.3.4"},
|
||||
}
|
||||
if err := store.BatchInsertNodeAccessLogs(ctx, nodeRows); err != nil {
|
||||
t.Fatalf("insert node access logs across months: %v", err)
|
||||
}
|
||||
|
||||
userRows := []analyticsmodel.UserAccessLog{
|
||||
{ID: 1, UserID: 101, Path: "/a", CreatedAt: time.Date(2026, 1, 16, 0, 0, 0, 0, time.UTC)},
|
||||
{ID: 2, UserID: 102, Path: "/b", CreatedAt: time.Date(2026, 3, 17, 0, 0, 0, 0, time.UTC)},
|
||||
}
|
||||
if err := ua.BatchInsert(ctx, userRows); err != nil {
|
||||
t.Fatalf("insert user access logs across months: %v", err)
|
||||
}
|
||||
|
||||
var nodeCount, userCount int64
|
||||
if err := gdb.Model(&analyticsmodel.NodeAccessLog{}).Count(&nodeCount).Error; err != nil {
|
||||
t.Fatalf("count node access logs: %v", err)
|
||||
}
|
||||
if err := gdb.Model(&analyticsmodel.UserAccessLog{}).Count(&userCount).Error; err != nil {
|
||||
t.Fatalf("count user access logs: %v", err)
|
||||
}
|
||||
if nodeCount != 6 {
|
||||
t.Fatalf("node access log count = %d, want 6", nodeCount)
|
||||
}
|
||||
if userCount != 2 {
|
||||
t.Fatalf("user access log count = %d, want 2", userCount)
|
||||
}
|
||||
|
||||
// MigrationRange 返回跨月范围（覆盖两表）。
|
||||
gotFrom, gotTo, err := store.MigrationRange(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("node MigrationRange: %v", err)
|
||||
}
|
||||
if !gotFrom.Equal(time.Date(2026, 1, 15, 0, 0, 0, 0, time.UTC)) || !gotTo.Equal(time.Date(2026, 3, 18, 0, 0, 0, 0, time.UTC)) {
|
||||
t.Fatalf("node MigrationRange = %s ~ %s, want 2026-01-15 ~ 2026-03-18", gotFrom, gotTo)
|
||||
}
|
||||
uaFrom, uaTo, err := ua.MigrationRange(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("user MigrationRange: %v", err)
|
||||
}
|
||||
if !uaFrom.Equal(time.Date(2026, 1, 16, 0, 0, 0, 0, time.UTC)) || !uaTo.Equal(time.Date(2026, 3, 17, 0, 0, 0, 0, time.UTC)) {
|
||||
t.Fatalf("user MigrationRange = %s ~ %s", uaFrom, uaTo)
|
||||
}
|
||||
}
|
||||
|
||||
// postgresNodeAccessLogsDDL 与 goose/postgres/202608080001_create_log_tables.sql 对齐。
|
||||
const postgresNodeAccessLogsDDL = `
|
||||
CREATE TABLE IF NOT EXISTS of_node_access_logs (
|
||||
id BIGINT NOT NULL,
|
||||
node_id VARCHAR(64) NOT NULL DEFAULT '',
|
||||
logged_at TIMESTAMPTZ NOT NULL,
|
||||
remote_addr VARCHAR(128) NOT NULL DEFAULT '',
|
||||
region VARCHAR(128) NOT NULL DEFAULT '',
|
||||
host VARCHAR(255) NOT NULL DEFAULT '',
|
||||
path VARCHAR(2048) NOT NULL DEFAULT '',
|
||||
user_agent TEXT NOT NULL DEFAULT '',
|
||||
cache_status VARCHAR(64) NOT NULL DEFAULT '',
|
||||
status_code INTEGER NOT NULL DEFAULT 0,
|
||||
bytes_sent BIGINT NOT NULL DEFAULT 0,
|
||||
request_length BIGINT NOT NULL DEFAULT 0,
|
||||
request_time_ms INTEGER NOT NULL DEFAULT 0,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
PRIMARY KEY (id, logged_at)
|
||||
) PARTITION BY RANGE (logged_at)`
|
||||
|
||||
// postgresUserAccessLogsDDL 与 goose/postgres/202608080001_create_log_tables.sql 对齐。
|
||||
const postgresUserAccessLogsDDL = `
|
||||
CREATE TABLE IF NOT EXISTS w_user_access_logs (
|
||||
id BIGINT NOT NULL,
|
||||
user_id BIGINT NOT NULL DEFAULT 0,
|
||||
path VARCHAR(2048) NOT NULL DEFAULT '',
|
||||
method VARCHAR(16) NOT NULL DEFAULT '',
|
||||
ip VARCHAR(128) NOT NULL DEFAULT '',
|
||||
user_agent TEXT NOT NULL DEFAULT '',
|
||||
headers TEXT NOT NULL DEFAULT '',
|
||||
status INTEGER NOT NULL DEFAULT 0,
|
||||
latency BIGINT NOT NULL DEFAULT 0,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
PRIMARY KEY (id, created_at)
|
||||
) PARTITION BY RANGE (created_at)`
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,170 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/infra/config"
|
||||
db "github.com/Rain-kl/Wavelet/internal/infra/persistence"
|
||||
"github.com/Rain-kl/Wavelet/internal/model"
|
||||
)
|
||||
|
||||
// logDatabaseKey / logMigrationKey 对应 model.ConfigKeyLogDatabase / ConfigKeyLogDBMigration。
|
||||
const (
|
||||
logDatabaseKey = model.ConfigKeyLogDatabase
|
||||
logMigrationKey = model.ConfigKeyLogDBMigration
|
||||
)
|
||||
|
||||
// 日志库名常量（与 model 配置值一致，集中避免散落字符串字面量）。
|
||||
const (
|
||||
dbNamePostgres = "postgres"
|
||||
dbNameSQLite = "sqlite"
|
||||
dbNameClickHouse = "clickhouse"
|
||||
)
|
||||
|
||||
// errConfigReaderNotWired 表示 config reader 尚未注入（首启/测试场景按 seed 规则兜底）。
|
||||
var errConfigReaderNotWired = errors.New("logstore: config reader not wired")
|
||||
|
||||
// ConfigReader 读取系统配置字符串值，由 bootstrap 注入（避免 logstore ↔ repository 循环依赖）。
|
||||
type ConfigReader func(ctx context.Context, key string) (string, error)
|
||||
|
||||
var (
|
||||
configReader ConfigReader
|
||||
|
||||
storeMu sync.RWMutex
|
||||
active *Store
|
||||
activeDB string
|
||||
)
|
||||
|
||||
// SetConfigReader 注入系统配置读取函数（bootstrap 调用，测试可注入内存实现）。
|
||||
func SetConfigReader(fn ConfigReader) { configReader = fn }
|
||||
|
||||
func getConfig(ctx context.Context, key string) (string, error) {
|
||||
if configReader == nil {
|
||||
return "", errConfigReaderNotWired
|
||||
}
|
||||
return configReader(ctx, key)
|
||||
}
|
||||
|
||||
// Active 返回当前生效的日志库 Store。按 log_database 系统配置惰性解析并缓存，
|
||||
// 配置更新（含迁移任务翻转）后自动重建。
|
||||
func Active(ctx context.Context) (*Store, error) {
|
||||
current, err := resolveDatabase(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
storeMu.RLock()
|
||||
if active != nil && activeDB == current {
|
||||
s := active
|
||||
storeMu.RUnlock()
|
||||
return s, nil
|
||||
}
|
||||
storeMu.RUnlock()
|
||||
|
||||
storeMu.Lock()
|
||||
defer storeMu.Unlock()
|
||||
if active != nil && activeDB == current {
|
||||
return active, nil
|
||||
}
|
||||
s, err := buildStore(ctx, current, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
active = s
|
||||
activeDB = current
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// Build 直接按目标构造 store（不经 Active 缓存）。
|
||||
func Build(ctx context.Context, database string) (*Store, error) {
|
||||
return buildStore(ctx, database, false)
|
||||
}
|
||||
|
||||
// BuildForMigration 构造迁移目标 store：与 Build 相同但不做冻结检查
|
||||
// （迁移期间 log_db_migration=migrating 已冻结源库写入，目标库的清空/复制写入必须放行）。
|
||||
func BuildForMigration(ctx context.Context, database string) (*Store, error) {
|
||||
return buildStore(ctx, database, true)
|
||||
}
|
||||
|
||||
// buildStore 按目标构造实现。skipFreeze 为 true 时该 store 跳过冻结检查
|
||||
// （仅迁移任务的目标 store 使用）。gorm 分支 UserAccessLogs 用独立包装类型
|
||||
// （gormLogStore 已占用 List/Count 方法名，无法再实现 UserAccessLogStore）。
|
||||
func buildStore(ctx context.Context, database string, skipFreeze bool) (*Store, error) {
|
||||
switch database {
|
||||
case dbNameClickHouse:
|
||||
ch := newClickHouseStore()
|
||||
ch.skipFreeze = skipFreeze
|
||||
return &Store{
|
||||
AccessLogs: ch,
|
||||
Observability: ch,
|
||||
UserAccessLogs: newClickHouseUserAccessLogStore(),
|
||||
Status: ch,
|
||||
}, nil
|
||||
case dbNamePostgres, dbNameSQLite:
|
||||
gdb := db.DB(ctx)
|
||||
g := newGormStore(gdb)
|
||||
g.skipFreeze = skipFreeze
|
||||
return &Store{
|
||||
AccessLogs: g,
|
||||
Observability: g,
|
||||
UserAccessLogs: newUserAccessLogGormStore(gdb),
|
||||
Status: g,
|
||||
}, nil
|
||||
default:
|
||||
return nil, fmt.Errorf("unsupported log database: %s", database)
|
||||
}
|
||||
}
|
||||
|
||||
// Migrating 返回日志库是否处于迁移冻结状态。
|
||||
func Migrating(ctx context.Context) bool {
|
||||
v, err := getConfig(ctx, logMigrationKey)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
return v == "migrating"
|
||||
}
|
||||
|
||||
// Init 在 bootstrap 阶段预热一次激活 store（幂等，失败不致命——首次使用时再解析）。
|
||||
func Init(ctx context.Context) {
|
||||
_, _ = Active(ctx)
|
||||
}
|
||||
|
||||
// ResetForTest 清空缓存的激活 store 与 config reader，便于测试注入。
|
||||
func ResetForTest() {
|
||||
storeMu.Lock()
|
||||
active = nil
|
||||
activeDB = ""
|
||||
storeMu.Unlock()
|
||||
configReader = nil
|
||||
}
|
||||
|
||||
// ActiveDatabase 返回当前日志主库名（postgres|sqlite|clickhouse）。
|
||||
func ActiveDatabase(ctx context.Context) (string, error) {
|
||||
return resolveDatabase(ctx)
|
||||
}
|
||||
|
||||
// resolveDatabase 读取 log_database：值缺失或 reader 未装配（首启）时按启动规则 seed；
|
||||
// 已装配 reader 的真实读取错误直接透出，避免把读失败当首次启动。
|
||||
func resolveDatabase(ctx context.Context) (string, error) {
|
||||
v, err := getConfig(ctx, logDatabaseKey)
|
||||
if err != nil && !errors.Is(err, errConfigReaderNotWired) {
|
||||
return "", err
|
||||
}
|
||||
if v != "" {
|
||||
return v, nil
|
||||
}
|
||||
// 首次启动 seed：CH 启用 → clickhouse；否则随主库。
|
||||
defaultDB := dbNameSQLite
|
||||
if config.Config.Database.Enabled {
|
||||
defaultDB = dbNamePostgres
|
||||
}
|
||||
if config.Config.ClickHouse.Enabled {
|
||||
defaultDB = dbNameClickHouse
|
||||
}
|
||||
return defaultDB, nil
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
|
||||
)
|
||||
|
||||
func TestMigratingReadsConfig(t *testing.T) {
|
||||
ResetForTest()
|
||||
SetConfigReader(func(_ context.Context, key string) (string, error) {
|
||||
if key == logMigrationKey {
|
||||
return "migrating", nil
|
||||
}
|
||||
return "", nil
|
||||
})
|
||||
if !Migrating(context.Background()) {
|
||||
t.Fatal("Migrating() = false, want true when key=migrating")
|
||||
}
|
||||
SetConfigReader(func(_ context.Context, key string) (string, error) {
|
||||
return "", nil
|
||||
})
|
||||
if Migrating(context.Background()) {
|
||||
t.Fatal("Migrating() = true, want false when key empty")
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveDatabaseDefaults(t *testing.T) {
|
||||
ResetForTest()
|
||||
// 配置缺失（reader 返回空值）时按主库规则 seed（config.Config 默认值由既有测试基建决定）。
|
||||
SetConfigReader(func(_ context.Context, key string) (string, error) {
|
||||
return "", nil
|
||||
})
|
||||
got, err := resolveDatabase(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("resolveDatabase: %v", err)
|
||||
}
|
||||
if got != "postgres" && got != "sqlite" && got != "clickhouse" {
|
||||
t.Fatalf("unexpected default log database: %s", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveDatabaseSurfacesReadError(t *testing.T) {
|
||||
ResetForTest()
|
||||
wantErr := errors.New("boom")
|
||||
SetConfigReader(func(_ context.Context, key string) (string, error) {
|
||||
return "", wantErr
|
||||
})
|
||||
if _, err := resolveDatabase(context.Background()); !errors.Is(err, wantErr) {
|
||||
t.Fatalf("resolveDatabase error = %v, want %v", err, wantErr)
|
||||
}
|
||||
}
|
||||
|
||||
func TestActiveBuildsStore(t *testing.T) {
|
||||
ResetForTest()
|
||||
SetConfigReader(func(_ context.Context, key string) (string, error) {
|
||||
if key == logDatabaseKey {
|
||||
return "sqlite", nil
|
||||
}
|
||||
return "", nil
|
||||
})
|
||||
store, err := Active(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("Active: %v", err)
|
||||
}
|
||||
if store == nil {
|
||||
t.Fatal("Active() returned nil store")
|
||||
}
|
||||
if store.AccessLogs == nil || store.Observability == nil || store.UserAccessLogs == nil || store.Status == nil {
|
||||
t.Fatalf("Active() store fields not fully wired: %+v", store)
|
||||
}
|
||||
// 再次调用应命中缓存。
|
||||
again, err := Active(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("Active (cached): %v", err)
|
||||
}
|
||||
if again != store {
|
||||
t.Fatal("Active() did not return cached store")
|
||||
}
|
||||
}
|
||||
|
||||
// TestClickHouseUserAccessLogBatchInsertFreeze 覆盖 CH 用户访问日志 flush 的冻结检查：
|
||||
// 冻结期非空批次返回 ErrMigrating（在触碰 CH 连接之前），空批次直接成功。
|
||||
func TestClickHouseUserAccessLogBatchInsertFreeze(t *testing.T) {
|
||||
ResetForTest()
|
||||
SetConfigReader(func(_ context.Context, key string) (string, error) {
|
||||
if key == logMigrationKey {
|
||||
return "migrating", nil
|
||||
}
|
||||
return "", nil
|
||||
})
|
||||
defer ResetForTest()
|
||||
s := newClickHouseUserAccessLogStore()
|
||||
ctx := context.Background()
|
||||
now := time.Now()
|
||||
|
||||
if err := s.BatchInsert(ctx, []analyticsmodel.UserAccessLog{{UserID: 1, CreatedAt: now}}); !errors.Is(err, ErrMigrating) {
|
||||
t.Fatalf("BatchInsert during migration: want ErrMigrating, got %v", err)
|
||||
}
|
||||
if err := s.BatchInsert(ctx, nil); err != nil {
|
||||
t.Fatalf("BatchInsert empty batch: %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user