mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-08 00:26:37 +08:00
feat(cordis): add OpenFlare Cordis 架构改造设计
docs(changelog): 修正表述笔误
refactor(cordis): 磁盘缓存改用上上游能力并清理本地副本
按上游/下游归属规约：类型断言守卫已回流 Wavelet（f3d85d5，附回归用例），
本仓库删除 OpenFlare/plugins/server/pkg/cache 整包并改 import 到
Wavelet/pkg/cache/disk，同步后与上游零漂移。
验证：go build 通过；go test ./... exit 0（137 包 ok）；256 条路由对拍与
232 条 swagger 操作均零差异；make build-all 四进制；前端零改动。
docs(cordis): 记录 T1 清理结果与五个复用阻塞点
refactor(cordis): server 复用上游 pkg 能力并删除等价本地副本
按上游/下游归属规约清理重复实现，删除 7 个与上游等价的本地包并改 import：
shared/response→pkg/response、pkg/{logger,mail,trace,httppool,cache/ram}→
上游同名包、infra/persistence/batchwriter→pkg/batchwriter。逐项核过差异：
httppool 逐字节相同；logger 的 Config 字段完全一致；response 的 7 个 Abort*
一致；cache/ram 换过去顺带把裸 go 变回带 panic 恢复的 util.Go。
两处非等价差异按语义处理：
- batchwriter.Stats 与 status DTO 原为类型别名，改为消费侧逐字段转换，
避免 model 反向依赖基础设施类型；
- 上游 pkg/idgen 要求显式 Init（本地副本为懒加载自动初始化），本次保留本地
副本，待与 infra 初始化一并迁移（已登记在清理计划）。
验证：go build 通过；go test ./... exit 0（138 包 ok）；256 条路由对拍零差异；
make swagger 232 条操作零增减，且归一化后与旧文档深度相等——差异仅为
response.Any / logger.LogEntry 两个定义名随包路径改名，接口形状未变。
chore(cordis): 回流内核与 pkg/util 通用能力并清理 vendoring 污染
按新增的上游/下游归属规约：HandleRaw/BasePath 与版本比较、网络、格式化助手
属通用能力，已提交到 Wavelet 分支 feat/cordis-router-raw-routes，本仓库改为
纯同步获取（pkg/util 已零漂移），补丁登记保留至上游合并。
同时修掉我此前 git add -A 造成的污染：首次 vendoring 把上游工作区里被
gitignore 的运行期产物一起提交进来（upload 的 diskcache 缓存块 650 个与
driver_http/dist 前端构建物 380 个，共 12872 行/1030 文件）。sync-upstream.sh
现显式排除 uploads/dist/data/*.db，.gitignore 补上对应兜底规则。
AGENTS.md 增加上游/下游改动归属规约，并把仍指向前 Cordis 布局的硬性约束
（internal/router + Serve、internal/repository/logstore、internal/platform/bootstrap、
internal/cmd）改到当前插件路径。
验证：go build 通过；go test ./... exit 0（144 包 ok）；make swagger 232 条
操作与基线逐条一致；make build-all 四进制；gofmt 干净。
feat(cordis): server 插件化并改由内核挂载控制面路由
新增 plugins/server/plugin.go：Apply 以 ctx.Router().Group(app.api_prefix)
声明根级与 /v1 全部路由；33 个注册函数由 *gin.RouterGroup 改为
core.RouterExtension，RegisterCollection 改用内核新增的 HandleRaw 保留
尾部斜杠变体，AdminMiddlewares 返回 []any（Go 不允许把 []T 展开为 ...any）。
删除 router.Serve 与 registerRoutes，装配根改为 core.App +
driver_http.New(WithEngine(router.BuildEngine()))，监听、信号与优雅退出归内核；
前端 SPA 的 NoRoute 兜底因内核暂无贡献点而保留在引擎层。
路由保真证据：plugin_parity_test 对拍 baseline/routes-engine.txt 的 256 条
(方法 路径) 零差异；go test ./... exit 0（144 包 ok，含真实 handler 的
openflare/integration 用例走同一条挂载路径）；make swagger 232 条操作与基线
逐条一致；golangci-lint 0 issues；make build-all 四进制；embed_frontend
标签编译通过；前端零改动。
已知待补：带 Redis 的实机 HTTP 冒烟（本机 6379 未启动，session store 与
改造前一样在建店阶段即 fatal），以及 bootstrap 的任务/设置/迁移注册迁入 Apply。
feat(core): RouterExtension 增加 HandleRaw 与 BasePath 以保真尾部斜杠路由
server 插件化的前置：Handle 经 cleanPath 会剥掉尾部斜杠，无法表达
/resource 与 /resource/ 两条不同路由，而 OpenFlare 有 20 个历史 list
端点两者都注册且部署关闭了 RedirectTrailingSlash，缺失即 404。新增
HandleRaw 与 BasePath（作用域包装器同样登记反注册），补 extpoints 用例；
并把 router.Serve 拆出 BuildEngine 以便交给 driver_http.WithEngine 复用，
新增路由表导出 harness，固化 256 条 (方法 路径) 基线供插件化对拍。
上游补丁登记于 backend/OpenFlare/upstream-patches.md，同步脚本改为按目录
前缀输出差异并在同步后提醒确认补丁是否仍在。
验证：go build 通过；go test ./... exit 0（143 包 ok）；gofmt 干净。
docs(cordis): 记录 server 插件接入内核的可行路径与内核能力缺口
feat(cordis): agent/relay/flared 落地为内核驱动插件
三个边缘守护进程各新增 plugin.go，实现 core.Plugin + core.Driver
（自定义 DriverType 与同名 profile），装配与生命周期从 main 迁入
Apply/Start/Stop：Apply 负责 JSON 配置加载、运行环境与用户确保、
openresty/frps/frpc 管理器与各服务装配；Start 以 util.Go 拉起阻塞式
runner 与 GeoIP 周期更新；Stop 收敛主循环结果并在超时时报错而非静默。
入口改为 core.NewApp(core.WithProfile(...)) + Prepare/Run，保持
-config 旗标、默认路径、退出码与启动/停止日志不变。
验证：go build 通过；go test ./... exit 0（143 包 ok，含 3 个插件身份
与配置失败路径测试）；make build-all 四进制产出；三进制实跑缺失配置
均 exit 1 且错误链保留 load {agent,relay,flared} config 原因；gofmt 干净。
refactor(cordis): 按功能职责拆分为 4 个插件与 share 共享层
backend/OpenFlare 不再平铺遗留分层，改为 plugins/{server,agent,relay,flared}
加 share/：控制面业务（openflare/admin/oauth/user/upload/cap/config/health 与
repository/model/infra/router 等支撑层）归 server；三个边缘守护进程各自成插件；
被两个以上插件消费的 protocol/geoip/wsclient/render/pagesarchive/edge 归 share。
同时把 pkg/util 与 buildinfo 合并回上游 pkg（上游已覆盖全部符号，仅 8 个函数与
2 个类型为 OpenFlare 独有，已一并迁入），装配根统一到 backend/cmd（含三个 daemon
入口），Dockerfile 与 release 工作流的构建路径和 -X 注入路径同步更新。
验证：go build 通过；go test ./... exit 0（141 包 ok）；make swagger exit 0 且
232 条 API 操作与基线逐条一致；make build-all 产出 4 进制；-X 注入经二进制
strings 实测生效；日志后端直连门禁改写为按 server 插件业务域扫描并在扫描数为 0
时报错（防门禁静默失效）；前端零改动。
feat(cordis): 落地 backend/share 共享层与上游同步脚本
跨插件共享资源（控制消息协议、GeoIP+iputil、边缘守护进程日志）从下游包
移入 backend/share，并声明其只能依赖 core/pkg 与标准/第三方库，禁止反向
引用下游业务与具体插件实现；新增 scripts/sync-upstream.sh 只覆盖
backend/{core,pkg,plugins}，同步后 --check 报告零差异，证明与上游逐字一致。
go build 通过，go test ./... exit 0（142 包 ok），前端零改动。
refactor(cordis): 采用与 Wavelet 同构的单模块布局并引入上游内核
按上游结构落位：backend/{core,pkg,plugins} 为 Wavelet 上游拷贝，OpenFlare
全部业务收拢到上游 downstream 所对应的位置 backend/OpenFlare/，模块名保持
Wavelet 以保证上游 import 路径逐字一致、同步零改写；三个 daemon 入口移至
backend/OpenFlare/cmd，backend/cmd 与 main.go 作为控制面装配根。
行为不变：go build 通过，142 个测试包全绿（含上游插件测试），232 条 API
操作与改造前逐条一致，四进制产物正常，前端零改动。swagger 暂只扫描下游代码，
待 P4 挂载上游路由后再纳入 plugins/。
style: 修正模块路径改写导致的 import 分组排序漂移
refactor(layout): Go 代码迁入 backend/ 并将模块名简化为 OpenFlare
对齐上游 Wavelet 的仓库布局，为以第二 module 形态 vendoring Cordis 内核与
平台插件做准备：模块路径整体改写为 OpenFlare，Go 目标加 cd backend，
swaggo 产物移至 backend/docs 并把 json/yaml 复制回 docs/ 供站点消费，
Dockerfile 与 release 工作流的构建目录、ldflags 模块路径同步更新。
行为保持不变：232 条路由与改造前逐条一致，95 个测试包全绿，
四进制产物正常，前端零改动。
chore(cordis): 落地改造计划与 schema/路由基线
新增 legacy_dump_test 迁移快照 harness：在临时 sqlite 库上按生产顺序
（goose.UpTo → zone 导入 → goose.Up）跑完 76 个历史迁移并导出 schema 与
版本序列，作为改造前后一致性门禁的唯一事实来源。同时记录 232 条路由清单
与 foundation 实施计划。
docs(cordis): add OpenFlare Cordis 架构改造设计
明确上游以第二 module 形态 vendoring 进 backend/Wavelet、4 个插件
(server/agent/relay/flared) 全部装载内核，并规定保留 76 个历史 goose
迁移 + 一次性版本 stamp 桥接的迁移方案，配套三方 schema 一致性门禁，
确保已部署库不重跑历史、不丢数据。
This commit is contained in:
ryan
@@ -0,0 +1,10 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package risk_control
|
||||
|
||||
// 模块内专用错误文案常量，集中在此文件维护，禁止在 handler 中内联。
|
||||
const (
|
||||
// errSystemBusy 访问日志缓冲队列满载时的限流提示，刻意使用模糊文案避免泄露内部容量细节。
|
||||
errSystemBusy = "系统繁忙，请稍后再试"
|
||||
)
|
||||
@@ -0,0 +1,105 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
// Package logstore provides data access for analytics tables.
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"Wavelet/pkg/util"
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// CountAccessLogs returns the number of access logs matching filter.
|
||||
func CountAccessLogs(ctx context.Context, filter AccessLogFilter) (uint64, error) {
|
||||
ch := getChDB(ctx)
|
||||
if ch == nil {
|
||||
return 0, fmt.Errorf("clickhouse gorm connection is not initialized")
|
||||
}
|
||||
|
||||
var count int64
|
||||
query := applyFilter(ch.Model(&UserAccessLog{}), filter)
|
||||
if err := query.Count(&count).Error; err != nil {
|
||||
return 0, fmt.Errorf("count access logs: %w", err)
|
||||
}
|
||||
return safeUint64Count(count), nil
|
||||
}
|
||||
|
||||
// ListAccessLogs returns paginated access logs and the total match count.
|
||||
func ListAccessLogs(ctx context.Context, filter AccessLogFilter, page, pageSize int) ([]UserAccessLog, uint64, error) {
|
||||
ch := getChDB(ctx)
|
||||
if ch == nil {
|
||||
return nil, 0, fmt.Errorf("clickhouse gorm connection is not initialized")
|
||||
}
|
||||
|
||||
if filter.UserIDs != nil && len(filter.UserIDs) == 0 {
|
||||
return []UserAccessLog{}, 0, nil
|
||||
}
|
||||
|
||||
var count int64
|
||||
query := applyFilter(ch.Model(&UserAccessLog{}), filter)
|
||||
if err := query.Count(&count).Error; err != nil {
|
||||
return nil, 0, fmt.Errorf("count access logs: %w", err)
|
||||
}
|
||||
|
||||
var logs []UserAccessLog
|
||||
offset := (page - 1) * pageSize
|
||||
if err := query.Order("created_at DESC").Limit(pageSize).Offset(offset).Find(&logs).Error; err != nil {
|
||||
return nil, 0, fmt.Errorf("list access logs: %w", err)
|
||||
}
|
||||
|
||||
return logs, safeUint64Count(count), nil
|
||||
}
|
||||
|
||||
// DeleteAllUserAccessLogs hard-deletes all user access logs via TRUNCATE.
|
||||
func DeleteAllUserAccessLogs(ctx context.Context) (int64, error) {
|
||||
conn := getChConn()
|
||||
if conn == nil {
|
||||
return 0, fmt.Errorf("clickhouse connection is not initialized")
|
||||
}
|
||||
if err := conn.Exec(ctx, "TRUNCATE TABLE "+UserAccessLog{}.TableName()); err != nil {
|
||||
return 0, fmt.Errorf("truncate user access logs: %w", err)
|
||||
}
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// DeleteUserAccessLogsBefore deletes user access logs older than cutoff.
|
||||
func DeleteUserAccessLogsBefore(ctx context.Context, cutoff time.Time) (int64, error) {
|
||||
conn := getChConn()
|
||||
if conn == nil {
|
||||
return 0, fmt.Errorf("clickhouse connection is not initialized")
|
||||
}
|
||||
if err := conn.Exec(ctx, "ALTER TABLE "+UserAccessLog{}.TableName()+" DELETE WHERE created_at < ?", cutoff); err != nil {
|
||||
return 0, fmt.Errorf("delete expired user access logs: %w", err)
|
||||
}
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
func safeUint64Count(count int64) uint64 {
|
||||
if count < 0 {
|
||||
return 0
|
||||
}
|
||||
return uint64(count)
|
||||
}
|
||||
|
||||
func applyFilter(query *gorm.DB, filter AccessLogFilter) *gorm.DB {
|
||||
if filter.UserIDs != nil {
|
||||
if len(filter.UserIDs) == 0 {
|
||||
return query.Where("1 = 0")
|
||||
}
|
||||
query = query.Where("user_id IN ?", filter.UserIDs)
|
||||
}
|
||||
if filter.Path != "" {
|
||||
query = query.Where("path LIKE ? ESCAPE '\\'", "%"+util.EscapeLike(filter.Path)+"%")
|
||||
}
|
||||
if filter.StartTime != nil {
|
||||
query = query.Where("created_at >= ?", *filter.StartTime)
|
||||
}
|
||||
if filter.EndTime != nil {
|
||||
query = query.Where("created_at <= ?", *filter.EndTime)
|
||||
}
|
||||
return query
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import "time"
|
||||
|
||||
// AccessLogFilter scopes user access log queries.
|
||||
type AccessLogFilter struct {
|
||||
// UserIDs filters by user IDs. nil means no user filter; an empty slice means no matches.
|
||||
UserIDs []uint64
|
||||
Path string
|
||||
// StartTime filters created_at >= StartTime when non-nil.
|
||||
StartTime *time.Time
|
||||
// EndTime filters created_at <= EndTime when non-nil.
|
||||
EndTime *time.Time
|
||||
}
|
||||
|
||||
// DailyTrend is a single day's access count.
|
||||
type DailyTrend struct {
|
||||
Date string
|
||||
Count uint64
|
||||
}
|
||||
|
||||
// BrowserShare is a browser group's share of access logs.
|
||||
type BrowserShare struct {
|
||||
Browser string
|
||||
Count uint64
|
||||
}
|
||||
|
||||
// TopUser is an active user ranked by access count.
|
||||
type TopUser struct {
|
||||
UserID uint64
|
||||
Count uint64
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
)
|
||||
|
||||
const hoursInDay = 24
|
||||
|
||||
// GetDailyTrend returns per-day access counts for the last days days (inclusive of today).
|
||||
func GetDailyTrend(ctx context.Context, days int) ([]DailyTrend, error) {
|
||||
if days < 1 {
|
||||
days = 7
|
||||
}
|
||||
|
||||
ch := getChDB(ctx)
|
||||
if ch == nil {
|
||||
return nil, fmt.Errorf("clickhouse gorm connection is not initialized")
|
||||
}
|
||||
|
||||
startTime := time.Now().AddDate(0, 0, -(days - 1)).Truncate(hoursInDay * time.Hour)
|
||||
tableName := UserAccessLog{}.TableName()
|
||||
|
||||
query := fmt.Sprintf(`
|
||||
SELECT toDate(created_at) AS date, count() AS count
|
||||
FROM %s
|
||||
WHERE created_at >= ?
|
||||
GROUP BY date
|
||||
ORDER BY date ASC
|
||||
`, tableName)
|
||||
|
||||
type trendRow struct {
|
||||
Date time.Time
|
||||
Count uint64
|
||||
}
|
||||
|
||||
var rows []trendRow
|
||||
if err := ch.Raw(query, startTime).Scan(&rows).Error; err != nil {
|
||||
return nil, fmt.Errorf("get daily trend: %w", err)
|
||||
}
|
||||
|
||||
trendMap := make(map[string]uint64, days)
|
||||
for i := 0; i < days; i++ {
|
||||
dateStr := time.Now().AddDate(0, 0, -i).Format("2006-01-02")
|
||||
trendMap[dateStr] = 0
|
||||
}
|
||||
for _, row := range rows {
|
||||
dateStr := row.Date.Format("2006-01-02")
|
||||
trendMap[dateStr] = row.Count
|
||||
}
|
||||
|
||||
result := make([]DailyTrend, 0, days)
|
||||
for i := days - 1; i >= 0; i-- {
|
||||
dateStr := time.Now().AddDate(0, 0, -i).Format("2006-01-02")
|
||||
result = append(result, DailyTrend{
|
||||
Date: dateStr,
|
||||
Count: trendMap[dateStr],
|
||||
})
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// GetBrowserDistribution returns browser-grouped access counts since startTime.
|
||||
func GetBrowserDistribution(ctx context.Context, startTime time.Time) ([]BrowserShare, error) {
|
||||
ch := getChDB(ctx)
|
||||
if ch == nil {
|
||||
return nil, fmt.Errorf("clickhouse gorm connection is not initialized")
|
||||
}
|
||||
|
||||
tableName := UserAccessLog{}.TableName()
|
||||
query := fmt.Sprintf(`
|
||||
SELECT user_agent, count() AS count
|
||||
FROM %s
|
||||
WHERE created_at >= ?
|
||||
GROUP BY user_agent
|
||||
`, tableName)
|
||||
|
||||
type uaRow struct {
|
||||
UserAgent string
|
||||
Count uint64
|
||||
}
|
||||
|
||||
var rows []uaRow
|
||||
if err := ch.Raw(query, startTime).Scan(&rows).Error; err != nil {
|
||||
return nil, fmt.Errorf("get browser distribution: %w", err)
|
||||
}
|
||||
|
||||
browserCounts := make(map[string]uint64)
|
||||
for _, row := range rows {
|
||||
browser := ParseBrowserName(row.UserAgent)
|
||||
browserCounts[browser] += row.Count
|
||||
}
|
||||
|
||||
result := make([]BrowserShare, 0, len(browserCounts))
|
||||
for browser, count := range browserCounts {
|
||||
result = append(result, BrowserShare{
|
||||
Browser: browser,
|
||||
Count: count,
|
||||
})
|
||||
}
|
||||
sort.Slice(result, func(i, j int) bool {
|
||||
return result[i].Count > result[j].Count
|
||||
})
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// GetTopActiveUsers returns the most active users since startTime.
|
||||
func GetTopActiveUsers(ctx context.Context, startTime time.Time, limit int) ([]TopUser, error) {
|
||||
if limit < 1 {
|
||||
limit = 10
|
||||
}
|
||||
|
||||
ch := getChDB(ctx)
|
||||
if ch == nil {
|
||||
return nil, fmt.Errorf("clickhouse gorm connection is not initialized")
|
||||
}
|
||||
|
||||
tableName := UserAccessLog{}.TableName()
|
||||
query := fmt.Sprintf(`
|
||||
SELECT user_id, count() AS count
|
||||
FROM %s
|
||||
WHERE created_at >= ? AND user_id > 0
|
||||
GROUP BY user_id
|
||||
ORDER BY count DESC
|
||||
LIMIT ?
|
||||
`, tableName)
|
||||
|
||||
var users []TopUser
|
||||
if err := ch.Raw(query, startTime, limit).Scan(&users).Error; err != nil {
|
||||
return nil, fmt.Errorf("get top active users: %w", err)
|
||||
}
|
||||
return users, nil
|
||||
}
|
||||
@@ -0,0 +1,213 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/column"
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
|
||||
|
||||
"github.com/glebarez/sqlite"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func setupChGormDB(t *testing.T) *gorm.DB {
|
||||
t.Helper()
|
||||
|
||||
gormDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{
|
||||
DisableForeignKeyConstraintWhenMigrating: true,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, gormDB.AutoMigrate(&UserAccessLog{}))
|
||||
SetChDBForTest(gormDB)
|
||||
return gormDB
|
||||
}
|
||||
|
||||
func TestParseBrowserName(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
ua string
|
||||
want string
|
||||
}{
|
||||
{name: "chrome", ua: "Mozilla/5.0 Chrome/120.0.0.0", want: "Chrome"},
|
||||
{name: "firefox", ua: "Mozilla/5.0 Firefox/121.0", want: "Firefox"},
|
||||
{name: "safari", ua: "Mozilla/5.0 Safari/605.1.15", want: "Safari"},
|
||||
{name: "edge", ua: "Mozilla/5.0 Edg/120.0.0.0", want: "Edge"},
|
||||
{name: "wechat", ua: "MicroMessenger/8.0", want: "WeChat"},
|
||||
{name: "postman", ua: "PostmanRuntime/7.36.0", want: "Postman"},
|
||||
{name: "other", ua: "curl/8.0", want: "Other"},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
assert.Equal(t, tt.want, ParseBrowserName(tt.ua))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCountAccessLogs_EmptyUserIDs(t *testing.T) {
|
||||
setupChGormDB(t)
|
||||
t.Cleanup(func() { SetChDBForTest(nil) })
|
||||
|
||||
count, err := CountAccessLogs(context.Background(), AccessLogFilter{UserIDs: []uint64{}})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, uint64(0), count)
|
||||
}
|
||||
|
||||
func TestListAccessLogs_EmptyUserIDs(t *testing.T) {
|
||||
setupChGormDB(t)
|
||||
t.Cleanup(func() { SetChDBForTest(nil) })
|
||||
|
||||
logs, total, err := ListAccessLogs(context.Background(), AccessLogFilter{UserIDs: []uint64{}}, 1, 20)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, uint64(0), total)
|
||||
assert.Empty(t, logs)
|
||||
}
|
||||
|
||||
func TestListAccessLogs_WithFilters(t *testing.T) {
|
||||
gormDB := setupChGormDB(t)
|
||||
t.Cleanup(func() { SetChDBForTest(nil) })
|
||||
|
||||
now := time.Now().UTC().Truncate(time.Second)
|
||||
logs := []UserAccessLog{
|
||||
{ID: 1, UserID: 10, Path: "/api/v1/users", Method: "GET", Status: 200, CreatedAt: now},
|
||||
{ID: 2, UserID: 20, Path: "/api/v1/admin/logs", Method: "GET", Status: 200, CreatedAt: now},
|
||||
{ID: 3, UserID: 10, Path: "/api/v1/other", Method: "POST", Status: 201, CreatedAt: now},
|
||||
}
|
||||
require.NoError(t, gormDB.Create(&logs).Error)
|
||||
|
||||
start := now.Add(-time.Hour)
|
||||
filter := AccessLogFilter{
|
||||
UserIDs: []uint64{10},
|
||||
Path: "users",
|
||||
StartTime: &start,
|
||||
}
|
||||
|
||||
count, err := CountAccessLogs(context.Background(), filter)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, uint64(1), count)
|
||||
|
||||
result, total, err := ListAccessLogs(context.Background(), filter, 1, 10)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, uint64(1), total)
|
||||
require.Len(t, result, 1)
|
||||
assert.Equal(t, uint64(1), result[0].ID)
|
||||
assert.Equal(t, "/api/v1/users", result[0].Path)
|
||||
}
|
||||
|
||||
func TestBatchInsert_Empty(t *testing.T) {
|
||||
err := BatchInsert(context.Background(), nil)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestBatchInsert_UsesModelBatchSQL(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
mockBatch := &mockBatch{}
|
||||
mockConn := &mockConn{
|
||||
batch: mockBatch,
|
||||
batchQuery: UserAccessLog{}.BatchInsertSQL(),
|
||||
}
|
||||
SetChConnForTest(mockConn)
|
||||
t.Cleanup(func() { SetChConnForTest(nil) })
|
||||
|
||||
createdAt := time.Now().UTC()
|
||||
err := BatchInsert(ctx, []UserAccessLog{
|
||||
{
|
||||
ID: 1,
|
||||
UserID: 42,
|
||||
Path: "/api/v1/test",
|
||||
Method: "GET",
|
||||
IP: "127.0.0.1",
|
||||
UserAgent: "test-agent",
|
||||
Headers: "{}",
|
||||
Status: 200,
|
||||
Latency: 12,
|
||||
CreatedAt: createdAt,
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.True(t, mockConn.prepareCalled)
|
||||
assert.Equal(t, UserAccessLog{}.BatchInsertSQL(), mockConn.preparedQuery)
|
||||
assert.True(t, mockBatch.sendCalled)
|
||||
require.Len(t, mockBatch.rows, 1)
|
||||
assert.Equal(t, uint64(42), mockBatch.rows[0][1])
|
||||
}
|
||||
|
||||
type mockConn struct {
|
||||
batch driver.Batch
|
||||
batchQuery string
|
||||
prepareCalled bool
|
||||
preparedQuery string
|
||||
}
|
||||
|
||||
func (m *mockConn) Contributors() []string { return nil }
|
||||
|
||||
func (m *mockConn) ServerVersion() (*driver.ServerVersion, error) { return nil, nil }
|
||||
|
||||
func (m *mockConn) Select(_ context.Context, _ any, _ string, _ ...any) error { return nil }
|
||||
|
||||
func (m *mockConn) Query(_ context.Context, _ string, _ ...any) (driver.Rows, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (m *mockConn) QueryRow(_ context.Context, _ string, _ ...any) driver.Row { return nil }
|
||||
|
||||
func (m *mockConn) PrepareBatch(_ context.Context, query string, _ ...driver.PrepareBatchOption) (driver.Batch, error) {
|
||||
m.prepareCalled = true
|
||||
m.preparedQuery = query
|
||||
return m.batch, nil
|
||||
}
|
||||
|
||||
func (m *mockConn) Exec(_ context.Context, _ string, _ ...any) error { return nil }
|
||||
|
||||
func (m *mockConn) AsyncInsert(_ context.Context, _ string, _ bool, _ ...any) error { return nil }
|
||||
|
||||
func (m *mockConn) InsertFormat(_ context.Context, _, _ string, _ io.Reader) error { return nil }
|
||||
|
||||
func (m *mockConn) QueryFormat(_ context.Context, _, _ string, _ ...any) (io.ReadCloser, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (m *mockConn) Ping(_ context.Context) error { return nil }
|
||||
|
||||
func (m *mockConn) Stats() driver.Stats { return driver.Stats{} }
|
||||
|
||||
func (m *mockConn) Close() error { return nil }
|
||||
|
||||
type mockBatch struct {
|
||||
rows [][]any
|
||||
sendCalled bool
|
||||
}
|
||||
|
||||
func (m *mockBatch) Abort() error { return nil }
|
||||
|
||||
func (m *mockBatch) Append(v ...any) error {
|
||||
m.rows = append(m.rows, v)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockBatch) AppendStruct(_ any) error { return nil }
|
||||
|
||||
func (m *mockBatch) Column(_ int) driver.BatchColumn { return nil }
|
||||
|
||||
func (m *mockBatch) Flush() error { return nil }
|
||||
|
||||
func (m *mockBatch) Send() error {
|
||||
m.sendCalled = true
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockBatch) IsSent() bool { return m.sendCalled }
|
||||
|
||||
func (m *mockBatch) Rows() int { return len(m.rows) }
|
||||
|
||||
func (m *mockBatch) Columns() []column.Interface { return nil }
|
||||
|
||||
func (m *mockBatch) Close() error { return nil }
|
||||
@@ -0,0 +1,47 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
// BatchInsert writes access logs to ClickHouse using the native batch API.
|
||||
func BatchInsert(ctx context.Context, logs []UserAccessLog) error {
|
||||
if len(logs) == 0 {
|
||||
return nil
|
||||
}
|
||||
conn := getChConn()
|
||||
if conn == nil {
|
||||
return fmt.Errorf("clickhouse connection is not initialized")
|
||||
}
|
||||
|
||||
batch, err := conn.PrepareBatch(ctx, UserAccessLog{}.BatchInsertSQL())
|
||||
if err != nil {
|
||||
return fmt.Errorf("prepare clickhouse batch: %w", err)
|
||||
}
|
||||
|
||||
for _, logItem := range logs {
|
||||
if err := batch.Append(
|
||||
logItem.ID,
|
||||
logItem.UserID,
|
||||
logItem.Path,
|
||||
logItem.Method,
|
||||
logItem.IP,
|
||||
logItem.UserAgent,
|
||||
logItem.Headers,
|
||||
logItem.Status,
|
||||
logItem.Latency,
|
||||
logItem.CreatedAt,
|
||||
); err != nil {
|
||||
return fmt.Errorf("append access log to batch: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := batch.Send(); err != nil {
|
||||
return fmt.Errorf("send clickhouse batch: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import "strings"
|
||||
|
||||
// ParseBrowserName performs lightweight User-Agent browser identification.
|
||||
func ParseBrowserName(ua string) string {
|
||||
uaLower := strings.ToLower(ua)
|
||||
if strings.Contains(uaLower, "micromessenger") {
|
||||
return "WeChat"
|
||||
}
|
||||
if strings.Contains(uaLower, "postman") {
|
||||
return "Postman"
|
||||
}
|
||||
if strings.Contains(uaLower, "edg/") || strings.Contains(uaLower, "edge") {
|
||||
return "Edge"
|
||||
}
|
||||
if strings.Contains(uaLower, "firefox") {
|
||||
return "Firefox"
|
||||
}
|
||||
if strings.Contains(uaLower, "chrome") {
|
||||
return "Chrome"
|
||||
}
|
||||
if strings.Contains(uaLower, "safari") {
|
||||
return "Safari"
|
||||
}
|
||||
return "Other"
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"Wavelet/pkg/logger"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultLogRetentionDays = 30
|
||||
partitionLeadMonths = 2
|
||||
userAccessLogTable = "w_user_access_logs"
|
||||
)
|
||||
|
||||
// CleanupSummary 汇总本次清理结果。
|
||||
type CleanupSummary struct {
|
||||
ActiveDatabase string `json:"active_database"`
|
||||
RetentionDays int `json:"retention_days"`
|
||||
Deleted int64 `json:"deleted"`
|
||||
}
|
||||
|
||||
// CleanupExpired 按当前日志库保留天数删除过期用户访问日志，并预建 PG 分区。
|
||||
func CleanupExpired(ctx context.Context) (CleanupSummary, error) {
|
||||
active, err := ActiveDatabase(ctx)
|
||||
if err != nil {
|
||||
return CleanupSummary{}, err
|
||||
}
|
||||
days := retentionDaysForDatabase(ctx, active)
|
||||
summary := CleanupSummary{ActiveDatabase: active, RetentionDays: days}
|
||||
|
||||
store, err := Active(ctx)
|
||||
if err != nil {
|
||||
return summary, err
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
if err := store.UserAccessLogs.EnsurePartitions(ctx, now, now.AddDate(0, partitionLeadMonths, 0)); err != nil {
|
||||
logger.WarnF(ctx, "logstore: ensure partitions during cleanup failed: %v", err)
|
||||
}
|
||||
cutoff := now.AddDate(0, 0, -days)
|
||||
// 先 DROP 完全过期的整月分区，再对边界月逐行 DeleteBefore。
|
||||
if err := store.UserAccessLogs.DropExpiredPartitions(ctx, cutoff); err != nil {
|
||||
return summary, fmt.Errorf("drop expired partitions: %w", err)
|
||||
}
|
||||
deleted, err := store.UserAccessLogs.DeleteBefore(ctx, cutoff)
|
||||
if err != nil {
|
||||
return summary, fmt.Errorf("delete expired user access logs: %w", err)
|
||||
}
|
||||
summary.Deleted = deleted
|
||||
if err := store.UserAccessLogs.DropEmptyPartitions(ctx, now); err != nil {
|
||||
logger.WarnF(ctx, "drop empty log partitions failed: %v", err)
|
||||
}
|
||||
return summary, nil
|
||||
}
|
||||
|
||||
func retentionDaysForDatabase(ctx context.Context, dbName string) int {
|
||||
key := "log_retention_days_postgres"
|
||||
switch dbName {
|
||||
case dbNameSQLite:
|
||||
key = "log_retention_days_sqlite"
|
||||
case dbNameClickHouse:
|
||||
key = "log_retention_days_clickhouse"
|
||||
}
|
||||
v, err := getConfig(ctx, key)
|
||||
if err != nil {
|
||||
if !errors.Is(err, errConfigReaderNotWired) {
|
||||
logger.ErrorF(ctx, "读取日志保留天数配置失败(key=%s)，回退默认 %d 天: %v", key, defaultLogRetentionDays, err)
|
||||
}
|
||||
return defaultLogRetentionDays
|
||||
}
|
||||
days, perr := strconv.Atoi(v)
|
||||
if perr != nil || days <= 0 {
|
||||
logger.ErrorF(ctx, "日志保留天数配置非法(key=%s, value=%q)，回退默认 %d 天", key, v, defaultLogRetentionDays)
|
||||
return defaultLogRetentionDays
|
||||
}
|
||||
return days
|
||||
}
|
||||
|
||||
func partitionStatementsRange(from, to time.Time) []string {
|
||||
var out []string
|
||||
start := time.Date(from.Year(), from.Month(), 1, 0, 0, 0, 0, time.UTC)
|
||||
end := time.Date(to.Year(), to.Month(), 1, 0, 0, 0, 0, time.UTC).AddDate(0, 1, 0)
|
||||
for ; start.Before(end); start = start.AddDate(0, 1, 0) {
|
||||
monthEnd := start.AddDate(0, 1, 0)
|
||||
suffix := start.Format("200601")
|
||||
fromDay := start.Format("2006-01-02")
|
||||
toDay := monthEnd.Format("2006-01-02")
|
||||
out = append(out, fmt.Sprintf(
|
||||
"CREATE TABLE IF NOT EXISTS %s_%s PARTITION OF %s FOR VALUES FROM ('%s') TO ('%s')",
|
||||
userAccessLogTable, suffix, userAccessLogTable, fromDay, toDay))
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
|
||||
)
|
||||
|
||||
type clickhouseUserAccessLogStore struct {
|
||||
skipFreeze bool
|
||||
}
|
||||
|
||||
func newClickHouseUserAccessLogStore() *clickhouseUserAccessLogStore {
|
||||
return &clickhouseUserAccessLogStore{}
|
||||
}
|
||||
|
||||
var (
|
||||
_ UserAccessLogStore = (*clickhouseUserAccessLogStore)(nil)
|
||||
_ StatusStore = (*clickhouseUserAccessLogStore)(nil)
|
||||
)
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) ActiveDatabase(_ context.Context) (string, error) {
|
||||
return dbNameClickHouse, nil
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) ensureWritable(ctx context.Context) error {
|
||||
if !s.skipFreeze && Migrating(ctx) {
|
||||
return ErrMigrating
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) BatchInsert(ctx context.Context, logs []UserAccessLog) error {
|
||||
if len(logs) == 0 {
|
||||
return nil
|
||||
}
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
return BatchInsert(ctx, logs)
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) DeleteAll(ctx context.Context) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return DeleteAllUserAccessLogs(ctx)
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return DeleteUserAccessLogsBefore(ctx, cutoff)
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) Count(ctx context.Context, filter AccessLogFilter) (uint64, error) {
|
||||
return CountAccessLogs(ctx, filter)
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) List(ctx context.Context, filter AccessLogFilter, page, pageSize int) ([]UserAccessLog, uint64, error) {
|
||||
return ListAccessLogs(ctx, filter, page, pageSize)
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) GetDailyTrend(ctx context.Context, days int) ([]DailyTrend, error) {
|
||||
return GetDailyTrend(ctx, days)
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) GetBrowserDistribution(ctx context.Context, startTime time.Time) ([]BrowserShare, error) {
|
||||
return GetBrowserDistribution(ctx, startTime)
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) GetTopActiveUsers(ctx context.Context, startTime time.Time, limit int) ([]TopUser, error) {
|
||||
return GetTopActiveUsers(ctx, startTime, limit)
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) EnsurePartitions(_ context.Context, _, _ time.Time) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) DropEmptyPartitions(_ context.Context, _ time.Time) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) DropExpiredPartitions(_ context.Context, _ time.Time) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) MigrationRange(ctx context.Context) (time.Time, time.Time, error) {
|
||||
conn := getChConn()
|
||||
if conn == nil {
|
||||
return time.Time{}, time.Time{}, fmt.Errorf("clickhouse connection is not initialized")
|
||||
}
|
||||
table := UserAccessLog{}.TableName()
|
||||
var minTime, maxTime *time.Time
|
||||
if err := conn.QueryRow(ctx, "SELECT min(created_at), max(created_at) FROM "+table).Scan(&minTime, &maxTime); err != nil {
|
||||
return time.Time{}, time.Time{}, fmt.Errorf("query migration range %s: %w", table, err)
|
||||
}
|
||||
if minTime == nil || maxTime == nil {
|
||||
return time.Time{}, time.Time{}, nil
|
||||
}
|
||||
return minTime.UTC(), maxTime.UTC(), nil
|
||||
}
|
||||
|
||||
func (s *clickhouseUserAccessLogStore) ListForMigration(ctx context.Context, afterID uint64, limit int) ([]UserAccessLog, error) {
|
||||
conn := getChConn()
|
||||
if conn == nil {
|
||||
return nil, fmt.Errorf("clickhouse connection is not initialized")
|
||||
}
|
||||
if limit <= 0 {
|
||||
limit = migrationPageSize
|
||||
}
|
||||
table := UserAccessLog{}.TableName()
|
||||
columns := UserAccessLog{}.InsertColumns()
|
||||
rows, err := conn.Query(ctx, fmt.Sprintf(
|
||||
"SELECT %s FROM %s WHERE id > ? ORDER BY id ASC LIMIT ?",
|
||||
columns, table,
|
||||
), afterID, limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list user access logs for migration: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
return scanUserAccessLogs(rows)
|
||||
}
|
||||
|
||||
func scanUserAccessLogs(rows driver.Rows) ([]UserAccessLog, error) {
|
||||
var result []UserAccessLog
|
||||
for rows.Next() {
|
||||
var item UserAccessLog
|
||||
if err := rows.Scan(
|
||||
&item.ID,
|
||||
&item.UserID,
|
||||
&item.Path,
|
||||
&item.Method,
|
||||
&item.IP,
|
||||
&item.UserAgent,
|
||||
&item.Headers,
|
||||
&item.Status,
|
||||
&item.Latency,
|
||||
&item.CreatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("scan user access log row: %w", err)
|
||||
}
|
||||
item.CreatedAt = item.CreatedAt.UTC()
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"Wavelet/core"
|
||||
"Wavelet/core/contracts"
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
var (
|
||||
dbMu sync.RWMutex
|
||||
dbSvc contracts.DBService
|
||||
chConn driver.Conn
|
||||
chDB *gorm.DB
|
||||
)
|
||||
|
||||
// SetDBService configures the DBService instance for logstore.
|
||||
func SetDBService(s contracts.DBService) {
|
||||
dbMu.Lock()
|
||||
defer dbMu.Unlock()
|
||||
dbSvc = s
|
||||
}
|
||||
|
||||
// SetChConnForTest configures ClickHouse native connection for test or runtime.
|
||||
func SetChConnForTest(conn driver.Conn) {
|
||||
dbMu.Lock()
|
||||
defer dbMu.Unlock()
|
||||
chConn = conn
|
||||
}
|
||||
|
||||
// SetChDBForTest configures ClickHouse GORM DB for test or runtime.
|
||||
func SetChDBForTest(db *gorm.DB) {
|
||||
dbMu.Lock()
|
||||
defer dbMu.Unlock()
|
||||
chDB = db
|
||||
}
|
||||
|
||||
func getDB(ctx context.Context) *gorm.DB {
|
||||
if c, ok := ctx.(*core.Context); ok && c != nil {
|
||||
if s, err := core.Inject[contracts.DBService](c); err == nil && s != nil {
|
||||
return s.DB(ctx)
|
||||
}
|
||||
}
|
||||
dbMu.RLock()
|
||||
s := dbSvc
|
||||
dbMu.RUnlock()
|
||||
if s != nil {
|
||||
return s.DB(ctx)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func getChDB(ctx context.Context) *gorm.DB {
|
||||
dbMu.RLock()
|
||||
customCh := chDB
|
||||
s := dbSvc
|
||||
dbMu.RUnlock()
|
||||
if customCh != nil {
|
||||
return customCh.WithContext(ctx)
|
||||
}
|
||||
if s != nil {
|
||||
if ch := s.Named("clickhouse"); ch != nil {
|
||||
return ch.WithContext(ctx)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func getChConn() driver.Conn {
|
||||
dbMu.RLock()
|
||||
defer dbMu.RUnlock()
|
||||
return chConn
|
||||
}
|
||||
@@ -0,0 +1,326 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"Wavelet/pkg/idgen"
|
||||
"Wavelet/pkg/util"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const (
|
||||
insertBatchSize = 500
|
||||
migrationPageSize = 100
|
||||
defaultPageSize = 20
|
||||
defaultTopN = 10
|
||||
topUserAgents = 100
|
||||
dayDuration = 24 * time.Hour
|
||||
)
|
||||
|
||||
type gormLogStore struct {
|
||||
db *gorm.DB
|
||||
skipFreeze bool
|
||||
}
|
||||
|
||||
func newGormStore(db *gorm.DB) *gormLogStore { return &gormLogStore{db: db} }
|
||||
|
||||
type userAccessLogGormStore struct {
|
||||
*gormLogStore
|
||||
}
|
||||
|
||||
func newUserAccessLogGormStore(db *gorm.DB) *userAccessLogGormStore {
|
||||
return &userAccessLogGormStore{gormLogStore: newGormStore(db)}
|
||||
}
|
||||
|
||||
var (
|
||||
_ UserAccessLogStore = (*userAccessLogGormStore)(nil)
|
||||
_ StatusStore = (*userAccessLogGormStore)(nil)
|
||||
)
|
||||
|
||||
func (s *gormLogStore) ActiveDatabase(_ context.Context) (string, error) {
|
||||
if isPostgresDialect(s.db) {
|
||||
return dbNamePostgres, nil
|
||||
}
|
||||
return dbNameSQLite, nil
|
||||
}
|
||||
|
||||
func (s *gormLogStore) ensureWritable(ctx context.Context) error {
|
||||
if !s.skipFreeze && Migrating(ctx) {
|
||||
return ErrMigrating
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *userAccessLogGormStore) BatchInsert(ctx context.Context, logs []UserAccessLog) error {
|
||||
if len(logs) == 0 {
|
||||
return nil
|
||||
}
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
for i := range logs {
|
||||
if logs[i].ID == 0 {
|
||||
logs[i].ID = idgen.NextUint64ID()
|
||||
}
|
||||
}
|
||||
return s.db.WithContext(ctx).CreateInBatches(logs, insertBatchSize).Error
|
||||
}
|
||||
|
||||
func (s *userAccessLogGormStore) DeleteAll(ctx context.Context) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
res := s.db.WithContext(ctx).Where("1 = 1").Delete(&UserAccessLog{})
|
||||
return res.RowsAffected, res.Error
|
||||
}
|
||||
|
||||
func (s *userAccessLogGormStore) DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error) {
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
res := s.db.WithContext(ctx).Where("created_at < ?", cutoff).Delete(&UserAccessLog{})
|
||||
if res.Error != nil && isMissingRelation(res.Error) {
|
||||
return 0, nil
|
||||
}
|
||||
return res.RowsAffected, res.Error
|
||||
}
|
||||
|
||||
func (s *userAccessLogGormStore) ListForMigration(ctx context.Context, afterID uint64, limit int) ([]UserAccessLog, error) {
|
||||
var rows []UserAccessLog
|
||||
q := s.db.WithContext(ctx).Model(&UserAccessLog{}).
|
||||
Where("id > ?", afterID).
|
||||
Order("id ASC").
|
||||
Limit(limitOr(limit, migrationPageSize))
|
||||
if err := q.Find(&rows).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return rows, nil
|
||||
}
|
||||
|
||||
func (s *userAccessLogGormStore) MigrationRange(ctx context.Context) (time.Time, time.Time, error) {
|
||||
return gormMigrationRange(ctx, s.db, "created_at", UserAccessLog{}, func(v *UserAccessLog) time.Time {
|
||||
return v.CreatedAt
|
||||
})
|
||||
}
|
||||
|
||||
func (s *userAccessLogGormStore) Count(ctx context.Context, filter AccessLogFilter) (uint64, error) {
|
||||
where, args, ok := buildUserAccessLogWhere(filter)
|
||||
if !ok {
|
||||
return 0, nil
|
||||
}
|
||||
var total int64
|
||||
if err := s.db.WithContext(ctx).Model(&UserAccessLog{}).Where(where, args...).Count(&total).Error; err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return countToUint64(total), nil
|
||||
}
|
||||
|
||||
func (s *userAccessLogGormStore) List(ctx context.Context, filter AccessLogFilter, page, pageSize int) ([]UserAccessLog, uint64, error) {
|
||||
where, args, ok := buildUserAccessLogWhere(filter)
|
||||
if !ok {
|
||||
return []UserAccessLog{}, 0, nil
|
||||
}
|
||||
var total int64
|
||||
if err := s.db.WithContext(ctx).Model(&UserAccessLog{}).Where(where, args...).Count(&total).Error; err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
if total == 0 {
|
||||
return []UserAccessLog{}, 0, nil
|
||||
}
|
||||
var rows []UserAccessLog
|
||||
q := s.db.WithContext(ctx).Where(where, args...).Order("created_at DESC, id DESC")
|
||||
if err := q.Limit(limitOr(pageSize, defaultPageSize)).Offset(offsetOf(page, pageSize)).Find(&rows).Error; err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return rows, countToUint64(total), nil
|
||||
}
|
||||
|
||||
func buildUserAccessLogWhere(filter AccessLogFilter) (string, []any, bool) {
|
||||
if filter.UserIDs != nil && len(filter.UserIDs) == 0 {
|
||||
return "", nil, false
|
||||
}
|
||||
var parts []string
|
||||
var args []any
|
||||
if filter.UserIDs != nil {
|
||||
parts = append(parts, "user_id IN ?")
|
||||
args = append(args, filter.UserIDs)
|
||||
}
|
||||
if trimmed := strings.TrimSpace(filter.Path); trimmed != "" {
|
||||
parts = append(parts, "path LIKE ? ESCAPE '\\'")
|
||||
args = append(args, "%"+util.EscapeLike(trimmed)+"%")
|
||||
}
|
||||
if filter.StartTime != nil {
|
||||
parts = append(parts, "created_at >= ?")
|
||||
args = append(args, *filter.StartTime)
|
||||
}
|
||||
if filter.EndTime != nil {
|
||||
parts = append(parts, "created_at <= ?")
|
||||
args = append(args, *filter.EndTime)
|
||||
}
|
||||
if len(parts) == 0 {
|
||||
return "1 = 1", args, true
|
||||
}
|
||||
return strings.Join(parts, " AND "), args, true
|
||||
}
|
||||
|
||||
func (s *userAccessLogGormStore) GetDailyTrend(ctx context.Context, days int) ([]DailyTrend, error) {
|
||||
if days <= 0 {
|
||||
days = 7
|
||||
}
|
||||
start := time.Now().AddDate(0, 0, -(days - 1)).Truncate(dayDuration)
|
||||
type row struct {
|
||||
Date string
|
||||
Cnt uint64
|
||||
}
|
||||
var rows []row
|
||||
err := s.db.WithContext(ctx).Model(&UserAccessLog{}).
|
||||
Select(dailyTrendDateSQL(s.db)+" AS date, COUNT(*) AS cnt").
|
||||
Where("created_at >= ?", start).
|
||||
Group("date").Order("date ASC").Scan(&rows).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
counts := make(map[string]uint64, len(rows))
|
||||
for _, r := range rows {
|
||||
counts[r.Date] = r.Cnt
|
||||
}
|
||||
out := make([]DailyTrend, 0, days)
|
||||
for i := 0; i < days; i++ {
|
||||
d := start.AddDate(0, 0, i).Format("2006-01-02")
|
||||
out = append(out, DailyTrend{Date: d, Count: counts[d]})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *userAccessLogGormStore) GetBrowserDistribution(ctx context.Context, startTime time.Time) ([]BrowserShare, error) {
|
||||
type row struct {
|
||||
UserAgent string
|
||||
Cnt uint64
|
||||
}
|
||||
var rows []row
|
||||
err := s.db.WithContext(ctx).Model(&UserAccessLog{}).
|
||||
Select("user_agent, COUNT(*) AS cnt").
|
||||
Where("created_at >= ?", startTime).
|
||||
Group("user_agent").Order("cnt DESC").Limit(topUserAgents).Scan(&rows).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
counts := make(map[string]uint64)
|
||||
for _, r := range rows {
|
||||
counts[ParseBrowserName(r.UserAgent)] += r.Cnt
|
||||
}
|
||||
out := make([]BrowserShare, 0, len(counts))
|
||||
for label, count := range counts {
|
||||
out = append(out, BrowserShare{Browser: label, Count: count})
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Count > out[j].Count })
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *userAccessLogGormStore) GetTopActiveUsers(ctx context.Context, startTime time.Time, limit int) ([]TopUser, error) {
|
||||
type row struct {
|
||||
UserID uint64
|
||||
Cnt uint64
|
||||
}
|
||||
var rows []row
|
||||
err := s.db.WithContext(ctx).Model(&UserAccessLog{}).
|
||||
Select("user_id, COUNT(*) AS cnt").
|
||||
Where("user_id <> 0 AND created_at >= ?", startTime).
|
||||
Group("user_id").Order("cnt DESC").Limit(limitOr(limit, defaultTopN)).Scan(&rows).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]TopUser, len(rows))
|
||||
for i, r := range rows {
|
||||
out[i] = TopUser{UserID: r.UserID, Count: r.Cnt}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *userAccessLogGormStore) EnsurePartitions(ctx context.Context, from, to time.Time) error {
|
||||
if !isPostgresDialect(s.db) {
|
||||
return nil
|
||||
}
|
||||
for _, sql := range partitionStatementsRange(from, to) {
|
||||
if err := s.db.WithContext(ctx).Exec(sql).Error; err != nil {
|
||||
return fmt.Errorf("ensure partition: %w", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func gormMigrationRange[T any](
|
||||
ctx context.Context,
|
||||
gdb *gorm.DB,
|
||||
column string,
|
||||
model T,
|
||||
timeOf func(*T) time.Time,
|
||||
) (time.Time, time.Time, error) {
|
||||
var first, last T
|
||||
found := false
|
||||
for _, order := range []string{"ASC", "DESC"} {
|
||||
out := &first
|
||||
if order == "DESC" {
|
||||
out = &last
|
||||
}
|
||||
res := gdb.WithContext(ctx).Model(model).Order(column + " " + order).Limit(1).Take(out)
|
||||
if res.Error != nil && !errors.Is(res.Error, gorm.ErrRecordNotFound) {
|
||||
return time.Time{}, time.Time{}, fmt.Errorf("query migration range %s: %w", column, res.Error)
|
||||
}
|
||||
if res.Error == nil {
|
||||
found = true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
return time.Time{}, time.Time{}, nil
|
||||
}
|
||||
return timeOf(&first).UTC(), timeOf(&last).UTC(), nil
|
||||
}
|
||||
|
||||
func limitOr(v, def int) int {
|
||||
if v <= 0 {
|
||||
return def
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
func offsetOf(page, pageSize int) int {
|
||||
if page < 1 {
|
||||
page = 1
|
||||
}
|
||||
return (page - 1) * limitOr(pageSize, defaultPageSize)
|
||||
}
|
||||
|
||||
func countToUint64(v int64) uint64 {
|
||||
if v < 0 {
|
||||
return 0
|
||||
}
|
||||
return uint64(v)
|
||||
}
|
||||
|
||||
func isPostgresDialect(db *gorm.DB) bool {
|
||||
return db != nil && db.Dialector != nil && db.Name() == "postgres"
|
||||
}
|
||||
|
||||
func dailyTrendDateSQL(db *gorm.DB) string {
|
||||
if isPostgresDialect(db) {
|
||||
return "to_char(created_at, 'YYYY-MM-DD')"
|
||||
}
|
||||
return "strftime('%Y-%m-%d', created_at)"
|
||||
}
|
||||
|
||||
func isMissingRelation(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
msg := strings.ToLower(err.Error())
|
||||
return strings.Contains(msg, "no such table") || strings.Contains(msg, "does not exist")
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"Wavelet/pkg/idgen"
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/glebarez/sqlite"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func newTestUserAccessStore(t *testing.T) *userAccessLogGormStore {
|
||||
t.Helper()
|
||||
_ = idgen.Init(1)
|
||||
gdb, err := gorm.Open(sqlite.Open("file:logstore-"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, gdb.AutoMigrate(&UserAccessLog{}))
|
||||
return newUserAccessLogGormStore(gdb)
|
||||
}
|
||||
|
||||
func TestGormUserAccessLogCountList(t *testing.T) {
|
||||
ua := newTestUserAccessStore(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC().Truncate(time.Second)
|
||||
require.NoError(t, ua.BatchInsert(ctx, []UserAccessLog{
|
||||
{UserID: 10, Path: "/api/v1/users", Method: "GET", Status: 200, CreatedAt: now},
|
||||
{UserID: 20, Path: "/api/v1/admin", Method: "GET", Status: 200, CreatedAt: now},
|
||||
{UserID: 10, Path: "/api/v1/other", Method: "POST", Status: 201, CreatedAt: now},
|
||||
}))
|
||||
|
||||
count, err := ua.Count(ctx, AccessLogFilter{UserIDs: []uint64{10}, Path: "users"})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, uint64(1), count)
|
||||
|
||||
rows, total, err := ua.List(ctx, AccessLogFilter{UserIDs: []uint64{10}, Path: "users"}, 1, 10)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, uint64(1), total)
|
||||
require.Len(t, rows, 1)
|
||||
require.Equal(t, "/api/v1/users", rows[0].Path)
|
||||
require.NotZero(t, rows[0].ID)
|
||||
}
|
||||
|
||||
func TestGormUserAccessLogFreeze(t *testing.T) {
|
||||
ua := newTestUserAccessStore(t)
|
||||
SetConfigReader(func(_ context.Context, key string) (string, error) {
|
||||
if key == logMigrationKey {
|
||||
return "migrating", nil
|
||||
}
|
||||
return "", nil
|
||||
})
|
||||
t.Cleanup(ResetForTest)
|
||||
|
||||
err := ua.BatchInsert(context.Background(), []UserAccessLog{{UserID: 1, CreatedAt: time.Now()}})
|
||||
require.ErrorIs(t, err, ErrMigrating)
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"os/exec"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// plugins/domain 不得直接 import 已废弃的 internal。
|
||||
var forbiddenImports = []string{
|
||||
"Wavelet/internal",
|
||||
"Wavelet/internal",
|
||||
}
|
||||
|
||||
func TestAppsMustNotImportPersistenceDirectly(t *testing.T) {
|
||||
t.Chdir("../../../..")
|
||||
out, err := exec.Command("go", "list", "-test", "-f", `{{.ImportPath}} {{join .Imports " "}}`, "./plugins/domain/...").Output()
|
||||
if err != nil {
|
||||
t.Fatalf("go list: %v", err)
|
||||
}
|
||||
for _, line := range strings.Split(string(out), "\n") {
|
||||
fields := strings.Fields(line)
|
||||
if len(fields) == 0 {
|
||||
continue
|
||||
}
|
||||
pkg := fields[0]
|
||||
if !strings.HasPrefix(pkg, "Wavelet/plugins/domain") {
|
||||
continue
|
||||
}
|
||||
for _, imp := range fields[1:] {
|
||||
for _, forbidden := range forbiddenImports {
|
||||
if imp == forbidden {
|
||||
t.Errorf("%s must not import forbidden package %s", pkg, forbidden)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
// Package logstore abstracts user access-log storage across ClickHouse, PostgreSQL and SQLite.
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ErrMigrating 表示日志数据库正在迁移，当前禁止写入。
|
||||
var ErrMigrating = errors.New("log database is migrating, writes are disabled")
|
||||
|
||||
// UserAccessLogStore 用户访问日志（w_user_access_logs）。
|
||||
type UserAccessLogStore interface {
|
||||
BatchInsert(ctx context.Context, logs []UserAccessLog) error
|
||||
DeleteAll(ctx context.Context) (int64, error)
|
||||
DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error)
|
||||
Count(ctx context.Context, filter AccessLogFilter) (uint64, error)
|
||||
List(ctx context.Context, filter AccessLogFilter, page, pageSize int) ([]UserAccessLog, uint64, error)
|
||||
GetDailyTrend(ctx context.Context, days int) ([]DailyTrend, error)
|
||||
GetBrowserDistribution(ctx context.Context, startTime time.Time) ([]BrowserShare, error)
|
||||
GetTopActiveUsers(ctx context.Context, startTime time.Time, limit int) ([]TopUser, error)
|
||||
ListForMigration(ctx context.Context, afterID uint64, limit int) ([]UserAccessLog, error)
|
||||
MigrationRange(ctx context.Context) (from, to time.Time, err error)
|
||||
EnsurePartitions(ctx context.Context, from, to time.Time) error
|
||||
// DropEmptyPartitions 幂等清理 PG 空分区表：删除 before 月份之前、且无任何数据的按月分区；
|
||||
// CH/SQLite 为 no-op。
|
||||
DropEmptyPartitions(ctx context.Context, before time.Time) error
|
||||
// DropExpiredPartitions 直接删除完全过期的 PG 整月分区（候选为月份早于 cutoff 月的分区，
|
||||
// 删除前校验分区内无保留期内数据，避免时区偏移下误删；迁移冻结期间拒绝执行）；CH/SQLite 为 no-op。
|
||||
DropExpiredPartitions(ctx context.Context, cutoff time.Time) error
|
||||
}
|
||||
|
||||
// StatusStore 日志库状态。
|
||||
type StatusStore interface {
|
||||
ActiveDatabase(ctx context.Context) (string, error)
|
||||
}
|
||||
|
||||
// Store 当前生效日志库。
|
||||
type Store struct {
|
||||
UserAccessLogs UserAccessLogStore
|
||||
Status StatusStore
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
-- +goose Up
|
||||
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;
|
||||
|
||||
-- +goose Down
|
||||
DROP TABLE IF EXISTS w_user_access_logs;
|
||||
@@ -0,0 +1,24 @@
|
||||
-- +goose Up
|
||||
-- +goose StatementBegin
|
||||
CREATE TABLE IF NOT EXISTS w_user_access_logs (
|
||||
id BIGINT NOT NULL,
|
||||
user_id BIGINT NOT NULL DEFAULT 0,
|
||||
path VARCHAR(2048) NOT NULL DEFAULT '',
|
||||
method VARCHAR(16) NOT NULL DEFAULT '',
|
||||
ip VARCHAR(128) NOT NULL DEFAULT '',
|
||||
user_agent TEXT NOT NULL DEFAULT '',
|
||||
headers TEXT NOT NULL DEFAULT '',
|
||||
status INTEGER NOT NULL DEFAULT 0,
|
||||
latency BIGINT NOT NULL DEFAULT 0,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
PRIMARY KEY (id, created_at)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_w_user_access_logs_user_id ON w_user_access_logs (user_id, created_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS idx_w_user_access_logs_created_at ON w_user_access_logs (created_at DESC);
|
||||
-- +goose StatementEnd
|
||||
|
||||
-- +goose Down
|
||||
-- +goose StatementBegin
|
||||
DROP TABLE IF EXISTS w_user_access_logs;
|
||||
-- +goose StatementEnd
|
||||
@@ -0,0 +1,24 @@
|
||||
-- +goose Up
|
||||
-- +goose StatementBegin
|
||||
CREATE TABLE IF NOT EXISTS w_user_access_logs (
|
||||
id BIGINT NOT NULL,
|
||||
user_id BIGINT NOT NULL DEFAULT 0,
|
||||
path VARCHAR(2048) NOT NULL DEFAULT '',
|
||||
method VARCHAR(16) NOT NULL DEFAULT '',
|
||||
ip VARCHAR(128) NOT NULL DEFAULT '',
|
||||
user_agent TEXT NOT NULL DEFAULT '',
|
||||
headers TEXT NOT NULL DEFAULT '',
|
||||
status INTEGER NOT NULL DEFAULT 0,
|
||||
latency BIGINT NOT NULL DEFAULT 0,
|
||||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
PRIMARY KEY (id, created_at)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_w_user_access_logs_user_id ON w_user_access_logs (user_id, created_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS idx_w_user_access_logs_created_at ON w_user_access_logs (created_at DESC);
|
||||
-- +goose StatementEnd
|
||||
|
||||
-- +goose Down
|
||||
-- +goose StatementBegin
|
||||
DROP TABLE IF EXISTS w_user_access_logs;
|
||||
-- +goose StatementEnd
|
||||
@@ -0,0 +1,44 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
// Package logstore abstracts user access-log storage across ClickHouse, PostgreSQL and SQLite.
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
userAccessLogTableName = "w_user_access_logs"
|
||||
userAccessLogInsertColumns = "id, user_id, path, method, ip, user_agent, headers, status, latency, created_at"
|
||||
)
|
||||
|
||||
// UserAccessLog stores HTTP access records in ClickHouse, PostgreSQL and SQLite.
|
||||
type UserAccessLog struct {
|
||||
ID uint64 `gorm:"column:id"`
|
||||
UserID uint64 `gorm:"column:user_id"`
|
||||
Path string `gorm:"column:path"`
|
||||
Method string `gorm:"column:method"`
|
||||
IP string `gorm:"column:ip"`
|
||||
UserAgent string `gorm:"column:user_agent"`
|
||||
Headers string `gorm:"column:headers"`
|
||||
Status int32 `gorm:"column:status"`
|
||||
Latency int64 `gorm:"column:latency"`
|
||||
CreatedAt time.Time `gorm:"column:created_at"`
|
||||
}
|
||||
|
||||
// TableName returns the table name.
|
||||
func (UserAccessLog) TableName() string {
|
||||
return userAccessLogTableName
|
||||
}
|
||||
|
||||
// InsertColumns returns comma-separated column names for batch insert.
|
||||
func (UserAccessLog) InsertColumns() string {
|
||||
return userAccessLogInsertColumns
|
||||
}
|
||||
|
||||
// BatchInsertSQL returns the INSERT prefix used by native batch writers.
|
||||
func (UserAccessLog) BatchInsertSQL() string {
|
||||
return fmt.Sprintf("INSERT INTO %s (%s)", userAccessLogTableName, userAccessLogInsertColumns)
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// listPartitionNames 列出 table 在当前 schema 下的全部直接分区表名（pg_inherits）。
|
||||
func listPartitionNames(ctx context.Context, gdb *gorm.DB, table string) ([]string, error) {
|
||||
var names []string
|
||||
if err := gdb.WithContext(ctx).Raw(`
|
||||
SELECT c.relname
|
||||
FROM pg_inherits i
|
||||
JOIN pg_class c ON c.oid = i.inhrelid
|
||||
JOIN pg_class p ON p.oid = i.inhparent
|
||||
JOIN pg_namespace n ON n.oid = p.relnamespace AND n.nspname = current_schema()
|
||||
WHERE p.relname = ?`, table).Scan(&names).Error; err != nil {
|
||||
return nil, fmt.Errorf("list partitions of %s: %w", table, err)
|
||||
}
|
||||
return names, nil
|
||||
}
|
||||
|
||||
// partitionNameMonth 解析按月分区表名 <table>_YYYYMM 的所属月份；命名不匹配返回 (零值, false)。
|
||||
func partitionNameMonth(table, name string) (time.Time, bool) {
|
||||
suffix, ok := strings.CutPrefix(name, table+"_")
|
||||
if !ok || len(suffix) != 6 {
|
||||
return time.Time{}, false
|
||||
}
|
||||
m, err := time.Parse("200601", suffix)
|
||||
if err != nil {
|
||||
return time.Time{}, false
|
||||
}
|
||||
return m, true
|
||||
}
|
||||
|
||||
// dropEligiblePartitionNames 返回 before 月份之前、命名合法的分区表名（是否为空由调用方校验）。
|
||||
func dropEligiblePartitionNames(table string, names []string, before time.Time) []string {
|
||||
beforeMonth := time.Date(before.Year(), before.Month(), 1, 0, 0, 0, 0, time.UTC)
|
||||
out := make([]string, 0, len(names))
|
||||
for _, name := range names {
|
||||
month, ok := partitionNameMonth(table, name)
|
||||
if !ok || !month.Before(beforeMonth) {
|
||||
continue
|
||||
}
|
||||
out = append(out, name)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// DropEmptyPartitions 幂等清理 PG 空分区表：仅删除 before 月份之前、且无任何数据的分区。
|
||||
// 非 PG 方言为 no-op。
|
||||
func (s *gormLogStore) DropEmptyPartitions(ctx context.Context, before time.Time) error {
|
||||
if !isPostgresDialect(s.db) {
|
||||
return nil
|
||||
}
|
||||
names, err := listPartitionNames(ctx, s.db, userAccessLogTable)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, name := range dropEligiblePartitionNames(userAccessLogTable, names, before) {
|
||||
var one int
|
||||
if err := s.db.WithContext(ctx).Raw("SELECT 1 FROM " + name + " LIMIT 1").Scan(&one).Error; err != nil {
|
||||
return fmt.Errorf("check partition %s empty: %w", name, err)
|
||||
}
|
||||
if one == 1 {
|
||||
continue
|
||||
}
|
||||
if err := s.db.WithContext(ctx).Exec("DROP TABLE IF EXISTS " + name).Error; err != nil {
|
||||
return fmt.Errorf("drop empty partition %s: %w", name, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// DropExpiredPartitions 直接删除完全过期的 PG 整月分区（避免 retention 清理逐行 DELETE）。
|
||||
// 候选 = 月份早于 cutoff 月（按 cutoff 的 UTC 时刻取月）；删除前校验分区内不存在 created_at >= cutoff 的行。
|
||||
// 迁移冻结期间返回 ErrMigrating。CH/SQLite 为 no-op。
|
||||
func (s *gormLogStore) DropExpiredPartitions(ctx context.Context, cutoff time.Time) error {
|
||||
if !isPostgresDialect(s.db) {
|
||||
return nil
|
||||
}
|
||||
if err := s.ensureWritable(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
names, err := listPartitionNames(ctx, s.db, userAccessLogTable)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cu := cutoff.UTC()
|
||||
cutoffMonth := time.Date(cu.Year(), cu.Month(), 1, 0, 0, 0, 0, time.UTC)
|
||||
for _, name := range names {
|
||||
month, ok := partitionNameMonth(userAccessLogTable, name)
|
||||
if !ok || !month.Before(cutoffMonth) {
|
||||
continue
|
||||
}
|
||||
var hasRetained int
|
||||
if err := s.db.WithContext(ctx).Raw("SELECT 1 FROM "+name+" WHERE created_at >= ? LIMIT 1", cu).Scan(&hasRetained).Error; err != nil {
|
||||
return fmt.Errorf("check partition %s retained rows: %w", name, err)
|
||||
}
|
||||
if hasRetained == 1 {
|
||||
continue
|
||||
}
|
||||
if err := s.db.WithContext(ctx).Exec("DROP TABLE IF EXISTS " + name).Error; err != nil {
|
||||
return fmt.Errorf("drop expired partition %s: %w", name, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestPartitionNameMonth(t *testing.T) {
|
||||
cases := []struct {
|
||||
table string
|
||||
name string
|
||||
want string
|
||||
}{
|
||||
{"w_user_access_logs", "w_user_access_logs_202612", "2026-12"},
|
||||
{"w_user_access_logs", "w_user_access_logs_202608", "2026-08"},
|
||||
{"w_user_access_logs", "of_node_access_logs_202608", ""},
|
||||
{"w_user_access_logs", "w_user_access_logs_20268", ""},
|
||||
{"w_user_access_logs", "w_user_access_logs_202613", ""},
|
||||
{"w_user_access_logs", "w_user_access_logs_default", ""},
|
||||
}
|
||||
for _, c := range cases {
|
||||
got, ok := partitionNameMonth(c.table, c.name)
|
||||
if c.want == "" {
|
||||
if ok {
|
||||
t.Fatalf("partitionNameMonth(%q, %q) ok = true, want false", c.table, c.name)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if !ok || got.Format("2006-01") != c.want {
|
||||
t.Fatalf("partitionNameMonth(%q, %q) = %v, want %s", c.table, c.name, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDropEligiblePartitionNames(t *testing.T) {
|
||||
before := time.Date(2026, 10, 15, 0, 0, 0, 0, time.UTC)
|
||||
names := []string{
|
||||
"w_user_access_logs_202608",
|
||||
"w_user_access_logs_202609",
|
||||
"w_user_access_logs_202610",
|
||||
"w_user_access_logs_202611",
|
||||
"w_user_access_logs_default",
|
||||
}
|
||||
got := dropEligiblePartitionNames(userAccessLogTable, names, before)
|
||||
want := []string{"w_user_access_logs_202608", "w_user_access_logs_202609"}
|
||||
require.Equal(t, want, got)
|
||||
|
||||
first := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
|
||||
require.Empty(t, dropEligiblePartitionNames(userAccessLogTable, []string{"w_user_access_logs_202610"}, first))
|
||||
}
|
||||
|
||||
func TestDropPartitionHelpersSQLiteNoop(t *testing.T) {
|
||||
ua := newTestUserAccessStore(t)
|
||||
ctx := context.Background()
|
||||
require.NoError(t, ua.BatchInsert(ctx, []UserAccessLog{
|
||||
{UserID: 1, Path: "/x", CreatedAt: time.Now().UTC()},
|
||||
}))
|
||||
require.NoError(t, ua.DropExpiredPartitions(ctx, time.Now().AddDate(0, 0, -90)))
|
||||
require.NoError(t, ua.DropEmptyPartitions(ctx, time.Now()))
|
||||
count, err := ua.Count(ctx, AccessLogFilter{})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, uint64(1), count)
|
||||
}
|
||||
@@ -0,0 +1,201 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package logstore
|
||||
|
||||
import (
|
||||
"Wavelet/pkg/logger"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
logDatabaseKey = "log_database"
|
||||
logMigrationKey = "log_db_migration"
|
||||
)
|
||||
|
||||
const (
|
||||
dbNamePostgres = "postgres"
|
||||
dbNameSQLite = "sqlite"
|
||||
dbNameClickHouse = "clickhouse"
|
||||
)
|
||||
|
||||
var errConfigReaderNotWired = errors.New("logstore: config reader not wired")
|
||||
|
||||
// ConfigReader 读取系统配置字符串值，由 bootstrap 注入（避免 logstore ↔ repository 循环依赖）。
|
||||
type ConfigReader func(ctx context.Context, key string) (string, error)
|
||||
|
||||
const resolveCacheTTL = 1 * time.Second
|
||||
|
||||
var (
|
||||
configReader ConfigReader
|
||||
|
||||
defaultDBMu sync.RWMutex
|
||||
defaultDB = dbNameSQLite
|
||||
|
||||
storeMu sync.RWMutex
|
||||
active *Store
|
||||
activeDB string
|
||||
lastResolveDB string
|
||||
lastResolveTime time.Time
|
||||
)
|
||||
|
||||
// SetDefaultDatabases configures the fallback database based on database and clickhouse enablement.
|
||||
func SetDefaultDatabases(dbEnabled, chEnabled bool) {
|
||||
defaultDBMu.Lock()
|
||||
defer defaultDBMu.Unlock()
|
||||
defaultDB = dbNameSQLite
|
||||
if dbEnabled {
|
||||
defaultDB = dbNamePostgres
|
||||
}
|
||||
if chEnabled {
|
||||
defaultDB = dbNameClickHouse
|
||||
}
|
||||
}
|
||||
|
||||
func getDefaultDatabase() string {
|
||||
defaultDBMu.RLock()
|
||||
defer defaultDBMu.RUnlock()
|
||||
return defaultDB
|
||||
}
|
||||
|
||||
// SetConfigReader 注入系统配置读取函数（bootstrap 调用，测试可注入内存实现）。
|
||||
func SetConfigReader(fn ConfigReader) { configReader = fn }
|
||||
|
||||
func getConfig(ctx context.Context, key string) (string, error) {
|
||||
if configReader == nil {
|
||||
return "", errConfigReaderNotWired
|
||||
}
|
||||
return configReader(ctx, key)
|
||||
}
|
||||
|
||||
// Active 返回当前生效的日志库 Store。
|
||||
func Active(ctx context.Context) (*Store, error) {
|
||||
current, err := resolveDatabase(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
storeMu.RLock()
|
||||
if active != nil && activeDB == current {
|
||||
s := active
|
||||
storeMu.RUnlock()
|
||||
return s, nil
|
||||
}
|
||||
storeMu.RUnlock()
|
||||
|
||||
storeMu.Lock()
|
||||
defer storeMu.Unlock()
|
||||
if active != nil && activeDB == current {
|
||||
return active, nil
|
||||
}
|
||||
s, err := buildStore(ctx, current, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
active = s
|
||||
activeDB = current
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// Build 直接按目标构造 store（不经 Active 缓存）。
|
||||
func Build(ctx context.Context, database string) (*Store, error) {
|
||||
return buildStore(ctx, database, false)
|
||||
}
|
||||
|
||||
// BuildForMigration 构造迁移目标 store，跳过冻结检查。
|
||||
func BuildForMigration(ctx context.Context, database string) (*Store, error) {
|
||||
return buildStore(ctx, database, true)
|
||||
}
|
||||
|
||||
func buildStore(ctx context.Context, database string, skipFreeze bool) (*Store, error) {
|
||||
switch database {
|
||||
case dbNameClickHouse:
|
||||
ual := newClickHouseUserAccessLogStore()
|
||||
ual.skipFreeze = skipFreeze
|
||||
return &Store{UserAccessLogs: ual, Status: ual}, nil
|
||||
case dbNamePostgres, dbNameSQLite:
|
||||
gdb := getDB(ctx)
|
||||
ual := newUserAccessLogGormStore(gdb)
|
||||
ual.skipFreeze = skipFreeze
|
||||
return &Store{UserAccessLogs: ual, Status: ual}, nil
|
||||
default:
|
||||
return nil, fmt.Errorf("unsupported log database: %s", database)
|
||||
}
|
||||
}
|
||||
|
||||
// Migrating 返回日志库是否处于迁移冻结状态。
|
||||
func Migrating(ctx context.Context) bool {
|
||||
v, err := getConfig(ctx, logMigrationKey)
|
||||
if err != nil {
|
||||
if !errors.Is(err, errConfigReaderNotWired) {
|
||||
logger.ErrorF(ctx, "read log migration config failed: %v", err)
|
||||
}
|
||||
return false
|
||||
}
|
||||
return v == "migrating"
|
||||
}
|
||||
|
||||
// Init 预热激活 store，并兜底预建当前月及未来分区。
|
||||
func Init(ctx context.Context) {
|
||||
s, err := Active(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
if err := s.UserAccessLogs.EnsurePartitions(ctx, now, now.AddDate(0, partitionLeadMonths, 0)); err != nil {
|
||||
logger.WarnF(ctx, "logstore: ensure startup partitions failed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// InvalidateCache 清空日志库解析缓存。
|
||||
func InvalidateCache() {
|
||||
storeMu.Lock()
|
||||
defer storeMu.Unlock()
|
||||
lastResolveTime = time.Time{}
|
||||
lastResolveDB = ""
|
||||
}
|
||||
|
||||
// ResetForTest 清空缓存的激活 store 与 config reader。
|
||||
func ResetForTest() {
|
||||
storeMu.Lock()
|
||||
active = nil
|
||||
activeDB = ""
|
||||
lastResolveDB = ""
|
||||
lastResolveTime = time.Time{}
|
||||
storeMu.Unlock()
|
||||
configReader = nil
|
||||
}
|
||||
|
||||
// ActiveDatabase 返回当前日志主库名。
|
||||
func ActiveDatabase(ctx context.Context) (string, error) {
|
||||
return resolveDatabase(ctx)
|
||||
}
|
||||
|
||||
func resolveDatabase(ctx context.Context) (string, error) {
|
||||
storeMu.RLock()
|
||||
if active != nil && time.Since(lastResolveTime) < resolveCacheTTL {
|
||||
name := lastResolveDB
|
||||
storeMu.RUnlock()
|
||||
return name, nil
|
||||
}
|
||||
storeMu.RUnlock()
|
||||
|
||||
v, err := getConfig(ctx, logDatabaseKey)
|
||||
if err != nil && !errors.Is(err, errConfigReaderNotWired) {
|
||||
return "", err
|
||||
}
|
||||
|
||||
resolved := v
|
||||
if resolved == "" {
|
||||
resolved = getDefaultDatabase()
|
||||
}
|
||||
|
||||
storeMu.Lock()
|
||||
lastResolveDB = resolved
|
||||
lastResolveTime = time.Now()
|
||||
storeMu.Unlock()
|
||||
return resolved, nil
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
// Package risk_control provides the access control, IP rate limiting, and telemetry risk analysis domain plugin for Cordis.
|
||||
package risk_control
|
||||
|
||||
import (
|
||||
"Wavelet/core/contracts"
|
||||
"Wavelet/pkg/ginutil"
|
||||
"Wavelet/pkg/idgen"
|
||||
"Wavelet/pkg/response"
|
||||
"Wavelet/plugins/domain/risk_control/logstore"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
var accessLogEnabled atomic.Bool
|
||||
|
||||
// SetAccessLogEnabled enables or disables access log collection.
|
||||
func SetAccessLogEnabled(enabled bool) {
|
||||
accessLogEnabled.Store(enabled)
|
||||
}
|
||||
|
||||
// IsAccessLogEnabled reports whether access log collection is enabled.
|
||||
func IsAccessLogEnabled() bool {
|
||||
return accessLogEnabled.Load()
|
||||
}
|
||||
|
||||
// Middleware is an alias for RiskControlMiddleware.
|
||||
var Middleware = RiskControlMiddleware
|
||||
|
||||
// RiskControlMiddleware 全局日志采集中间件
|
||||
func RiskControlMiddleware() gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
// 如果未启用日志采集，直接放行
|
||||
if !IsAccessLogEnabled() {
|
||||
c.Next()
|
||||
return
|
||||
}
|
||||
|
||||
// 1. 限流背压检测（检测本地缓冲队列是否已满）
|
||||
if IsBufferFull() {
|
||||
response.AbortTooManyRequests(c, errSystemBusy)
|
||||
return
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
|
||||
// 2. 执行后续请求（穿过业务处理和认证中间件）
|
||||
c.Next()
|
||||
|
||||
// 3. 后置身份检查：仅记录通过认证的请求
|
||||
userObj, exists := ginutil.GetFromContext[*contracts.UserDTO](c, contracts.AuthUserObjKey)
|
||||
if !exists || userObj == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// 4. 计算耗时并异步推送到缓冲队列
|
||||
latency := time.Since(start).Milliseconds()
|
||||
|
||||
var headersStr string
|
||||
if c.Request.Header != nil {
|
||||
// 克隆 Header，避免污染原 HTTP 请求的 Header 对象
|
||||
clonedHeaders := make(http.Header)
|
||||
for k, v := range c.Request.Header {
|
||||
clonedHeaders[k] = v
|
||||
}
|
||||
clonedHeaders.Del("Cookie")
|
||||
|
||||
if headersBytes, err := json.Marshal(clonedHeaders); err == nil {
|
||||
headersStr = string(headersBytes)
|
||||
}
|
||||
}
|
||||
|
||||
const maxHTTPStatus = 999
|
||||
status := c.Writer.Status()
|
||||
if status < 0 {
|
||||
status = 0
|
||||
} else if status > maxHTTPStatus {
|
||||
status = maxHTTPStatus
|
||||
}
|
||||
|
||||
logItem := &logstore.UserAccessLog{
|
||||
ID: idgen.NextUint64ID(),
|
||||
UserID: userObj.ID, // 直接从 Context 获取已登录用户ID，避免数据库查询
|
||||
Path: c.Request.URL.Path,
|
||||
Method: c.Request.Method,
|
||||
IP: c.ClientIP(),
|
||||
UserAgent: c.Request.UserAgent(),
|
||||
Headers: headersStr,
|
||||
Status: int32(status),
|
||||
Latency: latency,
|
||||
CreatedAt: time.Now(),
|
||||
}
|
||||
|
||||
// 非阻塞地推入缓存队列
|
||||
QueueAccessLog(logItem)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,227 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package risk_control_test
|
||||
|
||||
import (
|
||||
"Wavelet/core/contracts"
|
||||
"Wavelet/pkg/batchwriter"
|
||||
"Wavelet/pkg/ginutil"
|
||||
"Wavelet/pkg/idgen"
|
||||
"Wavelet/pkg/testhelper"
|
||||
"Wavelet/plugins/domain/risk_control"
|
||||
"Wavelet/plugins/domain/risk_control/logstore"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func init() {
|
||||
_ = idgen.Init(1)
|
||||
}
|
||||
|
||||
func newTestAccessLogWriter(t *testing.T, cfg batchwriter.Config) (*batchwriter.Writer[*logstore.UserAccessLog], func() []*logstore.UserAccessLog) {
|
||||
t.Helper()
|
||||
|
||||
var (
|
||||
mu sync.Mutex
|
||||
captured []*logstore.UserAccessLog
|
||||
)
|
||||
writer, err := batchwriter.New(cfg, func(_ context.Context, items []*logstore.UserAccessLog) error {
|
||||
mu.Lock()
|
||||
captured = append(captured, items...)
|
||||
mu.Unlock()
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("batchwriter.New() error = %v", err)
|
||||
}
|
||||
|
||||
writer.Start(context.Background())
|
||||
restore := risk_control.SetLogWriterForTest(writer)
|
||||
t.Cleanup(func() {
|
||||
restore()
|
||||
stopCtx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
_ = writer.Stop(stopCtx)
|
||||
})
|
||||
|
||||
return writer, func() []*logstore.UserAccessLog {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
return append([]*logstore.UserAccessLog(nil), captured...)
|
||||
}
|
||||
}
|
||||
|
||||
func drainAccessLogWriter(t *testing.T, writer *batchwriter.Writer[*logstore.UserAccessLog]) {
|
||||
t.Helper()
|
||||
|
||||
stopCtx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
if err := writer.Stop(stopCtx); err != nil {
|
||||
t.Fatalf("writer.Stop() error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRiskControlMiddleware(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
t.Run("ClickHouse disabled", func(t *testing.T) {
|
||||
risk_control.SetAccessLogEnabled(false)
|
||||
defer risk_control.SetAccessLogEnabled(false)
|
||||
|
||||
r := testhelper.NewTestGinEngine(risk_control.RiskControlMiddleware())
|
||||
r.GET("/test", func(c *gin.Context) {
|
||||
c.String(http.StatusOK, "ok")
|
||||
})
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
req, _ := http.NewRequest(http.MethodGet, "/test", nil)
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
assert.Equal(t, http.StatusOK, w.Code)
|
||||
assert.Equal(t, "ok", w.Body.String())
|
||||
})
|
||||
|
||||
t.Run("ClickHouse enabled - Normal Authenticated Request", func(t *testing.T) {
|
||||
risk_control.SetAccessLogEnabled(true)
|
||||
defer risk_control.SetAccessLogEnabled(false)
|
||||
|
||||
cfg := batchwriter.DefaultConfig()
|
||||
cfg.MaxBatchSize = 100
|
||||
cfg.FlushInterval = time.Hour
|
||||
|
||||
writer, getCaptured := newTestAccessLogWriter(t, cfg)
|
||||
|
||||
r := gin.New()
|
||||
r.Use(func(c *gin.Context) {
|
||||
user := &contracts.UserDTO{ID: 12345}
|
||||
ginutil.SetToContext[*contracts.UserDTO](c, contracts.AuthUserObjKey, user)
|
||||
c.Next()
|
||||
})
|
||||
r.Use(risk_control.RiskControlMiddleware())
|
||||
r.GET("/test", func(c *gin.Context) {
|
||||
c.String(http.StatusOK, "ok")
|
||||
})
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
req, _ := http.NewRequest(http.MethodGet, "/test", nil)
|
||||
req.Header.Set("X-Test-Header", "hello")
|
||||
req.Header.Set("Cookie", "session_id=abcdef123456")
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
assert.Equal(t, http.StatusOK, w.Code)
|
||||
assert.Equal(t, "ok", w.Body.String())
|
||||
|
||||
drainAccessLogWriter(t, writer)
|
||||
|
||||
captured := getCaptured()
|
||||
if len(captured) != 1 {
|
||||
t.Fatalf("captured access logs = %d, want 1", len(captured))
|
||||
}
|
||||
logItem := captured[0]
|
||||
assert.Equal(t, uint64(12345), logItem.UserID)
|
||||
assert.Equal(t, "/test", logItem.Path)
|
||||
assert.Equal(t, http.MethodGet, logItem.Method)
|
||||
assert.Equal(t, int32(http.StatusOK), logItem.Status)
|
||||
assert.NotEmpty(t, logItem.Headers)
|
||||
assert.Contains(t, logItem.Headers, "X-Test-Header")
|
||||
assert.NotContains(t, logItem.Headers, "Cookie")
|
||||
})
|
||||
|
||||
t.Run("ClickHouse enabled - Unauthenticated Request", func(t *testing.T) {
|
||||
risk_control.SetAccessLogEnabled(true)
|
||||
defer risk_control.SetAccessLogEnabled(false)
|
||||
|
||||
cfg := batchwriter.DefaultConfig()
|
||||
cfg.MaxBatchSize = 100
|
||||
cfg.FlushInterval = time.Hour
|
||||
|
||||
writer, getCaptured := newTestAccessLogWriter(t, cfg)
|
||||
|
||||
r := testhelper.NewTestGinEngine(risk_control.RiskControlMiddleware())
|
||||
r.GET("/test", func(c *gin.Context) {
|
||||
c.String(http.StatusOK, "ok")
|
||||
})
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
req, _ := http.NewRequest(http.MethodGet, "/test", nil)
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
assert.Equal(t, http.StatusOK, w.Code)
|
||||
assert.Equal(t, "ok", w.Body.String())
|
||||
|
||||
drainAccessLogWriter(t, writer)
|
||||
|
||||
if len(getCaptured()) != 0 {
|
||||
t.Fatal("expected no log item for unauthenticated request")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("ClickHouse enabled - Buffer Full Rate Limiting", func(t *testing.T) {
|
||||
risk_control.SetAccessLogEnabled(true)
|
||||
defer risk_control.SetAccessLogEnabled(false)
|
||||
|
||||
cfg := batchwriter.DefaultConfig()
|
||||
cfg.QueueSize = 2
|
||||
cfg.MaxBatchSize = 1
|
||||
cfg.FlushInterval = time.Hour
|
||||
|
||||
blockCh := make(chan struct{})
|
||||
enteredCh := make(chan struct{})
|
||||
|
||||
writer, err := batchwriter.New(cfg, func(_ context.Context, items []*logstore.UserAccessLog) error {
|
||||
select {
|
||||
case enteredCh <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
<-blockCh
|
||||
return nil
|
||||
})
|
||||
assert.NoError(t, err)
|
||||
writer.Start(context.Background())
|
||||
restore := risk_control.SetLogWriterForTest(writer)
|
||||
defer func() {
|
||||
close(blockCh)
|
||||
restore()
|
||||
stopCtx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
_ = writer.Stop(stopCtx)
|
||||
}()
|
||||
|
||||
// 1. 推入 1 个 item，worker 立即取走并触发 flush()，阻塞在 <-blockCh
|
||||
writer.TryEnqueue(&logstore.UserAccessLog{})
|
||||
<-enteredCh
|
||||
|
||||
// 2. 此时 worker 卡在 flush()，无法从 channel 取数据，推入 2 个 item 填满 channel
|
||||
for range cfg.QueueSize {
|
||||
writer.TryEnqueue(&logstore.UserAccessLog{})
|
||||
}
|
||||
if !risk_control.IsBufferFull() {
|
||||
t.Fatal("IsBufferFull() = false, want true")
|
||||
}
|
||||
|
||||
r := testhelper.NewTestGinEngine(risk_control.RiskControlMiddleware())
|
||||
r.GET("/test", func(c *gin.Context) {
|
||||
c.String(http.StatusOK, "ok")
|
||||
})
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
req, _ := http.NewRequest(http.MethodGet, "/test", nil)
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
assert.Equal(t, http.StatusTooManyRequests, w.Code)
|
||||
|
||||
var resp map[string]interface{}
|
||||
err = json.Unmarshal(w.Body.Bytes(), &resp)
|
||||
assert.NoError(t, err)
|
||||
assert.Contains(t, resp["error_msg"], "系统繁忙")
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package risk_control
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
userAccessLogTableName = "w_user_access_logs"
|
||||
userAccessLogInsertColumns = "id, user_id, path, method, ip, user_agent, headers, status, latency, created_at"
|
||||
)
|
||||
|
||||
// UserAccessLog stores HTTP access records in ClickHouse/database.
|
||||
type UserAccessLog struct {
|
||||
ID uint64 `gorm:"column:id"`
|
||||
UserID uint64 `gorm:"column:user_id"`
|
||||
Path string `gorm:"column:path"`
|
||||
Method string `gorm:"column:method"`
|
||||
IP string `gorm:"column:ip"`
|
||||
UserAgent string `gorm:"column:user_agent"`
|
||||
Headers string `gorm:"column:headers"`
|
||||
Status int32 `gorm:"column:status"`
|
||||
Latency int64 `gorm:"column:latency"`
|
||||
CreatedAt time.Time `gorm:"column:created_at"`
|
||||
}
|
||||
|
||||
// TableName returns the table name.
|
||||
func (UserAccessLog) TableName() string {
|
||||
return userAccessLogTableName
|
||||
}
|
||||
|
||||
// InsertColumns returns comma-separated column names for batch insert.
|
||||
func (UserAccessLog) InsertColumns() string {
|
||||
return userAccessLogInsertColumns
|
||||
}
|
||||
|
||||
// BatchInsertSQL returns the INSERT prefix used by native batch writers.
|
||||
func (UserAccessLog) BatchInsertSQL() string {
|
||||
return fmt.Sprintf("INSERT INTO %s (%s)", userAccessLogTableName, userAccessLogInsertColumns)
|
||||
}
|
||||
@@ -0,0 +1,148 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
// Package risk_control provides the access control, IP rate limiting, and telemetry risk analysis domain plugin for Cordis.
|
||||
package risk_control
|
||||
|
||||
import (
|
||||
"Wavelet/core"
|
||||
"Wavelet/core/contracts"
|
||||
"Wavelet/core/extpoints"
|
||||
"Wavelet/plugins/domain/risk_control/logstore"
|
||||
"context"
|
||||
"embed"
|
||||
"reflect"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
//go:embed logstore/migrations/*/*.sql
|
||||
var riskControlMigrations embed.FS
|
||||
|
||||
// Option configures the risk_control plugin.
|
||||
type Option func(*Plugin)
|
||||
|
||||
// WithMiddleware configures a custom risk control middleware.
|
||||
func WithMiddleware(mw gin.HandlerFunc) Option {
|
||||
return func(p *Plugin) {
|
||||
p.middleware = mw
|
||||
}
|
||||
}
|
||||
|
||||
// Plugin implements core.Plugin to provide risk control and access logging middleware.
|
||||
type Plugin struct {
|
||||
middleware gin.HandlerFunc
|
||||
}
|
||||
|
||||
// New creates a new risk_control domain plugin.
|
||||
func New(opts ...Option) *Plugin {
|
||||
p := &Plugin{}
|
||||
for _, opt := range opts {
|
||||
if opt != nil {
|
||||
opt(p)
|
||||
}
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
// Name returns the unique identifier for the risk_control domain plugin.
|
||||
func (p *Plugin) Name() string {
|
||||
return "risk_control"
|
||||
}
|
||||
|
||||
// Inject declares required dependencies for the risk_control domain plugin.
|
||||
func (p *Plugin) Inject() []reflect.Type {
|
||||
return []reflect.Type{
|
||||
reflect.TypeFor[contracts.DBService](),
|
||||
reflect.TypeFor[contracts.CacheService](),
|
||||
}
|
||||
}
|
||||
|
||||
// Manifest returns the plugin metadata.
|
||||
func (p *Plugin) Manifest() core.Manifest {
|
||||
return core.Manifest{
|
||||
Name: "risk_control",
|
||||
Version: "1.0.0",
|
||||
Description: "Access control, IP rate limiting, and access log telemetry domain plugin",
|
||||
Author: "Wavelet Team",
|
||||
}
|
||||
}
|
||||
|
||||
type rcClickHouseConfig struct {
|
||||
Enabled bool `config:"enabled" env:"CLICKHOUSE_ENABLED" default:"false" autoEnable:"CLICKHOUSE_HOST"`
|
||||
}
|
||||
|
||||
type rcDBConfig struct {
|
||||
Enabled bool `config:"enabled" env:"DB_ENABLED" default:"false" autoEnable:"DB_HOST"`
|
||||
}
|
||||
|
||||
// DeclareConfig declares configuration bindings for the risk_control plugin.
|
||||
func (p *Plugin) DeclareConfig() []core.ConfigBinding {
|
||||
return []core.ConfigBinding{
|
||||
{Prefix: "clickhouse", Target: &rcClickHouseConfig{}},
|
||||
{Prefix: "database", Target: &rcDBConfig{}},
|
||||
}
|
||||
}
|
||||
|
||||
// Apply registers risk control middlewares, settings, and cleanup hooks into the Context.
|
||||
func (p *Plugin) Apply(ctx *core.Context) error {
|
||||
var chCfg rcClickHouseConfig
|
||||
_ = ctx.Config().Bind("clickhouse", &chCfg)
|
||||
var dbCfg rcDBConfig
|
||||
_ = ctx.Config().Bind("database", &dbCfg)
|
||||
|
||||
SetAccessLogEnabled(chCfg.Enabled)
|
||||
logstore.SetDefaultDatabases(dbCfg.Enabled, chCfg.Enabled)
|
||||
|
||||
// 0. Bind DBService
|
||||
if db, err := core.Inject[contracts.DBService](ctx); err == nil && db != nil {
|
||||
logstore.SetDBService(db)
|
||||
} else {
|
||||
core.When[contracts.DBService](ctx, func(db contracts.DBService) {
|
||||
logstore.SetDBService(db)
|
||||
})
|
||||
}
|
||||
ctx.OnDispose(func() error {
|
||||
logstore.SetDBService(nil)
|
||||
return nil
|
||||
})
|
||||
|
||||
// 0. Register user access log table migrations
|
||||
ctx.Migrations().Register("risk_control/logstore", riskControlMigrations)
|
||||
|
||||
// 1. Initialize LogWriter if needed
|
||||
InitLogWriter(ctx.GoContext())
|
||||
|
||||
// 2. Register router middleware
|
||||
mw := p.middleware
|
||||
if mw == nil {
|
||||
mw = RiskControlMiddleware()
|
||||
}
|
||||
ctx.Router().Use(mw)
|
||||
|
||||
// 3. Register Settings Schemas
|
||||
ctx.Settings().Register(extpoints.SettingSchema{
|
||||
Key: "risk_control.ip_rate_limit_per_minute",
|
||||
Default: 60,
|
||||
Description: "Maximum requests allowed per IP per minute",
|
||||
Type: "integer",
|
||||
Category: "security",
|
||||
})
|
||||
ctx.Settings().Register(extpoints.SettingSchema{
|
||||
Key: "risk_control.enable_access_log",
|
||||
Default: true,
|
||||
Description: "Enable structured access log auditing and backpressure queueing",
|
||||
Type: "boolean",
|
||||
Category: "security",
|
||||
})
|
||||
|
||||
// 4. Register RiskControlService contract
|
||||
core.Provide[contracts.RiskControlService](ctx, &riskControlServiceImpl{})
|
||||
|
||||
// 5. Register lifecycle disposal cleanup
|
||||
ctx.OnDispose(func() error {
|
||||
return StopLogWriter(context.Background())
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package risk_control_test
|
||||
|
||||
import (
|
||||
"Wavelet/core"
|
||||
"Wavelet/plugins/domain/risk_control"
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestRiskControlPluginUnit(t *testing.T) {
|
||||
ctx := core.NewContext(context.Background())
|
||||
var customMWCalled bool
|
||||
customMW := func(c *gin.Context) {
|
||||
customMWCalled = true
|
||||
c.Next()
|
||||
}
|
||||
|
||||
p := risk_control.New(risk_control.WithMiddleware(customMW))
|
||||
assert.Equal(t, "risk_control", p.Name())
|
||||
assert.Equal(t, "1.0.0", p.Manifest().Version)
|
||||
require.NoError(t, p.Apply(ctx))
|
||||
|
||||
// Verify middlewares registered
|
||||
mws := ctx.Router().Middlewares()
|
||||
assert.NotEmpty(t, mws)
|
||||
|
||||
// Verify settings
|
||||
setting, ok := ctx.Settings().Get("risk_control.enable_access_log")
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, true, setting.Default)
|
||||
|
||||
require.NoError(t, ctx.Dispose())
|
||||
assert.False(t, customMWCalled) // not dispatched via gin engine here
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package risk_control
|
||||
|
||||
import (
|
||||
"Wavelet/plugins/domain/risk_control/logstore"
|
||||
"context"
|
||||
)
|
||||
|
||||
// Repository 层：本插件根包内唯一的持久化访问入口。
|
||||
//
|
||||
// 真正的 SQL / 驱动实现位于 logstore 子包（受 logstore skill 约束的存储抽象），
|
||||
// 本文件负责解析当前生效日志库并转发读写、迁移与查询，使 service.go 只做用例编排。
|
||||
|
||||
// accessLogMigrationBatchSize 是日志引擎迁移时单批搬运的行数。
|
||||
const accessLogMigrationBatchSize = 1000
|
||||
|
||||
// writeAccessLogBatch 持久化批写缓冲队列中取出的一批访问日志。
|
||||
func writeAccessLogBatch(ctx context.Context, items []*logstore.UserAccessLog) error {
|
||||
rows := make([]logstore.UserAccessLog, 0, len(items))
|
||||
for _, item := range items {
|
||||
if item == nil {
|
||||
continue
|
||||
}
|
||||
rows = append(rows, *item)
|
||||
}
|
||||
store, err := logstore.Active(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return store.UserAccessLogs.BatchInsert(ctx, rows)
|
||||
}
|
||||
|
||||
// listAccessLogs 按过滤条件读取一页访问日志。
|
||||
func listAccessLogs(ctx context.Context, filter logstore.AccessLogFilter, page, pageSize int) ([]logstore.UserAccessLog, uint64, error) {
|
||||
store, err := logstore.Active(ctx)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return store.UserAccessLogs.List(ctx, filter, page, pageSize)
|
||||
}
|
||||
|
||||
// accessLogDailyTrend 读取最近 days 天的按天访问趋势。
|
||||
func accessLogDailyTrend(ctx context.Context, days int) ([]logstore.DailyTrend, error) {
|
||||
store, err := logstore.Active(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return store.UserAccessLogs.GetDailyTrend(ctx, days)
|
||||
}
|
||||
|
||||
// activeLogDatabase 返回当前生效日志库的引擎标识。
|
||||
func activeLogDatabase(ctx context.Context) (string, error) {
|
||||
store, err := logstore.Active(ctx)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return store.Status.ActiveDatabase(ctx)
|
||||
}
|
||||
|
||||
// logStoreMigrating 报告日志库是否处于迁移冻结期。
|
||||
func logStoreMigrating(ctx context.Context) bool {
|
||||
return logstore.Migrating(ctx)
|
||||
}
|
||||
|
||||
// loadMigrationStores 解析迁移源（当前生效库）与目标引擎库。
|
||||
func loadMigrationStores(ctx context.Context, targetEngine string) (src, dst *logstore.Store, err error) {
|
||||
src, err = logstore.Active(ctx)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
dst, err = logstore.BuildForMigration(ctx, targetEngine)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return src, dst, nil
|
||||
}
|
||||
|
||||
// copyAccessLogs 清空目标库、按源库时间范围预建分区后分批搬运全部源数据，
|
||||
// 并在每批完成后通过 reportProgress 回调累计已搬运行数。
|
||||
func copyAccessLogs(ctx context.Context, src, dst *logstore.Store, reportProgress func(copied int)) error {
|
||||
if _, err := dst.UserAccessLogs.DeleteAll(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
from, to, err := src.UserAccessLogs.MigrationRange(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !from.IsZero() && !to.IsZero() {
|
||||
if err := dst.UserAccessLogs.EnsurePartitions(ctx, from, to); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
var afterID uint64
|
||||
var copied int
|
||||
for {
|
||||
rows, err := src.UserAccessLogs.ListForMigration(ctx, afterID, accessLogMigrationBatchSize)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(rows) == 0 {
|
||||
break
|
||||
}
|
||||
if err := dst.UserAccessLogs.BatchInsert(ctx, rows); err != nil {
|
||||
return err
|
||||
}
|
||||
afterID = rows[len(rows)-1].ID
|
||||
copied += len(rows)
|
||||
if reportProgress != nil {
|
||||
reportProgress(copied)
|
||||
}
|
||||
if len(rows) < accessLogMigrationBatchSize {
|
||||
break
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// resetLogStoreCache 丢弃缓存的生效日志库，使下一次访问重新解析。
|
||||
func resetLogStoreCache() {
|
||||
logstore.InvalidateCache()
|
||||
}
|
||||
@@ -0,0 +1,213 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package risk_control
|
||||
|
||||
import (
|
||||
"Wavelet/core/contracts"
|
||||
"Wavelet/pkg/batchwriter"
|
||||
"Wavelet/pkg/logger"
|
||||
"Wavelet/plugins/domain/risk_control/logstore"
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// fallbackLogEngine 是日志库状态不可得时对外暴露的引擎标识。
|
||||
const fallbackLogEngine = "sqlite"
|
||||
|
||||
var (
|
||||
logWriterMu sync.RWMutex
|
||||
logWriter *batchwriter.Writer[*logstore.UserAccessLog]
|
||||
)
|
||||
|
||||
// InitLogWriter initializes the access-log batch writer for the active log database.
|
||||
func InitLogWriter(ctx context.Context) {
|
||||
logWriterMu.Lock()
|
||||
defer logWriterMu.Unlock()
|
||||
if logWriter != nil {
|
||||
return
|
||||
}
|
||||
|
||||
cfg := batchwriter.DefaultConfig()
|
||||
writer, err := batchwriter.New[*logstore.UserAccessLog](cfg, writeAccessLogBatch,
|
||||
batchwriter.WithDropHandler[*logstore.UserAccessLog](func(item *logstore.UserAccessLog) {
|
||||
path := ""
|
||||
if item != nil {
|
||||
path = item.Path
|
||||
}
|
||||
logger.WarnF(context.Background(), "[RiskControl] Log queue full, dropping log item for path: %s", path)
|
||||
}),
|
||||
batchwriter.WithFlushErrorHandler[*logstore.UserAccessLog](func(ctx context.Context, items []*logstore.UserAccessLog, err error) {
|
||||
logger.ErrorF(ctx, "[RiskControl] flush access-log batch failed (batch=%d): %v", len(items), err)
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
logger.ErrorF(ctx, "[RiskControl] init log writer failed: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
writer.Start(ctx)
|
||||
logWriter = writer
|
||||
}
|
||||
|
||||
// StopLogWriter stops the ClickHouse access-log batch writer and drains pending logs.
|
||||
func StopLogWriter(ctx context.Context) error {
|
||||
writer := currentLogWriter()
|
||||
if writer == nil {
|
||||
return nil
|
||||
}
|
||||
return writer.Stop(ctx)
|
||||
}
|
||||
|
||||
// IsBufferFull reports whether the access-log queue has no remaining capacity.
|
||||
func IsBufferFull() bool {
|
||||
writer := currentLogWriter()
|
||||
if writer == nil {
|
||||
return false
|
||||
}
|
||||
return writer.IsFull()
|
||||
}
|
||||
|
||||
// QueueAccessLog enqueues an access log without blocking.
|
||||
func QueueAccessLog(logItem *logstore.UserAccessLog) {
|
||||
writer := currentLogWriter()
|
||||
if writer == nil || logItem == nil {
|
||||
return
|
||||
}
|
||||
writer.TryEnqueue(logItem)
|
||||
}
|
||||
|
||||
// SetLogWriterForTest swaps the access-log writer for unit tests.
|
||||
func SetLogWriterForTest(writer *batchwriter.Writer[*logstore.UserAccessLog]) func() {
|
||||
logWriterMu.Lock()
|
||||
previous := logWriter
|
||||
logWriter = writer
|
||||
logWriterMu.Unlock()
|
||||
return func() {
|
||||
logWriterMu.Lock()
|
||||
logWriter = previous
|
||||
logWriterMu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
func currentLogWriter() *batchwriter.Writer[*logstore.UserAccessLog] {
|
||||
logWriterMu.RLock()
|
||||
defer logWriterMu.RUnlock()
|
||||
return logWriter
|
||||
}
|
||||
|
||||
const drainPollInterval = 50 * time.Millisecond
|
||||
|
||||
// Drain waits until the in-memory access-log queue has been empty for one flush interval.
|
||||
func Drain(ctx context.Context) error {
|
||||
writer := currentLogWriter()
|
||||
if writer == nil {
|
||||
return nil
|
||||
}
|
||||
quietPeriod := batchwriter.DefaultConfig().FlushInterval
|
||||
if quietPeriod <= 0 {
|
||||
quietPeriod = time.Second
|
||||
}
|
||||
ticker := time.NewTicker(drainPollInterval)
|
||||
defer ticker.Stop()
|
||||
var quietSince time.Time
|
||||
for {
|
||||
if writer.Len() == 0 {
|
||||
if quietSince.IsZero() {
|
||||
quietSince = time.Now()
|
||||
} else if time.Since(quietSince) >= quietPeriod {
|
||||
return nil
|
||||
}
|
||||
} else {
|
||||
quietSince = time.Time{}
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-ticker.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// MigrateAndSwitchEngine migrates access logs to target database and switches the active store.
|
||||
func MigrateAndSwitchEngine(ctx context.Context, targetEngine string, reportProgress func(copied int)) error {
|
||||
if err := Drain(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
src, dst, err := loadMigrationStores(ctx, targetEngine)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := copyAccessLogs(ctx, src, dst, reportProgress); err != nil {
|
||||
return err
|
||||
}
|
||||
resetLogStoreCache()
|
||||
return nil
|
||||
}
|
||||
|
||||
// riskControlServiceImpl implements contracts.RiskControlService by orchestrating
|
||||
// the repository layer and mapping persistence rows into contract DTOs.
|
||||
type riskControlServiceImpl struct{}
|
||||
|
||||
func (s *riskControlServiceImpl) QueryAccessLogs(ctx context.Context, filter contracts.AccessLogFilterDTO, page, pageSize int) ([]contracts.AccessLogDTO, uint64, error) {
|
||||
list, total, err := listAccessLogs(ctx, logstore.AccessLogFilter{
|
||||
UserIDs: filter.UserIDs,
|
||||
Path: filter.Path,
|
||||
StartTime: filter.StartTime,
|
||||
EndTime: filter.EndTime,
|
||||
}, page, pageSize)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
items := make([]contracts.AccessLogDTO, len(list))
|
||||
for i, item := range list {
|
||||
items[i] = contracts.AccessLogDTO{
|
||||
ID: item.ID,
|
||||
UserID: item.UserID,
|
||||
IP: item.IP,
|
||||
UserAgent: item.UserAgent,
|
||||
Method: item.Method,
|
||||
Path: item.Path,
|
||||
Status: item.Status,
|
||||
Latency: item.Latency,
|
||||
CreatedAt: item.CreatedAt,
|
||||
}
|
||||
}
|
||||
return items, total, nil
|
||||
}
|
||||
|
||||
func (s *riskControlServiceImpl) QueryAccessLogStats(ctx context.Context, days int) ([]contracts.AccessLogDailyStatsDTO, error) {
|
||||
trend, err := accessLogDailyTrend(ctx, days)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
res := make([]contracts.AccessLogDailyStatsDTO, len(trend))
|
||||
for i, t := range trend {
|
||||
res[i] = contracts.AccessLogDailyStatsDTO{
|
||||
Date: t.Date,
|
||||
PV: t.Count,
|
||||
}
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (s *riskControlServiceImpl) ActiveLogEngine(ctx context.Context) string {
|
||||
engine, err := activeLogDatabase(ctx)
|
||||
if err != nil {
|
||||
return fallbackLogEngine
|
||||
}
|
||||
return engine
|
||||
}
|
||||
|
||||
func (s *riskControlServiceImpl) IsLogEngineMigrating(ctx context.Context) bool {
|
||||
return logStoreMigrating(ctx)
|
||||
}
|
||||
|
||||
func (s *riskControlServiceImpl) Drain(ctx context.Context) error {
|
||||
return Drain(ctx)
|
||||
}
|
||||
|
||||
func (s *riskControlServiceImpl) SwitchLogEngine(ctx context.Context, targetEngine string) error {
|
||||
return MigrateAndSwitchEngine(ctx, targetEngine, nil)
|
||||
}
|
||||
Reference in New Issue
Block a user