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

merge: migrate of_node_access_logs to ClickHouse

Browse Source 
Integrate clickhouse worktree: node access logs now stored in ClickHouse
(openflare database) with mandatory startup dependency.
This commit is contained in:
ryan
2026-06-19 11:45:51 +08:00
parent 7eb943f02f 425ca89765
commit fff26aa343
24 changed files with 1224 additions and 458 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/changelog/index.md
+5
View File
@@ -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 统一控制面。
openflare-server/.env.example
+5 -7
View File
@@ -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
openflare-server/config.example.yaml
+2 -2
View File
@@ -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"
openflare-server/docker-compose.yml
+22 -23
View File
@@ -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
openflare-server/internal/apps/admin/logs/routers.go
+4 -7
View File
@@ -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
}
openflare-server/internal/apps/openflare/agent/observability.go
+52 -56
View File
@@ -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, " ", "_")
openflare-server/internal/apps/openflare/async_tasks_test.go
+4 -3
View File
@@ -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
openflare-server/internal/apps/openflare/dashboard/logics_test.go
+2
View File
@@ -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)
}
}
openflare-server/internal/apps/openflare/node/logics_test.go
+2
View File
@@ -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()
}
openflare-server/internal/apps/openflare/option/logics_test.go
+20 -18
View File
@@ -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{
openflare-server/internal/apps/openflare/tasks/database_cleanup_test.go
+22 -19
View File
@@ -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),
openflare-server/internal/apps/openflare/waf/ip_group_sync_test.go
+6 -3
View File
@@ -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))
}
openflare-server/internal/config/config.go
+36 -2
View File
@@ -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
}
}
// ─── 环境变量覆盖层 ────────────────────────────────────────────────────────────
openflare-server/internal/db/clickhouse.go
+5
View File
@@ -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")
}
openflare-server/internal/db/clickhouse_schema.go
+51
View File
@@ -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)")
}
openflare-server/internal/db/migrator/goose/postgres/202606200004_drop_of_node_access_logs.sql
+28
View File
@@ -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);
openflare-server/internal/db/migrator/goose/sqlite/202606200004_drop_of_node_access_logs.sql
+28
View File
@@ -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);
openflare-server/internal/model/errs.go
+1
View File
@@ -6,6 +6,7 @@ package model
const (
errRegistrationDisabled = "注册已关闭"
errDatabaseNotInitialized = "database not initialized"
errClickHouseNotInitialized = "clickhouse not initialized"
errUsernameExists = "用户名已存在"
errEmailAlreadyBound = "该邮箱已被其他账号绑定"
errConfigIntParseFailed = "配置 %s 的值 '%s' 无法转换为整数: %w"
openflare-server/internal/model/openflare_access_log.go
+40 -286
View File
@@ -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
}
}
openflare-server/internal/model/openflare_access_log_store.go
+465
View File
@@ -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, &region, &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))"
}
openflare-server/internal/model/openflare_access_log_store_memory.go
+379
View File
@@ -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, &copyRecord)
}
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] = &copyRecord
}
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
})
}
openflare-server/internal/model/openflare_access_log_test.go
+6 -31
View File
@@ -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)
}
}
openflare-server/internal/model/openflare_observability.go
+1 -1
View File
@@ -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"`
openflare-server/support-files/sql/create_clickhouse_openflare.sql
+38
View File
@@ -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;