diff --git a/docs/changelog/index.md b/docs/changelog/index.md index 0f2213ba..cef3b3eb 100644 --- a/docs/changelog/index.md +++ b/docs/changelog/index.md @@ -16,6 +16,11 @@ sidebar: false ## [Unreleased] +### 变更 + +- `of_node_access_logs` 从 PostgreSQL/SQLite 迁移至 ClickHouse(数据库 `openflare`);系统启动时强依赖 ClickHouse 连接与表结构自动初始化。 +- ClickHouse 默认启用(`clickhouse.enabled: true`),`docker-compose` 默认启动 `clickhouse` 服务并纳入 `wavelet` 健康依赖。 + ### 移除 - 删除旧版 `openflare-server/` 单体实现(含 `web/` 前端);仓库内 `openflare-server/` 现指迁移后的 Wavelet 统一控制面。 diff --git a/openflare-server/.env.example b/openflare-server/.env.example index ab495817..396e49c2 100644 --- a/openflare-server/.env.example +++ b/openflare-server/.env.example @@ -61,13 +61,11 @@ REDIS_ADDR=redis:6379 REDIS_KEY_PREFIX=openflare: # REDIS_POOL_SIZE=100 -# ─── ClickHouse(可选,默认关闭)────────────────────────────────────────── -# 设置 CLICKHOUSE_HOST 后自动启用,也可显式控制 -# CLICKHOUSE_ENABLED=false -# CLICKHOUSE_HOST=clickhouse:9000 -# CLICKHOUSE_USERNAME=default -# CLICKHOUSE_PASSWORD= -# CLICKHOUSE_NAME=openflare +# ─── ClickHouse(必需)──────────────────────────────────────────────────── +CLICKHOUSE_HOST=clickhouse:9000 +CLICKHOUSE_USERNAME=default +CLICKHOUSE_PASSWORD= +CLICKHOUSE_NAME=openflare # ─── 日志 ────────────────────────────────────────────────────────────────────── LOG_LEVEL=info diff --git a/openflare-server/config.example.yaml b/openflare-server/config.example.yaml index 111a778d..033f3d90 100644 --- a/openflare-server/config.example.yaml +++ b/openflare-server/config.example.yaml @@ -98,9 +98,9 @@ otel: tracer_name: "github.com/Rain-kl/OpenFlare" # Global tracer instrumentation name -# ─── ClickHouse (optional) ────────────────────────────────────────────────────── +# ─── ClickHouse (required) ────────────────────────────────────────────────────── clickhouse: - enabled: false + enabled: true hosts: - "127.0.0.1:9000" username: "default" diff --git a/openflare-server/docker-compose.yml b/openflare-server/docker-compose.yml index 4859467e..0952cfcf 100644 --- a/openflare-server/docker-compose.yml +++ b/openflare-server/docker-compose.yml @@ -23,6 +23,8 @@ services: condition: service_healthy redis: condition: service_healthy + clickhouse: + condition: service_healthy jaeger: condition: service_started @@ -69,26 +71,23 @@ services: - "${JAEGER_UI_PORT:-16686}:16686" - "${JAEGER_OTLP_GRPC_PORT:-4317}:4317" - "${JAEGER_OTLP_HTTP_PORT:-4318}:4318" -# -# clickhouse: -# image: clickhouse/clickhouse-server:25.3-alpine -# restart: unless-stopped -# profiles: -# - clickhouse -# environment: -# CLICKHOUSE_DB: ${CLICKHOUSE_DB:-openflare} -# CLICKHOUSE_USER: ${CLICKHOUSE_USER:-default} -# CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-123456} -# CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1 -# TZ: ${TZ:-Asia/Shanghai} -# ports: -# - "${CLICKHOUSE_HTTP_PORT:-8123}:8123" -# - "${CLICKHOUSE_NATIVE_PORT:-9000}:9000" -# volumes: -# - ./data/clickhouse_data:/var/lib/clickhouse -# healthcheck: -# test: ["CMD", "clickhouse-client", "--query", "SELECT 1"] -# interval: 10s -# timeout: 5s -# retries: 5 -# start_period: 15s + clickhouse: + image: clickhouse/clickhouse-server:25.3-alpine + restart: unless-stopped + environment: + CLICKHOUSE_DB: ${CLICKHOUSE_DB:-openflare} + CLICKHOUSE_USER: ${CLICKHOUSE_USER:-default} + CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-} + CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1 + TZ: ${TZ:-Asia/Shanghai} + ports: + - "${CLICKHOUSE_HTTP_PORT:-8123}:8123" + - "${CLICKHOUSE_NATIVE_PORT:-9000}:9000" + volumes: + - ./data/clickhouse_data:/var/lib/clickhouse + healthcheck: + test: ["CMD", "clickhouse-client", "--query", "SELECT 1"] + interval: 10s + timeout: 5s + retries: 5 + start_period: 15s diff --git a/openflare-server/internal/apps/admin/logs/routers.go b/openflare-server/internal/apps/admin/logs/routers.go index 78504343..5ca78969 100644 --- a/openflare-server/internal/apps/admin/logs/routers.go +++ b/openflare-server/internal/apps/admin/logs/routers.go @@ -16,7 +16,6 @@ import ( "time" "github.com/Rain-kl/Wavelet/internal/apps/admin" - "github.com/Rain-kl/Wavelet/internal/config" "github.com/Rain-kl/Wavelet/internal/db" "github.com/Rain-kl/Wavelet/internal/model" "github.com/Rain-kl/Wavelet/pkg/logger" @@ -288,9 +287,8 @@ func fetchAccessLogDetails(ctx context.Context, whereClause string, args []inter // @Failure 403 {object} response.Any "无管理员权限" // @Router /api/v1/admin/logs/access [get] func GetAccessLogs(c *gin.Context) { - // 1. 检查 ClickHouse 是否启用 - if !config.Config.ClickHouse.Enabled || db.ChConn == nil { - response.AbortWithError(c, http.StatusBadRequest, "ClickHouse 存储服务未启用,无法检索访问日志") + if db.ChConn == nil { + response.AbortWithError(c, http.StatusInternalServerError, "ClickHouse 未初始化,无法检索访问日志") return } @@ -388,9 +386,8 @@ type logsAnalyticsResponse struct { // @Failure 403 {object} response.Any "无管理员权限" // @Router /api/v1/admin/logs/analytics [get] func GetLogsAnalytics(c *gin.Context) { - // 1. 检查 ClickHouse 是否启用 - if !config.Config.ClickHouse.Enabled || db.ChConn == nil { - response.AbortWithError(c, http.StatusBadRequest, "ClickHouse 存储服务未启用,无法获取分析数据") + if db.ChConn == nil { + response.AbortWithError(c, http.StatusInternalServerError, "ClickHouse 未初始化,无法获取分析数据") return } diff --git a/openflare-server/internal/apps/openflare/agent/observability.go b/openflare-server/internal/apps/openflare/agent/observability.go index fd9f3745..d8acf8fc 100644 --- a/openflare-server/internal/apps/openflare/agent/observability.go +++ b/openflare-server/internal/apps/openflare/agent/observability.go @@ -124,6 +124,12 @@ func PersistHeartbeatObservability(ctx context.Context, nodeID string, payload N return } + accessLogRecords, err := buildNodeAccessLogRecords(nodeID, payload.AccessLogs, payload.BufferedObservability, reportedAt) + if err != nil { + zap.L().Error("build heartbeat access logs failed", zap.String("node_id", nodeID), zap.Error(err)) + return + } + if err := conn.Transaction(func(tx *gorm.DB) error { if err := persistNodeSystemProfile(tx, nodeID, payload.Profile, reportedAt); err != nil { return err @@ -140,9 +146,6 @@ func PersistHeartbeatObservability(ctx context.Context, nodeID string, payload N if err := persistNodeTrafficReport(tx, nodeID, payload.TrafficReport, reportedAt); err != nil { return err } - if err := persistNodeAccessLogs(tx, nodeID, payload.AccessLogs, reportedAt); err != nil { - return err - } if payload.HealthEvents != nil { if err := reconcileNodeHealthEvents(tx, nodeID, payload.HealthEvents, reportedAt); err != nil { return err @@ -151,6 +154,11 @@ func PersistHeartbeatObservability(ctx context.Context, nodeID string, payload N return nil }); err != nil { zap.L().Error("persist heartbeat observability failed", zap.String("node_id", nodeID), zap.Error(err)) + return + } + + if err := persistNodeAccessLogs(ctx, nodeID, accessLogRecords, reportedAt); err != nil { + zap.L().Error("persist heartbeat access logs failed", zap.String("node_id", nodeID), zap.Error(err)) } } @@ -165,9 +173,7 @@ func persistBufferedObservability(tx *gorm.DB, nodeID string, records []Buffered if err := persistNodeTrafficReport(tx, nodeID, record.TrafficReport, reportedAt); err != nil { return err } - if err := persistNodeAccessLogs(tx, nodeID, record.AccessLogs, reportedAt); err != nil { - return err - } + } return nil } @@ -278,10 +284,15 @@ func persistNodeTrafficReport(tx *gorm.DB, nodeID string, report *NodeTrafficRep return tx.Create(record).Error } -func persistNodeAccessLogs(tx *gorm.DB, nodeID string, logs []NodeAccessLog, reportedAt time.Time) error { - if len(logs) == 0 { - return nil +func buildNodeAccessLogRecords(nodeID string, direct []NodeAccessLog, buffered []BufferedObservabilityRecord, reportedAt time.Time) ([]*model.OpenFlareAccessLog, error) { + total := len(direct) + for _, record := range buffered { + total += len(record.AccessLogs) } + if total == 0 { + return nil, nil + } + resolver, err := newAccessLogRegionResolver() if err != nil { slog.Warn("initialize access log geo resolver failed", "node_id", nodeID, "error", err) @@ -289,31 +300,40 @@ func persistNodeAccessLogs(tx *gorm.DB, nodeID string, logs []NodeAccessLog, rep if resolver != nil { defer resolver.Close() } - for _, item := range logs { - record := &model.OpenFlareAccessLog{ - NodeID: nodeID, - LoggedAt: timeFromUnix(item.LoggedAtUnix, reportedAt), - RemoteAddr: strings.TrimSpace(item.RemoteAddr), - Region: "", - Host: strings.TrimSpace(item.Host), - Path: truncateForDatabase(strings.TrimSpace(item.Path), accessLogPathMaxLength), - StatusCode: item.StatusCode, - } - if resolver != nil { - record.Region = resolver.Resolve(record.RemoteAddr) - } - exists, err := accessLogExists(tx, record) - if err != nil { - return err - } - if exists { - continue - } - if err := tx.Create(record).Error; err != nil { - return err + + records := make([]*model.OpenFlareAccessLog, 0, total) + appendLogs := func(logs []NodeAccessLog) { + for _, item := range logs { + record := &model.OpenFlareAccessLog{ + NodeID: nodeID, + LoggedAt: timeFromUnix(item.LoggedAtUnix, reportedAt), + RemoteAddr: strings.TrimSpace(item.RemoteAddr), + Region: "", + Host: strings.TrimSpace(item.Host), + Path: truncateForDatabase(strings.TrimSpace(item.Path), accessLogPathMaxLength), + StatusCode: item.StatusCode, + } + if resolver != nil { + record.Region = resolver.Resolve(record.RemoteAddr) + } + records = append(records, record) } } - _, err = deleteAccessLogsByNodeBefore(tx, nodeID, reportedAt.Add(-nodeAccessLogRetentionWindow)) + appendLogs(direct) + for _, record := range buffered { + appendLogs(record.AccessLogs) + } + return records, nil +} + +func persistNodeAccessLogs(ctx context.Context, nodeID string, records []*model.OpenFlareAccessLog, reportedAt time.Time) error { + if len(records) == 0 { + return nil + } + if err := model.InsertOpenFlareAccessLogsBatch(ctx, records); err != nil { + return err + } + _, err := model.DeleteOpenFlareAccessLogsByNodeBefore(ctx, nodeID, reportedAt.Add(-nodeAccessLogRetentionWindow)) return err } @@ -434,30 +454,6 @@ func requestReportExists(tx *gorm.DB, nodeID string, windowStartedAt, windowEnde return count > 0, nil } -func deleteAccessLogsByNodeBefore(tx *gorm.DB, nodeID string, before time.Time) (int64, error) { - result := tx.Where("node_id = ? AND logged_at < ?", nodeID, before).Delete(&model.OpenFlareAccessLog{}) - return result.RowsAffected, result.Error -} - -func accessLogExists(tx *gorm.DB, record *model.OpenFlareAccessLog) (bool, error) { - var count int64 - if err := tx.Model(&model.OpenFlareAccessLog{}). - Where( - "node_id = ? AND logged_at = ? AND remote_addr = ? AND host = ? AND path = ? AND status_code = ?", - record.NodeID, - record.LoggedAt, - record.RemoteAddr, - record.Host, - record.Path, - record.StatusCode, - ). - Limit(1). - Count(&count).Error; err != nil { - return false, err - } - return count > 0, nil -} - func normalizeHealthEventType(eventType string) string { eventType = strings.TrimSpace(strings.ToLower(eventType)) eventType = strings.ReplaceAll(eventType, " ", "_") diff --git a/openflare-server/internal/apps/openflare/async_tasks_test.go b/openflare-server/internal/apps/openflare/async_tasks_test.go index c9d752c6..acbc2fb2 100644 --- a/openflare-server/internal/apps/openflare/async_tasks_test.go +++ b/openflare-server/internal/apps/openflare/async_tasks_test.go @@ -34,21 +34,22 @@ func TestDatabaseAutoCleanupHandlerDeletesRowsWhenEnabled(t *testing.T) { DisableForeignKeyConstraintWhenMigrating: true, }) require.NoError(t, err) - require.NoError(t, sqliteDB.AutoMigrate(&model.OpenFlareAccessLog{})) db.SetDB(sqliteDB) + resetAccessLogStore := model.SetAccessLogStoreForTest(model.NewMemoryAccessLogStore()) t.Cleanup(func() { + resetAccessLogStore() db.SetDB(nil) }) now := time.Now().UTC() - require.NoError(t, db.DB(context.Background()).Create(&model.OpenFlareAccessLog{ + require.NoError(t, model.InsertOpenFlareAccessLogsBatch(context.Background(), []*model.OpenFlareAccessLog{{ NodeID: "node-a", LoggedAt: now.Add(-48 * time.Hour), RemoteAddr: "203.0.113.10", Host: "example.com", Path: "/access", StatusCode: 200, - }).Error) + }})) previousEnabled := model.DatabaseAutoCleanupEnabled previousRetentionDays := model.DatabaseAutoCleanupRetentionDays diff --git a/openflare-server/internal/apps/openflare/dashboard/logics_test.go b/openflare-server/internal/apps/openflare/dashboard/logics_test.go index 9c8a1ed1..8eed2448 100644 --- a/openflare-server/internal/apps/openflare/dashboard/logics_test.go +++ b/openflare-server/internal/apps/openflare/dashboard/logics_test.go @@ -24,7 +24,9 @@ func setupDashboardTestDB(t *testing.T) func() { require.NoError(t, sqliteDB.AutoMigrate(&model.OpenFlareNode{})) db.SetDB(sqliteDB) + resetAccessLogStore := model.SetAccessLogStoreForTest(model.NewMemoryAccessLogStore()) return func() { + resetAccessLogStore() db.SetDB(nil) } } diff --git a/openflare-server/internal/apps/openflare/node/logics_test.go b/openflare-server/internal/apps/openflare/node/logics_test.go index db647b7f..596bbf95 100644 --- a/openflare-server/internal/apps/openflare/node/logics_test.go +++ b/openflare-server/internal/apps/openflare/node/logics_test.go @@ -35,8 +35,10 @@ func setupNodeTestDB(t *testing.T) func() { db.SetDB(sqliteDB) option.ResetInitializationForTest() + resetAccessLogStore := model.SetAccessLogStoreForTest(model.NewMemoryAccessLogStore()) return func() { + resetAccessLogStore() db.SetDB(nil) option.ResetInitializationForTest() } diff --git a/openflare-server/internal/apps/openflare/option/logics_test.go b/openflare-server/internal/apps/openflare/option/logics_test.go index 36e87c01..ce384f7d 100644 --- a/openflare-server/internal/apps/openflare/option/logics_test.go +++ b/openflare-server/internal/apps/openflare/option/logics_test.go @@ -101,26 +101,28 @@ func TestCleanupDatabaseObservabilityDeletesRows(t *testing.T) { defer cleanup() ctx := context.Background() - sqliteDB := db.DB(ctx) - require.NoError(t, sqliteDB.AutoMigrate(&model.OpenFlareAccessLog{})) + resetAccessLogStore := model.SetAccessLogStoreForTest(model.NewMemoryAccessLogStore()) + defer resetAccessLogStore() now := time.Now().UTC() - require.NoError(t, sqliteDB.Create(&model.OpenFlareAccessLog{ - NodeID: "node-a", - LoggedAt: now.Add(-10 * 24 * time.Hour), - RemoteAddr: "203.0.113.1", - Host: "example.com", - Path: "/old", - StatusCode: 200, - }).Error) - require.NoError(t, sqliteDB.Create(&model.OpenFlareAccessLog{ - NodeID: "node-a", - LoggedAt: now.Add(-2 * time.Hour), - RemoteAddr: "203.0.113.2", - Host: "example.com", - Path: "/recent", - StatusCode: 200, - }).Error) + require.NoError(t, model.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 := 7 result, err := cleanupDatabaseObservability(ctx, databaseCleanupInput{ diff --git a/openflare-server/internal/apps/openflare/tasks/database_cleanup_test.go b/openflare-server/internal/apps/openflare/tasks/database_cleanup_test.go index fc4d008d..3c899849 100644 --- a/openflare-server/internal/apps/openflare/tasks/database_cleanup_test.go +++ b/openflare-server/internal/apps/openflare/tasks/database_cleanup_test.go @@ -24,12 +24,13 @@ func setupDatabaseCleanupTestDB(t *testing.T) context.Context { }) require.NoError(t, err) require.NoError(t, sqliteDB.AutoMigrate( - &model.OpenFlareAccessLog{}, &model.OpenFlareMetricSnapshot{}, &model.OpenFlareRequestReport{}, )) db.SetDB(sqliteDB) + resetAccessLogStore := model.SetAccessLogStoreForTest(model.NewMemoryAccessLogStore()) t.Cleanup(func() { + resetAccessLogStore() db.SetDB(nil) }) return context.Background() @@ -69,22 +70,24 @@ func TestCleanupDatabaseObservabilityDeletesAllRowsWhenRetentionMissing(t *testi ctx := setupDatabaseCleanupTestDB(t) now := time.Now().UTC() - require.NoError(t, db.DB(ctx).Create(&model.OpenFlareAccessLog{ - NodeID: "node-a", - LoggedAt: now.Add(-3 * time.Hour), - RemoteAddr: "203.0.113.1", - Host: "example.com", - Path: "/one", - StatusCode: 200, - }).Error) - require.NoError(t, db.DB(ctx).Create(&model.OpenFlareAccessLog{ - NodeID: "node-a", - LoggedAt: now.Add(-2 * time.Hour), - RemoteAddr: "203.0.113.2", - Host: "example.com", - Path: "/two", - StatusCode: 502, - }).Error) + require.NoError(t, model.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, @@ -102,14 +105,14 @@ func TestRunDatabaseAutoCleanupOnceDeletesAllObservabilityTargets(t *testing.T) ctx := setupDatabaseCleanupTestDB(t) now := time.Now().UTC() - require.NoError(t, db.DB(ctx).Create(&model.OpenFlareAccessLog{ + require.NoError(t, model.InsertOpenFlareAccessLogsBatch(ctx, []*model.OpenFlareAccessLog{{ NodeID: "node-a", LoggedAt: now.Add(-48 * time.Hour), RemoteAddr: "203.0.113.10", Host: "example.com", Path: "/access", StatusCode: 200, - }).Error) + }})) require.NoError(t, db.DB(ctx).Create(&model.OpenFlareMetricSnapshot{ NodeID: "node-a", CapturedAt: now.Add(-48 * time.Hour), diff --git a/openflare-server/internal/apps/openflare/waf/ip_group_sync_test.go b/openflare-server/internal/apps/openflare/waf/ip_group_sync_test.go index c119ad95..e2e9f0ff 100644 --- a/openflare-server/internal/apps/openflare/waf/ip_group_sync_test.go +++ b/openflare-server/internal/apps/openflare/waf/ip_group_sync_test.go @@ -29,11 +29,12 @@ func setupIPGroupSyncTestDB(t *testing.T) func() { require.NoError(t, sqliteDB.AutoMigrate( &model.OpenFlareWAFRuleGroup{}, &model.OpenFlareWAFIPGroup{}, - &model.OpenFlareAccessLog{}, )) db.SetDB(sqliteDB) + resetAccessLogStore := model.SetAccessLogStoreForTest(model.NewMemoryAccessLogStore()) return func() { + resetAccessLogStore() db.SetDB(nil) } } @@ -191,18 +192,20 @@ func TestListDueOpenFlareWAFIPGroups(t *testing.T) { func seedWAFAccessLogs(t *testing.T, ctx context.Context, loggedAt time.Time, remoteAddr string, host string, total int, notFound int) { t.Helper() + records := make([]*model.OpenFlareAccessLog, 0, total) for i := 0; i < total; i++ { statusCode := http.StatusOK if i < notFound { statusCode = http.StatusNotFound } - require.NoError(t, db.DB(ctx).Create(&model.OpenFlareAccessLog{ + records = append(records, &model.OpenFlareAccessLog{ NodeID: "node-waf-auto", LoggedAt: loggedAt.Add(-time.Duration(i%30) * time.Second), RemoteAddr: remoteAddr, Host: host, Path: "/probe", StatusCode: statusCode, - }).Error) + }) } + require.NoError(t, model.InsertOpenFlareAccessLogsBatch(ctx, records)) } diff --git a/openflare-server/internal/config/config.go b/openflare-server/internal/config/config.go index f1ee67db..268c7997 100644 --- a/openflare-server/internal/config/config.go +++ b/openflare-server/internal/config/config.go @@ -93,12 +93,11 @@ func init() { applyEnvOverrides(&c) applyDefaults(&c) - // Disable standard DB/Redis initializations during tests to prevent connection attempts. + // Disable standard DB/Redis/ClickHouse initializations during tests to prevent connection attempts. if isTest() { c.Database.Enabled = false c.Database.SQLitePath = ":memory:" c.Redis.Enabled = false - c.ClickHouse.Enabled = false } // 设置全局配置 @@ -115,6 +114,41 @@ func applyDefaults(c *configModel) { if c.Otel.TracerName == "" { c.Otel.TracerName = "github.com/Rain-kl/OpenFlare" } + applyClickHouseDefaults(c) +} + +func applyClickHouseDefaults(c *configModel) { + if isTest() { + c.ClickHouse.Enabled = false + return + } + if !c.ClickHouse.Enabled { + c.ClickHouse.Enabled = true + } + if c.ClickHouse.Database == "" { + c.ClickHouse.Database = "openflare" + } + if len(c.ClickHouse.Hosts) == 0 { + c.ClickHouse.Hosts = []string{"127.0.0.1:9000"} + } + if c.ClickHouse.Username == "" { + c.ClickHouse.Username = "default" + } + if c.ClickHouse.MaxIdleConn <= 0 { + c.ClickHouse.MaxIdleConn = 10 + } + if c.ClickHouse.MaxOpenConn <= 0 { + c.ClickHouse.MaxOpenConn = 100 + } + if c.ClickHouse.ConnMaxLifetime <= 0 { + c.ClickHouse.ConnMaxLifetime = 3600 + } + if c.ClickHouse.DialTimeout <= 0 { + c.ClickHouse.DialTimeout = 5 + } + if c.ClickHouse.BlockBufferSize == 0 { + c.ClickHouse.BlockBufferSize = 10 + } } // ─── 环境变量覆盖层 ──────────────────────────────────────────────────────────── diff --git a/openflare-server/internal/db/clickhouse.go b/openflare-server/internal/db/clickhouse.go index b8783efb..c6ee961a 100644 --- a/openflare-server/internal/db/clickhouse.go +++ b/openflare-server/internal/db/clickhouse.go @@ -31,6 +31,10 @@ func init() { } cfg := config.Config.ClickHouse + if cfg.Database == "" { + log.Fatalf("[ClickHouse] database name is required (expected: openflare)\n") + } + var err error // 配置 ClickHouse 连接 @@ -64,5 +68,6 @@ func init() { log.Fatalf("[ClickHouse] ping failed: %v\n", err) } + ensureClickHouseSchemaOnStartup() log.Println("[ClickHouse] connection established successfully") } diff --git a/openflare-server/internal/db/clickhouse_schema.go b/openflare-server/internal/db/clickhouse_schema.go new file mode 100644 index 00000000..90ee1af9 --- /dev/null +++ b/openflare-server/internal/db/clickhouse_schema.go @@ -0,0 +1,51 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package db + +import ( + "context" + "fmt" + "log" +) + +const openFlareNodeAccessLogsDDL = ` +CREATE TABLE IF NOT EXISTS of_node_access_logs +( + id UInt64, + node_id String, + logged_at DateTime64(3, 'UTC'), + remote_addr String, + region String, + host String, + path String, + status_code Int32, + created_at DateTime64(3, 'UTC') +) +ENGINE = MergeTree() +PARTITION BY toYYYYMM(logged_at) +ORDER BY (node_id, logged_at, remote_addr, host, path, status_code) +SETTINGS index_granularity = 8192` + +// EnsureClickHouseSchema creates the openflare database and required tables. +func EnsureClickHouseSchema(ctx context.Context) error { + if ChConn == nil { + return fmt.Errorf("clickhouse connection is not initialized") + } + + if err := ChConn.Exec(ctx, "CREATE DATABASE IF NOT EXISTS openflare"); err != nil { + return fmt.Errorf("create database openflare: %w", err) + } + if err := ChConn.Exec(ctx, openFlareNodeAccessLogsDDL); err != nil { + return fmt.Errorf("create table of_node_access_logs: %w", err) + } + return nil +} + +func ensureClickHouseSchemaOnStartup() { + ctx := context.Background() + if err := EnsureClickHouseSchema(ctx); err != nil { + log.Fatalf("[ClickHouse] ensure schema failed: %v\n", err) + } + log.Println("[ClickHouse] schema ready (database: openflare)") +} \ No newline at end of file diff --git a/openflare-server/internal/db/migrator/goose/postgres/202606200004_drop_of_node_access_logs.sql b/openflare-server/internal/db/migrator/goose/postgres/202606200004_drop_of_node_access_logs.sql new file mode 100644 index 00000000..aafeeebe --- /dev/null +++ b/openflare-server/internal/db/migrator/goose/postgres/202606200004_drop_of_node_access_logs.sql @@ -0,0 +1,28 @@ +-- +goose Up +DROP INDEX IF EXISTS idx_of_node_access_logs_node_id_logged_at; +DROP INDEX IF EXISTS idx_of_node_access_logs_status_code; +DROP INDEX IF EXISTS idx_of_node_access_logs_host; +DROP INDEX IF EXISTS idx_of_node_access_logs_remote_addr; +DROP INDEX IF EXISTS idx_of_node_access_logs_logged_at; +DROP INDEX IF EXISTS idx_of_node_access_logs_node_id; +DROP TABLE IF EXISTS of_node_access_logs; + +-- +goose Down +CREATE TABLE of_node_access_logs ( + id BIGSERIAL PRIMARY KEY, + node_id VARCHAR(64) NOT NULL, + 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 '', + status_code INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX idx_of_node_access_logs_node_id ON of_node_access_logs (node_id); +CREATE INDEX idx_of_node_access_logs_logged_at ON of_node_access_logs (logged_at); +CREATE INDEX idx_of_node_access_logs_remote_addr ON of_node_access_logs (remote_addr); +CREATE INDEX idx_of_node_access_logs_host ON of_node_access_logs (host); +CREATE INDEX idx_of_node_access_logs_status_code ON of_node_access_logs (status_code); +CREATE INDEX idx_of_node_access_logs_node_id_logged_at ON of_node_access_logs (node_id, logged_at); \ No newline at end of file diff --git a/openflare-server/internal/db/migrator/goose/sqlite/202606200004_drop_of_node_access_logs.sql b/openflare-server/internal/db/migrator/goose/sqlite/202606200004_drop_of_node_access_logs.sql new file mode 100644 index 00000000..06fe18f2 --- /dev/null +++ b/openflare-server/internal/db/migrator/goose/sqlite/202606200004_drop_of_node_access_logs.sql @@ -0,0 +1,28 @@ +-- +goose Up +DROP INDEX IF EXISTS idx_of_node_access_logs_node_id_logged_at; +DROP INDEX IF EXISTS idx_of_node_access_logs_status_code; +DROP INDEX IF EXISTS idx_of_node_access_logs_host; +DROP INDEX IF EXISTS idx_of_node_access_logs_remote_addr; +DROP INDEX IF EXISTS idx_of_node_access_logs_logged_at; +DROP INDEX IF EXISTS idx_of_node_access_logs_node_id; +DROP TABLE IF EXISTS of_node_access_logs; + +-- +goose Down +CREATE TABLE of_node_access_logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + node_id TEXT NOT NULL, + logged_at DATETIME NOT NULL, + remote_addr TEXT NOT NULL DEFAULT '', + region TEXT NOT NULL DEFAULT '', + host TEXT NOT NULL DEFAULT '', + path TEXT NOT NULL DEFAULT '', + status_code INTEGER NOT NULL DEFAULT 0, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX idx_of_node_access_logs_node_id ON of_node_access_logs (node_id); +CREATE INDEX idx_of_node_access_logs_logged_at ON of_node_access_logs (logged_at); +CREATE INDEX idx_of_node_access_logs_remote_addr ON of_node_access_logs (remote_addr); +CREATE INDEX idx_of_node_access_logs_host ON of_node_access_logs (host); +CREATE INDEX idx_of_node_access_logs_status_code ON of_node_access_logs (status_code); +CREATE INDEX idx_of_node_access_logs_node_id_logged_at ON of_node_access_logs (node_id, logged_at); \ No newline at end of file diff --git a/openflare-server/internal/model/errs.go b/openflare-server/internal/model/errs.go index 6c2a64ee..e8723ad6 100644 --- a/openflare-server/internal/model/errs.go +++ b/openflare-server/internal/model/errs.go @@ -6,6 +6,7 @@ package model const ( errRegistrationDisabled = "注册已关闭" errDatabaseNotInitialized = "database not initialized" + errClickHouseNotInitialized = "clickhouse not initialized" errUsernameExists = "用户名已存在" errEmailAlreadyBound = "该邮箱已被其他账号绑定" errConfigIntParseFailed = "配置 %s 的值 '%s' 无法转换为整数: %w" diff --git a/openflare-server/internal/model/openflare_access_log.go b/openflare-server/internal/model/openflare_access_log.go index cc719c08..cea5bca5 100644 --- a/openflare-server/internal/model/openflare_access_log.go +++ b/openflare-server/internal/model/openflare_access_log.go @@ -5,18 +5,11 @@ package model import ( "context" - "errors" - "fmt" "sort" "strings" "time" - - "github.com/Rain-kl/Wavelet/internal/db" - "gorm.io/gorm" ) -const openFlareAccessLogTable = "of_node_access_logs" - type openFlareAccessLogBucketAggregateRow struct { BucketEpoch int64 `gorm:"column:bucket_epoch"` RequestCount int64 `gorm:"column:request_count"` @@ -56,81 +49,24 @@ func ListOpenFlareAccessLogsForWAFIPGroup(ctx context.Context, query OpenFlareAc return ListOpenFlareAccessLogs(ctx, query) } +// InsertOpenFlareAccessLogsBatch inserts access log rows into ClickHouse. +func InsertOpenFlareAccessLogsBatch(ctx context.Context, records []*OpenFlareAccessLog) error { + return currentAccessLogStore().InsertBatch(ctx, records) +} + // ListOpenFlareAccessLogs lists access logs matching the query. func ListOpenFlareAccessLogs(ctx context.Context, query OpenFlareAccessLogQuery) ([]*OpenFlareAccessLog, error) { - conn := db.DB(ctx) - if conn == nil { - return nil, errors.New(errDatabaseNotInitialized) - } - tx := applyOpenFlareAccessLogFilters(conn.Model(&OpenFlareAccessLog{}), query) - tx = tx.Order(openFlareAccessLogOrderClause(query.SortBy, query.SortOrder)) - if query.PageSize > 0 { - if query.Page < 0 { - query.Page = 0 - } - tx = tx.Offset(query.Page * query.PageSize).Limit(query.PageSize) - } - var rows []*OpenFlareAccessLog - if err := tx.Find(&rows).Error; err != nil { - if isMissingTableError(err) { - return []*OpenFlareAccessLog{}, nil - } - return nil, err - } - return rows, nil + return currentAccessLogStore().List(ctx, query) } // CountOpenFlareAccessLogs counts access logs and distinct IPs matching the query. func CountOpenFlareAccessLogs(ctx context.Context, query OpenFlareAccessLogQuery) (int64, int64, error) { - conn := db.DB(ctx) - if conn == nil { - return 0, 0, errors.New(errDatabaseNotInitialized) - } - totalRecords, err := countOpenFlareAccessLogRecords(conn, query) - if err != nil { - if isMissingTableError(err) { - return 0, 0, nil - } - return 0, 0, err - } - totalIPs, err := countDistinctOpenFlareAccessLogIPs(conn, query) - if err != nil { - if isMissingTableError(err) { - return 0, 0, nil - } - return 0, 0, err - } - return totalRecords, totalIPs, nil + return currentAccessLogStore().Count(ctx, query) } // ListOpenFlareAccessLogRegionCounts returns region counts for access logs. func ListOpenFlareAccessLogRegionCounts(ctx context.Context, nodeID string, since time.Time, limit int) ([]*OpenFlareAccessLogRegionCount, error) { - conn := db.DB(ctx) - if conn == nil { - return nil, errors.New(errDatabaseNotInitialized) - } - filter := OpenFlareAccessLogQuery{ - NodeID: nodeID, - Since: since, - } - clause, args := buildOpenFlareAccessLogFilterClause(filter) - sql := fmt.Sprintf(` -SELECT TRIM(region) AS region, COUNT(*) AS count -FROM %s -WHERE %s AND TRIM(region) <> '' -GROUP BY TRIM(region) -ORDER BY count DESC, region ASC`, openFlareAccessLogTable, clause) - if limit > 0 { - sql += fmt.Sprintf(" LIMIT %d", limit) - } - var rows []*OpenFlareAccessLogRegionCount - if err := conn.Raw(sql, args...).Scan(&rows).Error; err != nil { - if isMissingTableError(err) { - return []*OpenFlareAccessLogRegionCount{}, nil - } - return nil, err - } - return rows, nil + return currentAccessLogStore().RegionCounts(ctx, nodeID, since, limit) } // ListOpenFlareAccessLogBuckets lists folded access log buckets. @@ -201,10 +137,6 @@ func CountOpenFlareAccessLogIPSummaries(ctx context.Context, query OpenFlareAcce // ListOpenFlareAccessLogIPTrend lists IP trend points. func ListOpenFlareAccessLogIPTrend(ctx context.Context, query OpenFlareAccessLogIPTrendQuery) ([]*OpenFlareAccessLogIPTrendRow, error) { - conn := db.DB(ctx) - if conn == nil { - return nil, errors.New(errDatabaseNotInitialized) - } remoteAddr := strings.TrimSpace(query.RemoteAddr) if remoteAddr == "" { return []*OpenFlareAccessLogIPTrendRow{}, nil @@ -215,76 +147,45 @@ func ListOpenFlareAccessLogIPTrend(ctx context.Context, query OpenFlareAccessLog Host: query.Host, Since: query.Since, } - clause, args := buildOpenFlareAccessLogFilterClause(filter) bucketSeconds := int64(query.BucketMinutes * 60) if bucketSeconds <= 0 { bucketSeconds = 1800 } - bucketExpr := openFlareAccessLogBucketEpochExpr(openFlareAccessLogDialect(conn), bucketSeconds) - queryClause := combineOpenFlareAccessLogSQLClauses(clause, "TRIM(remote_addr) = ?") - queryArgs := append(append([]any{}, args...), remoteAddr) - sql := fmt.Sprintf(` -SELECT - %s AS bucket_epoch, - COUNT(*) AS request_count -FROM %s -WHERE %s -GROUP BY bucket_epoch -ORDER BY bucket_epoch ASC`, bucketExpr, openFlareAccessLogTable, queryClause) - var rows []*OpenFlareAccessLogIPTrendRow - if err := conn.Raw(sql, queryArgs...).Scan(&rows).Error; err != nil { - if isMissingTableError(err) { - return []*OpenFlareAccessLogIPTrendRow{}, nil - } + rows, err := currentAccessLogStore().IPTrend(ctx, filter, bucketSeconds) + if err != nil { return nil, err } - return rows, nil + result := make([]*OpenFlareAccessLogIPTrendRow, len(rows)) + for index, row := range rows { + result[index] = &OpenFlareAccessLogIPTrendRow{ + BucketEpoch: row.BucketEpoch, + RequestCount: row.RequestCount, + } + } + return result, nil } // DeleteAllOpenFlareAccessLogs deletes all access logs. func DeleteAllOpenFlareAccessLogs(ctx context.Context) (int64, error) { - conn := db.DB(ctx) - if conn == nil { - return 0, errors.New(errDatabaseNotInitialized) - } - result := conn.Where("1 = 1").Delete(&OpenFlareAccessLog{}) - if result.Error != nil { - if isMissingTableError(result.Error) { - return 0, nil - } - return 0, result.Error - } - return result.RowsAffected, nil + return currentAccessLogStore().DeleteAll(ctx) } // DeleteOpenFlareAccessLogsBefore deletes access logs older than cutoff. func DeleteOpenFlareAccessLogsBefore(ctx context.Context, cutoff time.Time) (int64, error) { - conn := db.DB(ctx) - if conn == nil { - return 0, errors.New(errDatabaseNotInitialized) - } - result := conn.Where("logged_at < ?", cutoff).Delete(&OpenFlareAccessLog{}) - if result.Error != nil { - if isMissingTableError(result.Error) { - return 0, nil - } - return 0, result.Error - } - return result.RowsAffected, nil + return currentAccessLogStore().DeleteBefore(ctx, cutoff) +} + +// DeleteOpenFlareAccessLogsByNodeBefore deletes access logs for a node older than cutoff. +func DeleteOpenFlareAccessLogsByNodeBefore(ctx context.Context, nodeID string, cutoff time.Time) (int64, error) { + return currentAccessLogStore().DeleteByNodeBefore(ctx, nodeID, cutoff) } func buildOpenFlareAccessLogBucketRows(ctx context.Context, query OpenFlareAccessLogBucketQuery) ([]*OpenFlareAccessLogBucketRow, error) { - conn := db.DB(ctx) - if conn == nil { - return nil, errors.New(errDatabaseNotInitialized) - } filter := openFlareAccessLogQueryFromBucket(query) - clause, args := buildOpenFlareAccessLogFilterClause(filter) bucketSeconds := int64(query.FoldMinutes * 60) if bucketSeconds <= 0 { bucketSeconds = 180 } - bucketExpr := openFlareAccessLogBucketEpochExpr(openFlareAccessLogDialect(conn), bucketSeconds) type bucketAccumulator struct { requestCount int64 @@ -296,21 +197,8 @@ func buildOpenFlareAccessLogBucketRows(ctx context.Context, query OpenFlareAcces } accumulators := make(map[int64]*bucketAccumulator) - var partials []openFlareAccessLogBucketAggregateRow - sql := fmt.Sprintf(` -SELECT - %s AS bucket_epoch, - COUNT(*) AS request_count, - SUM(CASE WHEN status_code < 400 THEN 1 ELSE 0 END) AS success_count, - SUM(CASE WHEN status_code >= 400 AND status_code < 500 THEN 1 ELSE 0 END) AS client_error_count, - SUM(CASE WHEN status_code >= 500 THEN 1 ELSE 0 END) AS server_error_count -FROM %s -WHERE %s -GROUP BY bucket_epoch`, bucketExpr, openFlareAccessLogTable, clause) - if err := conn.Raw(sql, args...).Scan(&partials).Error; err != nil { - if isMissingTableError(err) { - return []*OpenFlareAccessLogBucketRow{}, nil - } + partials, err := currentAccessLogStore().BucketAggregates(ctx, filter, bucketSeconds) + if err != nil { return nil, err } for _, partial := range partials { @@ -329,7 +217,7 @@ GROUP BY bucket_epoch`, bucketExpr, openFlareAccessLogTable, clause) } for _, column := range []string{"remote_addr", "host"} { - dimensions, err := queryOpenFlareAccessLogBucketDimensionRows(conn, clause, args, column, bucketExpr) + dimensions, err := currentAccessLogStore().BucketDimensions(ctx, filter, column, bucketSeconds) if err != nil { return nil, err } @@ -371,24 +259,6 @@ GROUP BY bucket_epoch`, bucketExpr, openFlareAccessLogTable, clause) return rows, nil } -func queryOpenFlareAccessLogBucketDimensionRows(conn *gorm.DB, clause string, args []any, column string, bucketExpr string) ([]openFlareAccessLogBucketDimensionRow, error) { - var rows []openFlareAccessLogBucketDimensionRow - sql := fmt.Sprintf(` -SELECT - %s AS bucket_epoch, - TRIM(%s) AS value -FROM %s -WHERE %s AND TRIM(%s) <> '' -GROUP BY bucket_epoch, TRIM(%s)`, bucketExpr, column, openFlareAccessLogTable, clause, column, column) - if err := conn.Raw(sql, args...).Scan(&rows).Error; err != nil { - if isMissingTableError(err) { - return []openFlareAccessLogBucketDimensionRow{}, nil - } - return nil, err - } - return rows, nil -} - func buildOpenFlareAccessLogBucketIPRows(ctx context.Context, query OpenFlareAccessLogBucketIPQuery) ([]*OpenFlareAccessLogBucketIPRow, error) { if query.BucketStartedAt.IsZero() { return []*OpenFlareAccessLogBucketIPRow{}, nil @@ -415,39 +285,14 @@ func buildOpenFlareAccessLogBucketIPRows(ctx context.Context, query OpenFlareAcc } func buildOpenFlareAccessLogIPSummaryRows(ctx context.Context, query OpenFlareAccessLogIPSummaryQuery, recentSince time.Time) ([]*OpenFlareAccessLogIPSummaryRow, error) { - conn := db.DB(ctx) - if conn == nil { - return nil, errors.New(errDatabaseNotInitialized) - } filter := OpenFlareAccessLogQuery{ NodeID: query.NodeID, RemoteAddr: query.RemoteAddr, Host: query.Host, Since: query.Since, } - clause, args := buildOpenFlareAccessLogFilterClause(filter) - lastSeenExpr := openFlareAccessLogEpochExpr(openFlareAccessLogDialect(conn)) - recentClause := "0" - queryArgs := make([]any, 0, len(args)+1) - if !recentSince.IsZero() { - recentClause = "CASE WHEN logged_at >= ? THEN 1 ELSE 0 END" - queryArgs = append(queryArgs, recentSince) - } - queryArgs = append(queryArgs, args...) - sql := fmt.Sprintf(` -SELECT - TRIM(remote_addr) AS remote_addr, - COUNT(*) AS total_requests, - SUM(%s) AS recent_requests, - MAX(%s) AS last_seen_epoch -FROM %s -WHERE %s AND TRIM(remote_addr) <> '' -GROUP BY TRIM(remote_addr)`, recentClause, lastSeenExpr, openFlareAccessLogTable, clause) - var partials []openFlareAccessLogIPSummaryRow - if err := conn.Raw(sql, queryArgs...).Scan(&partials).Error; err != nil { - if isMissingTableError(err) { - return []*OpenFlareAccessLogIPSummaryRow{}, nil - } + partials, err := currentAccessLogStore().IPSummaries(ctx, filter, recentSince) + if err != nil { return nil, err } rows := make([]*OpenFlareAccessLogIPSummaryRow, 0, len(partials)) @@ -468,38 +313,8 @@ GROUP BY TRIM(remote_addr)`, recentClause, lastSeenExpr, openFlareAccessLogTable } func queryOpenFlareAccessLogIPAggregateRows(ctx context.Context, filter OpenFlareAccessLogQuery, exactRemoteAddr bool) ([]*OpenFlareAccessLogBucketIPRow, error) { - conn := db.DB(ctx) - if conn == nil { - return nil, errors.New(errDatabaseNotInitialized) - } - clause, args := buildOpenFlareAccessLogFilterClause(filter) - lastSeenExpr := openFlareAccessLogEpochExpr(openFlareAccessLogDialect(conn)) - queryClause := clause - queryArgs := append([]any{}, args...) - if exactRemoteAddr { - trimmed := strings.TrimSpace(filter.RemoteAddr) - if trimmed == "" { - return []*OpenFlareAccessLogBucketIPRow{}, nil - } - queryClause = combineOpenFlareAccessLogSQLClauses(queryClause, "TRIM(remote_addr) = ?") - queryArgs = append(queryArgs, trimmed) - } - sql := fmt.Sprintf(` -SELECT - TRIM(remote_addr) AS remote_addr, - COUNT(*) AS request_count, - SUM(CASE WHEN status_code < 400 THEN 1 ELSE 0 END) AS success_count, - SUM(CASE WHEN status_code >= 400 AND status_code < 500 THEN 1 ELSE 0 END) AS client_error_count, - SUM(CASE WHEN status_code >= 500 THEN 1 ELSE 0 END) AS server_error_count, - MAX(%s) AS last_seen_epoch -FROM %s -WHERE %s AND TRIM(remote_addr) <> '' -GROUP BY TRIM(remote_addr)`, lastSeenExpr, openFlareAccessLogTable, queryClause) - var partials []openFlareAccessLogIPAggregateRow - if err := conn.Raw(sql, queryArgs...).Scan(&partials).Error; err != nil { - if isMissingTableError(err) { - return []*OpenFlareAccessLogBucketIPRow{}, nil - } + partials, err := currentAccessLogStore().IPAggregates(ctx, filter, exactRemoteAddr) + if err != nil { return nil, err } rows := make([]*OpenFlareAccessLogBucketIPRow, 0, len(partials)) @@ -551,82 +366,20 @@ func buildOpenFlareAccessLogFilterClause(query OpenFlareAccessLogQuery) (string, } if !query.Since.IsZero() { parts = append(parts, "logged_at >= ?") - args = append(args, query.Since) + args = append(args, query.Since.UTC()) } if !query.Until.IsZero() { parts = append(parts, "logged_at < ?") - args = append(args, query.Until) + args = append(args, query.Until.UTC()) } if len(parts) == 0 { - return "TRUE", nil + return "1", nil } return strings.Join(parts, " AND "), args } -func applyOpenFlareAccessLogFilters(tx *gorm.DB, query OpenFlareAccessLogQuery) *gorm.DB { - clause, args := buildOpenFlareAccessLogFilterClause(query) - if clause == "TRUE" { - return tx - } - return tx.Where(clause, args...) -} - -func countOpenFlareAccessLogRecords(conn *gorm.DB, query OpenFlareAccessLogQuery) (int64, error) { - var count int64 - if err := applyOpenFlareAccessLogFilters(conn.Model(&OpenFlareAccessLog{}), query).Count(&count).Error; err != nil { - return 0, err - } - return count, nil -} - -func countDistinctOpenFlareAccessLogIPs(conn *gorm.DB, query OpenFlareAccessLogQuery) (int64, error) { - clause, args := buildOpenFlareAccessLogFilterClause(query) - sql := fmt.Sprintf(` -SELECT COUNT(*) FROM ( - SELECT TRIM(remote_addr) AS remote_addr - FROM %s - WHERE %s AND remote_addr <> '' - GROUP BY TRIM(remote_addr) -) AS ips`, openFlareAccessLogTable, clause) - var total int64 - if err := conn.Raw(sql, args...).Scan(&total).Error; err != nil { - return 0, err - } - return total, nil -} - -func openFlareAccessLogDialect(conn *gorm.DB) string { - if conn == nil || conn.Dialector == nil { - return "sqlite" - } - switch conn.Dialector.Name() { - case "postgres": - return "postgres" - default: - return "sqlite" - } -} - -func openFlareAccessLogBucketEpochExpr(dialect string, bucketSeconds int64) string { - switch dialect { - case "postgres": - return fmt.Sprintf("FLOOR(EXTRACT(EPOCH FROM logged_at AT TIME ZONE 'UTC') / %d) * %d", bucketSeconds, bucketSeconds) - default: - return fmt.Sprintf("(CAST(strftime('%%s', logged_at) AS INTEGER) / %d) * %d", bucketSeconds, bucketSeconds) - } -} - -func openFlareAccessLogEpochExpr(dialect string) string { - switch dialect { - case "postgres": - return "FLOOR(EXTRACT(EPOCH FROM logged_at AT TIME ZONE 'UTC'))::bigint" - default: - return "CAST((julianday(logged_at) - 2440587.5) * 86400 AS INTEGER)" - } -} - func combineOpenFlareAccessLogSQLClauses(left string, right string) string { - if strings.TrimSpace(left) == "" || left == "TRUE" { + if strings.TrimSpace(left) == "" || left == "TRUE" || left == "1" { return right } return left + " AND " + right @@ -656,7 +409,7 @@ func openFlareAccessLogOrderClause(sortBy string, sortOrder string) string { func sortOpenFlareAccessLogBucketIPRows(items []*OpenFlareAccessLogBucketIPRow, sortBy string, sortOrder string) { desc := openFlareAccessLogNormalizeSortOrder(sortOrder) != "asc" - sort.Slice(items, func(i int, j int) bool { + sort.Slice(items, func(i, j int) bool { left := items[i] right := items[j] if left == nil || right == nil { @@ -686,7 +439,7 @@ func sortOpenFlareAccessLogBucketIPRows(items []*OpenFlareAccessLogBucketIPRow, func sortOpenFlareAccessLogBucketRows(items []*OpenFlareAccessLogBucketRow, sortBy string, sortOrder string) { desc := openFlareAccessLogNormalizeSortOrder(sortOrder) != "asc" - sort.Slice(items, func(i int, j int) bool { + sort.Slice(items, func(i, j int) bool { left := items[i] right := items[j] if left == nil || right == nil { @@ -711,7 +464,7 @@ func sortOpenFlareAccessLogBucketRows(items []*OpenFlareAccessLogBucketRow, sort func sortOpenFlareAccessLogIPSummaryRows(items []*OpenFlareAccessLogIPSummaryRow, sortBy string, sortOrder string) { desc := openFlareAccessLogNormalizeSortOrder(sortOrder) != "asc" - sort.Slice(items, func(i int, j int) bool { + sort.Slice(items, func(i, j int) bool { left := items[i] right := items[j] if left == nil || right == nil { @@ -776,3 +529,4 @@ func openFlareAccessLogCompareInt64(left int64, right int64) int { return 0 } } + diff --git a/openflare-server/internal/model/openflare_access_log_store.go b/openflare-server/internal/model/openflare_access_log_store.go new file mode 100644 index 00000000..cbe5d2cf --- /dev/null +++ b/openflare-server/internal/model/openflare_access_log_store.go @@ -0,0 +1,465 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package model + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "time" + + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" + "github.com/Rain-kl/Wavelet/internal/db" + "github.com/Rain-kl/Wavelet/internal/db/idgen" +) + +const openFlareAccessLogTable = "of_node_access_logs" + +type accessLogStore interface { + InsertBatch(ctx context.Context, records []*OpenFlareAccessLog) error + List(ctx context.Context, query OpenFlareAccessLogQuery) ([]*OpenFlareAccessLog, error) + Count(ctx context.Context, query OpenFlareAccessLogQuery) (int64, int64, error) + RegionCounts(ctx context.Context, nodeID string, since time.Time, limit int) ([]*OpenFlareAccessLogRegionCount, error) + BucketAggregates(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogBucketAggregateRow, error) + BucketDimensions(ctx context.Context, filter OpenFlareAccessLogQuery, column string, bucketSeconds int64) ([]openFlareAccessLogBucketDimensionRow, error) + IPAggregates(ctx context.Context, filter OpenFlareAccessLogQuery, exactRemoteAddr bool) ([]openFlareAccessLogIPAggregateRow, error) + IPSummaries(ctx context.Context, filter OpenFlareAccessLogQuery, recentSince time.Time) ([]openFlareAccessLogIPSummaryRow, error) + IPTrend(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogIPTrendRow, 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) +} + +var ( + accessLogStoreMu sync.RWMutex + accessLogStoreHolder accessLogStore +) + +func currentAccessLogStore() accessLogStore { + accessLogStoreMu.RLock() + defer accessLogStoreMu.RUnlock() + if accessLogStoreHolder != nil { + return accessLogStoreHolder + } + return clickhouseAccessLogStore{} +} + +// SetAccessLogStoreForTest swaps the access log store implementation for unit tests. +func SetAccessLogStoreForTest(store accessLogStore) func() { + accessLogStoreMu.Lock() + previous := accessLogStoreHolder + accessLogStoreHolder = store + accessLogStoreMu.Unlock() + return func() { + accessLogStoreMu.Lock() + accessLogStoreHolder = previous + accessLogStoreMu.Unlock() + } +} + +// NewMemoryAccessLogStore returns an in-memory access log store for unit tests. +func NewMemoryAccessLogStore() accessLogStore { + return &memoryAccessLogStore{ + records: make([]*OpenFlareAccessLog, 0), + } +} + +type clickhouseAccessLogStore struct{} + +func (clickhouseAccessLogStore) conn() (driver.Conn, error) { + if db.ChConn == nil { + return nil, errors.New(errClickHouseNotInitialized) + } + return db.ChConn, nil +} + +func (clickhouseAccessLogStore) InsertBatch(ctx context.Context, records []*OpenFlareAccessLog) error { + if len(records) == 0 { + return nil + } + conn, err := clickhouseAccessLogStore{}.conn() + if err != nil { + return err + } + batch, err := conn.PrepareBatch(ctx, fmt.Sprintf( + "INSERT INTO %s (id, node_id, logged_at, remote_addr, region, host, path, status_code, created_at)", + openFlareAccessLogTable, + )) + if err != nil { + return err + } + now := time.Now().UTC() + for _, record := range records { + if record == nil { + continue + } + id := record.ID + if id == 0 { + id = uint(idgen.NextUint64ID()) + } + createdAt := record.CreatedAt + if createdAt.IsZero() { + createdAt = now + } + if err := batch.Append( + uint64(id), + record.NodeID, + record.LoggedAt.UTC(), + record.RemoteAddr, + record.Region, + record.Host, + record.Path, + int32(record.StatusCode), + createdAt.UTC(), + ); err != nil { + return err + } + } + return batch.Send() +} + +func (clickhouseAccessLogStore) List(ctx context.Context, query OpenFlareAccessLogQuery) ([]*OpenFlareAccessLog, error) { + conn, err := clickhouseAccessLogStore{}.conn() + if err != nil { + return nil, err + } + clause, args := buildOpenFlareAccessLogFilterClause(query) + sql := fmt.Sprintf(` +SELECT id, node_id, logged_at, remote_addr, region, host, path, status_code, created_at +FROM %s +WHERE %s +ORDER BY %s`, openFlareAccessLogTable, clause, openFlareAccessLogOrderClause(query.SortBy, query.SortOrder)) + if query.PageSize > 0 { + if query.Page < 0 { + query.Page = 0 + } + sql += " LIMIT ? OFFSET ?" + args = append(args, query.PageSize, query.Page*query.PageSize) + } + rows, err := conn.Query(ctx, sql, args...) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + return scanOpenFlareAccessLogRows(rows) +} + +func scanOpenFlareAccessLogRows(rows driver.Rows) ([]*OpenFlareAccessLog, error) { + var result []*OpenFlareAccessLog + for rows.Next() { + var ( + id uint64 + nodeID string + loggedAt time.Time + remoteAddr string + region string + host string + path string + statusCode int32 + createdAt time.Time + ) + if err := rows.Scan(&id, &nodeID, &loggedAt, &remoteAddr, ®ion, &host, &path, &statusCode, &createdAt); err != nil { + return nil, err + } + result = append(result, &OpenFlareAccessLog{ + ID: uint(id), + NodeID: nodeID, + LoggedAt: loggedAt.UTC(), + RemoteAddr: remoteAddr, + Region: region, + Host: host, + Path: path, + StatusCode: int(statusCode), + CreatedAt: createdAt.UTC(), + }) + } + return result, nil +} + +func (clickhouseAccessLogStore) Count(ctx context.Context, query OpenFlareAccessLogQuery) (int64, int64, error) { + conn, err := clickhouseAccessLogStore{}.conn() + if err != nil { + return 0, 0, err + } + clause, args := buildOpenFlareAccessLogFilterClause(query) + var totalRecords int64 + countSQL := fmt.Sprintf("SELECT count() FROM %s WHERE %s", openFlareAccessLogTable, clause) + if err := conn.QueryRow(ctx, countSQL, args...).Scan(&totalRecords); err != nil { + return 0, 0, err + } + ipSQL := fmt.Sprintf(` +SELECT count() FROM ( + SELECT trim(remote_addr) AS remote_addr + FROM %s + WHERE %s AND remote_addr != '' + GROUP BY trim(remote_addr) +)`, openFlareAccessLogTable, clause) + var totalIPs int64 + if err := conn.QueryRow(ctx, ipSQL, args...).Scan(&totalIPs); err != nil { + return 0, 0, err + } + return totalRecords, totalIPs, nil +} + +func (clickhouseAccessLogStore) RegionCounts(ctx context.Context, nodeID string, since time.Time, limit int) ([]*OpenFlareAccessLogRegionCount, error) { + conn, err := clickhouseAccessLogStore{}.conn() + if err != nil { + return nil, err + } + filter := OpenFlareAccessLogQuery{NodeID: nodeID, Since: since} + clause, args := buildOpenFlareAccessLogFilterClause(filter) + sql := fmt.Sprintf(` +SELECT trim(region) AS region, count() AS count +FROM %s +WHERE %s AND trim(region) != '' +GROUP BY trim(region) +ORDER BY count DESC, region ASC`, openFlareAccessLogTable, clause) + if limit > 0 { + sql += " LIMIT ?" + args = append(args, limit) + } + rows, err := conn.Query(ctx, sql, args...) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var result []*OpenFlareAccessLogRegionCount + for rows.Next() { + var item OpenFlareAccessLogRegionCount + if err := rows.Scan(&item.Region, &item.Count); err != nil { + return nil, err + } + result = append(result, &item) + } + return result, nil +} + +func (clickhouseAccessLogStore) BucketAggregates(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogBucketAggregateRow, error) { + conn, err := clickhouseAccessLogStore{}.conn() + if err != nil { + return nil, err + } + clause, args := buildOpenFlareAccessLogFilterClause(filter) + bucketExpr := openFlareAccessLogBucketEpochExpr(bucketSeconds) + sql := fmt.Sprintf(` +SELECT + %s AS bucket_epoch, + count() AS request_count, + countIf(status_code < 400) AS success_count, + countIf(status_code >= 400 AND status_code < 500) AS client_error_count, + countIf(status_code >= 500) AS server_error_count +FROM %s +WHERE %s +GROUP BY bucket_epoch`, bucketExpr, openFlareAccessLogTable, clause) + rows, err := conn.Query(ctx, sql, args...) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var result []openFlareAccessLogBucketAggregateRow + for rows.Next() { + var item openFlareAccessLogBucketAggregateRow + if err := rows.Scan(&item.BucketEpoch, &item.RequestCount, &item.SuccessCount, &item.ClientErrorCount, &item.ServerErrorCount); err != nil { + return nil, err + } + result = append(result, item) + } + return result, nil +} + +func (clickhouseAccessLogStore) BucketDimensions(ctx context.Context, filter OpenFlareAccessLogQuery, column string, bucketSeconds int64) ([]openFlareAccessLogBucketDimensionRow, error) { + conn, err := clickhouseAccessLogStore{}.conn() + if err != nil { + return nil, err + } + clause, args := buildOpenFlareAccessLogFilterClause(filter) + bucketExpr := openFlareAccessLogBucketEpochExpr(bucketSeconds) + sql := fmt.Sprintf(` +SELECT + %s AS bucket_epoch, + trim(%s) AS value +FROM %s +WHERE %s AND trim(%s) != '' +GROUP BY bucket_epoch, trim(%s)`, bucketExpr, column, openFlareAccessLogTable, clause, column, column) + rows, err := conn.Query(ctx, sql, args...) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var result []openFlareAccessLogBucketDimensionRow + for rows.Next() { + var item openFlareAccessLogBucketDimensionRow + if err := rows.Scan(&item.BucketEpoch, &item.Value); err != nil { + return nil, err + } + result = append(result, item) + } + return result, nil +} + +func (clickhouseAccessLogStore) IPAggregates(ctx context.Context, filter OpenFlareAccessLogQuery, exactRemoteAddr bool) ([]openFlareAccessLogIPAggregateRow, error) { + conn, err := clickhouseAccessLogStore{}.conn() + if err != nil { + return nil, err + } + clause, args := buildOpenFlareAccessLogFilterClause(filter) + queryClause := clause + queryArgs := append([]any{}, args...) + if exactRemoteAddr { + trimmed := strings.TrimSpace(filter.RemoteAddr) + if trimmed == "" { + return []openFlareAccessLogIPAggregateRow{}, nil + } + queryClause = combineOpenFlareAccessLogSQLClauses(queryClause, "trim(remote_addr) = ?") + queryArgs = append(queryArgs, trimmed) + } + lastSeenExpr := openFlareAccessLogEpochExpr() + sql := fmt.Sprintf(` +SELECT + trim(remote_addr) AS remote_addr, + count() AS request_count, + countIf(status_code < 400) AS success_count, + countIf(status_code >= 400 AND status_code < 500) AS client_error_count, + countIf(status_code >= 500) AS server_error_count, + max(%s) AS last_seen_epoch +FROM %s +WHERE %s AND trim(remote_addr) != '' +GROUP BY trim(remote_addr)`, lastSeenExpr, openFlareAccessLogTable, queryClause) + rows, err := conn.Query(ctx, sql, queryArgs...) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var result []openFlareAccessLogIPAggregateRow + for rows.Next() { + var item openFlareAccessLogIPAggregateRow + if err := rows.Scan(&item.RemoteAddr, &item.RequestCount, &item.SuccessCount, &item.ClientErrorCount, &item.ServerErrorCount, &item.LastSeenEpoch); err != nil { + return nil, err + } + result = append(result, item) + } + return result, nil +} + +func (clickhouseAccessLogStore) IPSummaries(ctx context.Context, filter OpenFlareAccessLogQuery, recentSince time.Time) ([]openFlareAccessLogIPSummaryRow, error) { + conn, err := clickhouseAccessLogStore{}.conn() + if err != nil { + return nil, err + } + clause, args := buildOpenFlareAccessLogFilterClause(filter) + lastSeenExpr := openFlareAccessLogEpochExpr() + recentClause := "0" + queryArgs := make([]any, 0, len(args)+1) + if !recentSince.IsZero() { + recentClause = "if(logged_at >= ?, 1, 0)" + queryArgs = append(queryArgs, recentSince) + } + queryArgs = append(queryArgs, args...) + sql := fmt.Sprintf(` +SELECT + trim(remote_addr) AS remote_addr, + count() AS total_requests, + sum(%s) AS recent_requests, + max(%s) AS last_seen_epoch +FROM %s +WHERE %s AND trim(remote_addr) != '' +GROUP BY trim(remote_addr)`, recentClause, lastSeenExpr, openFlareAccessLogTable, clause) + rows, err := conn.Query(ctx, sql, queryArgs...) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var result []openFlareAccessLogIPSummaryRow + for rows.Next() { + var item openFlareAccessLogIPSummaryRow + if err := rows.Scan(&item.RemoteAddr, &item.TotalRequests, &item.RecentRequests, &item.LastSeenEpoch); err != nil { + return nil, err + } + result = append(result, item) + } + return result, nil +} + +func (clickhouseAccessLogStore) IPTrend(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogIPTrendRow, error) { + conn, err := clickhouseAccessLogStore{}.conn() + if err != nil { + return nil, err + } + clause, args := buildOpenFlareAccessLogFilterClause(filter) + bucketExpr := openFlareAccessLogBucketEpochExpr(bucketSeconds) + sql := fmt.Sprintf(` +SELECT + %s AS bucket_epoch, + count() AS request_count +FROM %s +WHERE %s +GROUP BY bucket_epoch +ORDER BY bucket_epoch ASC`, bucketExpr, openFlareAccessLogTable, clause) + rows, err := conn.Query(ctx, sql, args...) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var result []openFlareAccessLogIPTrendRow + for rows.Next() { + var item openFlareAccessLogIPTrendRow + if err := rows.Scan(&item.BucketEpoch, &item.RequestCount); err != nil { + return nil, err + } + result = append(result, item) + } + return result, nil +} + +func (s clickhouseAccessLogStore) DeleteAll(ctx context.Context) (int64, error) { + return s.deleteWithCount(ctx, "SELECT count() FROM "+openFlareAccessLogTable, nil, "ALTER TABLE "+openFlareAccessLogTable+" DELETE WHERE 1") +} + +func (s clickhouseAccessLogStore) DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error) { + return s.deleteWithCount( + ctx, + fmt.Sprintf("SELECT count() FROM %s WHERE logged_at < ?", openFlareAccessLogTable), + []any{cutoff.UTC()}, + fmt.Sprintf("ALTER TABLE %s DELETE WHERE logged_at < ?", openFlareAccessLogTable), + cutoff.UTC(), + ) +} + +func (s clickhouseAccessLogStore) DeleteByNodeBefore(ctx context.Context, nodeID string, before time.Time) (int64, error) { + return s.deleteWithCount( + ctx, + fmt.Sprintf("SELECT count() FROM %s WHERE node_id = ? AND logged_at < ?", openFlareAccessLogTable), + []any{nodeID, before.UTC()}, + fmt.Sprintf("ALTER TABLE %s DELETE WHERE node_id = ? AND logged_at < ?", openFlareAccessLogTable), + nodeID, before.UTC(), + ) +} + +func (clickhouseAccessLogStore) deleteWithCount(ctx context.Context, countSQL string, countArgs []any, deleteSQL string, deleteArgs ...any) (int64, error) { + conn, err := clickhouseAccessLogStore{}.conn() + if err != nil { + return 0, err + } + var count int64 + if err := conn.QueryRow(ctx, countSQL, countArgs...).Scan(&count); err != nil { + return 0, err + } + if count == 0 { + return 0, nil + } + if err := conn.Exec(ctx, deleteSQL, deleteArgs...); err != nil { + return 0, err + } + return count, nil +} + +func openFlareAccessLogBucketEpochExpr(bucketSeconds int64) string { + return fmt.Sprintf("toInt64(intDiv(toUnixTimestamp(logged_at), %d) * %d)", bucketSeconds, bucketSeconds) +} + +func openFlareAccessLogEpochExpr() string { + return "toInt64(toUnixTimestamp(logged_at))" +} \ No newline at end of file diff --git a/openflare-server/internal/model/openflare_access_log_store_memory.go b/openflare-server/internal/model/openflare_access_log_store_memory.go new file mode 100644 index 00000000..9ca71659 --- /dev/null +++ b/openflare-server/internal/model/openflare_access_log_store_memory.go @@ -0,0 +1,379 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package model + +import ( + "context" + "sort" + "strings" + "sync" + "time" + + "github.com/Rain-kl/Wavelet/internal/db/idgen" +) + +type memoryAccessLogStore struct { + mu sync.RWMutex + records []*OpenFlareAccessLog +} + +func (s *memoryAccessLogStore) InsertBatch(_ context.Context, records []*OpenFlareAccessLog) error { + s.mu.Lock() + defer s.mu.Unlock() + now := time.Now().UTC() + for _, record := range records { + if record == nil { + continue + } + copyRecord := *record + if copyRecord.ID == 0 { + copyRecord.ID = uint(idgen.NextUint64ID()) + } + if copyRecord.CreatedAt.IsZero() { + copyRecord.CreatedAt = now + } + copyRecord.LoggedAt = copyRecord.LoggedAt.UTC() + copyRecord.CreatedAt = copyRecord.CreatedAt.UTC() + s.records = append(s.records, ©Record) + } + return nil +} + +func (s *memoryAccessLogStore) List(_ context.Context, query OpenFlareAccessLogQuery) ([]*OpenFlareAccessLog, error) { + s.mu.RLock() + defer s.mu.RUnlock() + rows := s.filterRecords(query) + sortOpenFlareAccessLogRows(rows, query.SortBy, query.SortOrder) + if query.PageSize > 0 { + start, end := openFlareAccessLogPaginateBounds(len(rows), query.Page, query.PageSize) + return cloneAccessLogSlice(rows[start:end]), nil + } + return cloneAccessLogSlice(rows), nil +} + +func (s *memoryAccessLogStore) Count(_ context.Context, query OpenFlareAccessLogQuery) (int64, int64, error) { + s.mu.RLock() + defer s.mu.RUnlock() + rows := s.filterRecords(query) + ips := make(map[string]struct{}) + for _, row := range rows { + remoteAddr := strings.TrimSpace(row.RemoteAddr) + if remoteAddr == "" { + continue + } + ips[remoteAddr] = struct{}{} + } + return int64(len(rows)), int64(len(ips)), nil +} + +func (s *memoryAccessLogStore) RegionCounts(_ context.Context, nodeID string, since time.Time, limit int) ([]*OpenFlareAccessLogRegionCount, error) { + s.mu.RLock() + defer s.mu.RUnlock() + rows := s.filterRecords(OpenFlareAccessLogQuery{NodeID: nodeID, Since: since}) + counts := make(map[string]int64) + for _, row := range rows { + region := strings.TrimSpace(row.Region) + if region == "" { + continue + } + counts[region]++ + } + result := make([]*OpenFlareAccessLogRegionCount, 0, len(counts)) + for region, count := range counts { + result = append(result, &OpenFlareAccessLogRegionCount{Region: region, Count: count}) + } + sort.Slice(result, func(i, j int) bool { + if result[i].Count == result[j].Count { + return result[i].Region < result[j].Region + } + return result[i].Count > result[j].Count + }) + if limit > 0 && len(result) > limit { + result = result[:limit] + } + return result, nil +} + +func (s *memoryAccessLogStore) BucketAggregates(_ context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogBucketAggregateRow, error) { + s.mu.RLock() + defer s.mu.RUnlock() + rows := s.filterRecords(filter) + aggregates := make(map[int64]*openFlareAccessLogBucketAggregateRow) + for _, row := range rows { + bucketEpoch := memoryAccessLogBucketEpoch(row.LoggedAt, bucketSeconds) + item := aggregates[bucketEpoch] + if item == nil { + item = &openFlareAccessLogBucketAggregateRow{BucketEpoch: bucketEpoch} + aggregates[bucketEpoch] = item + } + item.RequestCount++ + switch { + case row.StatusCode < 400: + item.SuccessCount++ + case row.StatusCode < 500: + item.ClientErrorCount++ + default: + item.ServerErrorCount++ + } + } + result := make([]openFlareAccessLogBucketAggregateRow, 0, len(aggregates)) + for _, item := range aggregates { + result = append(result, *item) + } + return result, nil +} + +func (s *memoryAccessLogStore) BucketDimensions(_ context.Context, filter OpenFlareAccessLogQuery, column string, bucketSeconds int64) ([]openFlareAccessLogBucketDimensionRow, error) { + s.mu.RLock() + defer s.mu.RUnlock() + rows := s.filterRecords(filter) + seen := make(map[int64]map[string]struct{}) + var result []openFlareAccessLogBucketDimensionRow + for _, row := range rows { + var value string + switch column { + case "remote_addr": + value = strings.TrimSpace(row.RemoteAddr) + case "host": + value = strings.TrimSpace(row.Host) + default: + continue + } + if value == "" { + continue + } + bucketEpoch := memoryAccessLogBucketEpoch(row.LoggedAt, bucketSeconds) + if seen[bucketEpoch] == nil { + seen[bucketEpoch] = make(map[string]struct{}) + } + if _, ok := seen[bucketEpoch][value]; ok { + continue + } + seen[bucketEpoch][value] = struct{}{} + result = append(result, openFlareAccessLogBucketDimensionRow{BucketEpoch: bucketEpoch, Value: value}) + } + return result, nil +} + +func (s *memoryAccessLogStore) IPAggregates(_ context.Context, filter OpenFlareAccessLogQuery, exactRemoteAddr bool) ([]openFlareAccessLogIPAggregateRow, error) { + s.mu.RLock() + defer s.mu.RUnlock() + if exactRemoteAddr && strings.TrimSpace(filter.RemoteAddr) == "" { + return []openFlareAccessLogIPAggregateRow{}, nil + } + rows := s.filterRecords(filter) + aggregates := make(map[string]*openFlareAccessLogIPAggregateRow) + for _, row := range rows { + remoteAddr := strings.TrimSpace(row.RemoteAddr) + if remoteAddr == "" { + continue + } + if exactRemoteAddr && remoteAddr != strings.TrimSpace(filter.RemoteAddr) { + continue + } + item := aggregates[remoteAddr] + if item == nil { + item = &openFlareAccessLogIPAggregateRow{RemoteAddr: remoteAddr} + aggregates[remoteAddr] = item + } + item.RequestCount++ + epoch := row.LoggedAt.UTC().Unix() + if epoch > item.LastSeenEpoch { + item.LastSeenEpoch = epoch + } + switch { + case row.StatusCode < 400: + item.SuccessCount++ + case row.StatusCode < 500: + item.ClientErrorCount++ + default: + item.ServerErrorCount++ + } + } + result := make([]openFlareAccessLogIPAggregateRow, 0, len(aggregates)) + for _, item := range aggregates { + result = append(result, *item) + } + return result, nil +} + +func (s *memoryAccessLogStore) IPSummaries(_ context.Context, filter OpenFlareAccessLogQuery, recentSince time.Time) ([]openFlareAccessLogIPSummaryRow, error) { + s.mu.RLock() + defer s.mu.RUnlock() + rows := s.filterRecords(filter) + aggregates := make(map[string]*openFlareAccessLogIPSummaryRow) + for _, row := range rows { + remoteAddr := strings.TrimSpace(row.RemoteAddr) + if remoteAddr == "" { + continue + } + item := aggregates[remoteAddr] + if item == nil { + item = &openFlareAccessLogIPSummaryRow{RemoteAddr: remoteAddr} + aggregates[remoteAddr] = item + } + item.TotalRequests++ + if !recentSince.IsZero() && !row.LoggedAt.Before(recentSince) { + item.RecentRequests++ + } + epoch := row.LoggedAt.UTC().Unix() + if epoch > item.LastSeenEpoch { + item.LastSeenEpoch = epoch + } + } + result := make([]openFlareAccessLogIPSummaryRow, 0, len(aggregates)) + for _, item := range aggregates { + result = append(result, *item) + } + return result, nil +} + +func (s *memoryAccessLogStore) IPTrend(_ context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogIPTrendRow, error) { + s.mu.RLock() + defer s.mu.RUnlock() + rows := s.filterRecords(filter) + aggregates := make(map[int64]int64) + for _, row := range rows { + bucketEpoch := memoryAccessLogBucketEpoch(row.LoggedAt, bucketSeconds) + aggregates[bucketEpoch]++ + } + result := make([]openFlareAccessLogIPTrendRow, 0, len(aggregates)) + for bucketEpoch, count := range aggregates { + result = append(result, openFlareAccessLogIPTrendRow{BucketEpoch: bucketEpoch, RequestCount: count}) + } + sort.Slice(result, func(i, j int) bool { return result[i].BucketEpoch < result[j].BucketEpoch }) + return result, nil +} + +func (s *memoryAccessLogStore) DeleteAll(_ context.Context) (int64, error) { + s.mu.Lock() + defer s.mu.Unlock() + count := int64(len(s.records)) + s.records = nil + return count, nil +} + +func (s *memoryAccessLogStore) DeleteBefore(_ context.Context, cutoff time.Time) (int64, error) { + s.mu.Lock() + defer s.mu.Unlock() + cutoff = cutoff.UTC() + remaining := make([]*OpenFlareAccessLog, 0, len(s.records)) + var deleted int64 + for _, row := range s.records { + if row.LoggedAt.Before(cutoff) { + deleted++ + continue + } + remaining = append(remaining, row) + } + s.records = remaining + return deleted, nil +} + +func (s *memoryAccessLogStore) DeleteByNodeBefore(_ context.Context, nodeID string, before time.Time) (int64, error) { + s.mu.Lock() + defer s.mu.Unlock() + before = before.UTC() + remaining := make([]*OpenFlareAccessLog, 0, len(s.records)) + var deleted int64 + for _, row := range s.records { + if row.NodeID == nodeID && row.LoggedAt.Before(before) { + deleted++ + continue + } + remaining = append(remaining, row) + } + s.records = remaining + return deleted, nil +} + +func (s *memoryAccessLogStore) filterRecords(query OpenFlareAccessLogQuery) []*OpenFlareAccessLog { + result := make([]*OpenFlareAccessLog, 0, len(s.records)) + for _, row := range s.records { + if !memoryAccessLogMatches(row, query) { + continue + } + result = append(result, row) + } + return result +} + +func memoryAccessLogMatches(row *OpenFlareAccessLog, query OpenFlareAccessLogQuery) bool { + if row == nil { + return false + } + if trimmed := strings.TrimSpace(query.NodeID); trimmed != "" && row.NodeID != trimmed { + return false + } + if trimmed := strings.TrimSpace(query.RemoteAddr); trimmed != "" && !strings.HasPrefix(strings.TrimSpace(row.RemoteAddr), trimmed) { + return false + } + if trimmed := strings.TrimSpace(query.Host); trimmed != "" && !strings.HasPrefix(strings.TrimSpace(row.Host), trimmed) { + return false + } + if trimmed := strings.TrimSpace(query.Path); trimmed != "" && !strings.HasPrefix(strings.TrimSpace(row.Path), trimmed) { + return false + } + if !query.Since.IsZero() && row.LoggedAt.Before(query.Since) { + return false + } + if !query.Until.IsZero() && !row.LoggedAt.Before(query.Until) { + return false + } + return true +} + +func memoryAccessLogBucketEpoch(loggedAt time.Time, bucketSeconds int64) int64 { + if bucketSeconds <= 0 { + bucketSeconds = 180 + } + epoch := loggedAt.UTC().Unix() + return (epoch / bucketSeconds) * bucketSeconds +} + +func cloneAccessLogSlice(rows []*OpenFlareAccessLog) []*OpenFlareAccessLog { + result := make([]*OpenFlareAccessLog, len(rows)) + for index, row := range rows { + if row == nil { + continue + } + copyRecord := *row + result[index] = ©Record + } + return result +} + +func sortOpenFlareAccessLogRows(items []*OpenFlareAccessLog, sortBy string, sortOrder string) { + desc := openFlareAccessLogNormalizeSortOrder(sortOrder) != "asc" + sort.Slice(items, func(i, j int) bool { + left := items[i] + right := items[j] + if left == nil || right == nil { + return left != nil + } + var compare int + switch strings.TrimSpace(sortBy) { + case "status_code": + compare = left.StatusCode - right.StatusCode + case "remote_addr": + compare = strings.Compare(left.RemoteAddr, right.RemoteAddr) + case "host": + compare = strings.Compare(left.Host, right.Host) + case "path": + compare = strings.Compare(left.Path, right.Path) + default: + compare = openFlareAccessLogCompareInt64(left.LoggedAt.Unix(), right.LoggedAt.Unix()) + } + if compare == 0 { + compare = openFlareAccessLogCompareInt64(left.LoggedAt.Unix(), right.LoggedAt.Unix()) + } + if compare == 0 { + compare = openFlareAccessLogCompareInt64(int64(left.ID), int64(right.ID)) + } + if desc { + return compare > 0 + } + return compare < 0 + }) +} \ No newline at end of file diff --git a/openflare-server/internal/model/openflare_access_log_test.go b/openflare-server/internal/model/openflare_access_log_test.go index e0c50549..920f1227 100644 --- a/openflare-server/internal/model/openflare_access_log_test.go +++ b/openflare-server/internal/model/openflare_access_log_test.go @@ -9,23 +9,16 @@ import ( "testing" "time" - "github.com/Rain-kl/Wavelet/internal/db" - "github.com/glebarez/sqlite" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "gorm.io/gorm" ) func setupOpenFlareAccessLogTestEnvironment(t *testing.T) (context.Context, func()) { t.Helper() - sqliteDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{ - DisableForeignKeyConstraintWhenMigrating: true, - }) - require.NoError(t, err) - require.NoError(t, sqliteDB.AutoMigrate(&OpenFlareAccessLog{})) - db.SetDB(sqliteDB) + store := NewMemoryAccessLogStore() + reset := SetAccessLogStoreForTest(store) return context.Background(), func() { - db.SetDB(nil) + reset() } } @@ -38,9 +31,7 @@ func seedOpenFlareAccessLogs(t *testing.T, ctx context.Context, now time.Time) { {NodeID: "node-b", LoggedAt: now.Add(-2 * time.Minute), RemoteAddr: "3.3.3.3", Region: "EU", Host: "b.example.com", Path: "/delta", StatusCode: 200}, {NodeID: "node-b", LoggedAt: now.Add(-1 * time.Minute), RemoteAddr: "", Region: "", Host: "b.example.com", Path: "/empty-ip", StatusCode: 200}, } - for index, record := range records { - require.NoError(t, db.DB(ctx).Create(record).Error, "seed access log %d", index) - } + require.NoError(t, InsertOpenFlareAccessLogsBatch(ctx, records)) } func TestListOpenFlareAccessLogsPaginated(t *testing.T) { @@ -57,7 +48,7 @@ func TestListOpenFlareAccessLogsPaginated(t *testing.T) { Path: fmt.Sprintf("/path-%02d", index), StatusCode: 200, } - require.NoError(t, db.DB(ctx).Create(record).Error) + require.NoError(t, InsertOpenFlareAccessLogsBatch(ctx, []*OpenFlareAccessLog{record})) } query := OpenFlareAccessLogQuery{ @@ -111,22 +102,6 @@ func TestListOpenFlareAccessLogsFiltersAndSort(t *testing.T) { assert.Equal(t, 200, rows[1].StatusCode) } -func TestListOpenFlareAccessLogsMissingTableGraceful(t *testing.T) { - ctx, cleanup := setupOpenFlareAccessLogTestEnvironment(t) - defer cleanup() - require.NoError(t, db.DB(ctx).Migrator().DropTable(&OpenFlareAccessLog{})) - - query := OpenFlareAccessLogQuery{Since: time.Now().UTC().Add(-time.Hour)} - rows, err := ListOpenFlareAccessLogs(ctx, query) - require.NoError(t, err) - assert.Empty(t, rows) - - totalRecords, totalIPs, err := CountOpenFlareAccessLogs(ctx, query) - require.NoError(t, err) - assert.Zero(t, totalRecords) - assert.Zero(t, totalIPs) -} - func TestDeleteOpenFlareAccessLogsBefore(t *testing.T) { ctx, cleanup := setupOpenFlareAccessLogTestEnvironment(t) defer cleanup() @@ -141,4 +116,4 @@ func TestDeleteOpenFlareAccessLogsBefore(t *testing.T) { totalRecords, _, err := CountOpenFlareAccessLogs(ctx, OpenFlareAccessLogQuery{Since: now.Add(-10 * time.Minute)}) require.NoError(t, err) assert.Equal(t, int64(2), totalRecords) -} +} \ No newline at end of file diff --git a/openflare-server/internal/model/openflare_observability.go b/openflare-server/internal/model/openflare_observability.go index 8611b5fb..7963829a 100644 --- a/openflare-server/internal/model/openflare_observability.go +++ b/openflare-server/internal/model/openflare_observability.go @@ -55,7 +55,7 @@ func (OpenFlareRequestReport) TableName() string { return "of_node_request_reports" } -// OpenFlareAccessLog stores a single access log row (v1 single table, no sharding). +// OpenFlareAccessLog stores a single access log row in ClickHouse (database: openflare). type OpenFlareAccessLog struct { ID uint `json:"id" gorm:"primaryKey;autoIncrement"` NodeID string `json:"node_id" gorm:"index;size:64;not null"` diff --git a/openflare-server/support-files/sql/create_clickhouse_openflare.sql b/openflare-server/support-files/sql/create_clickhouse_openflare.sql new file mode 100644 index 00000000..52c2f178 --- /dev/null +++ b/openflare-server/support-files/sql/create_clickhouse_openflare.sql @@ -0,0 +1,38 @@ +CREATE DATABASE IF NOT EXISTS openflare; + +USE openflare; + +CREATE TABLE IF NOT EXISTS of_node_access_logs +( + id UInt64, + node_id String, + logged_at DateTime64(3, 'UTC'), + remote_addr String, + region String, + host String, + path String, + status_code Int32, + created_at DateTime64(3, 'UTC') +) +ENGINE = MergeTree() +PARTITION BY toYYYYMM(logged_at) +ORDER BY (node_id, logged_at, remote_addr, host, path, status_code) +SETTINGS index_granularity = 8192; + +CREATE TABLE IF NOT EXISTS w_user_access_logs +( + id UInt64, + user_id UInt64, + path String, + method String, + ip String, + user_agent String, + headers String, + status Int32, + latency Int64, + created_at DateTime +) +ENGINE = MergeTree() +PARTITION BY toYYYYMM(created_at) +ORDER BY (created_at, ip, user_id) +SETTINGS index_granularity = 8192; \ No newline at end of file