tcjlj/OpenFlare
1
0
Fork 0
mirror of https://github.com/Rain-kl/OpenFlare.git synced 2026-10-08 00:26:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(clickhouse): integrate goose migrations and analytics repository

Browse Source 
Merge upstream Wavelet patch to manage ClickHouse DDL via goose
(goose_clickhouse_version), add GORM ChDB access, and centralize
w_user_access_logs reads/writes in internal/repository/analytics.
OpenFlare of_node_access_logs DDL moves to goose/clickhouse; remove
manual clickhouse_schema startup and support-files SQL duplicates.
This commit is contained in:
ryan
2026-06-19 12:11:41 +08:00
parent fff26aa343
commit 8a00f53b16
25 changed files with 1095 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files
openflare-server/internal/apps/admin/logs/routers.go
+112 -246
View File
@@ -10,14 +10,14 @@ import (
"encoding/json"
"fmt"
"net/http"
"sort"
"strconv"
"strings"
"time"
"github.com/Rain-kl/Wavelet/internal/apps/admin"
"github.com/Rain-kl/Wavelet/internal/config"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/model"
analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
"github.com/Rain-kl/Wavelet/pkg/logger"
"github.com/gin-gonic/gin"
@@ -30,7 +30,7 @@ const (
maxPageSize = 100
hoursInDay = 24
analyticsDays = 7
queryExtraArgs = 2 // pageSize + offset
topActiveLimit = 10
)
// logsResponse 历史日志查询响应
@@ -158,115 +158,75 @@ type accessLogsResponse struct {
List []accessLogItem `json:"list"`
}
// buildAccessLogFilters 构建 ClickHouse 访问日志查询过滤条件
func buildAccessLogFilters(ctx context.Context, c *gin.Context) ([]string, []interface{}, []uint64, error) {
var conditions []string
var args []interface{}
var userIDs []uint64
func buildAccessLogFilter(ctx context.Context, c *gin.Context) (analyticsrepo.AccessLogFilter, error) {
filter := analyticsrepo.AccessLogFilter{}
// 按用户名过滤
username := c.Query("username")
if username != "" {
var userIDs []uint64
err := db.DB(ctx).Model(&model.User{}).
Where("username LIKE ?", "%"+username+"%").
Pluck("id", &userIDs).Error
if err != nil {
return nil, nil, nil, fmt.Errorf("查询用户信息失败: %w", err)
return filter, fmt.Errorf("查询用户信息失败: %w", err)
}
if len(userIDs) == 0 {
return nil, nil, nil, nil // 无匹配用户
}
}
if len(userIDs) > 0 {
placeholders := make([]string, len(userIDs))
for i := range userIDs {
placeholders[i] = "?"
args = append(args, userIDs[i])
}
conditions = append(conditions, fmt.Sprintf("user_id IN (%s)", strings.Join(placeholders, ",")))
filter.UserIDs = userIDs
}
if path := c.Query("path"); path != "" {
conditions = append(conditions, "path LIKE ?")
args = append(args, "%"+path+"%")
filter.Path = path
}
if startTime := c.Query("start_time"); startTime != "" {
if t, err := time.Parse(time.RFC3339, startTime); err == nil {
conditions = append(conditions, "created_at >= ?")
args = append(args, t)
} else if t, err := time.Parse("2006-01-02 15:04:05", startTime); err == nil {
conditions = append(conditions, "created_at >= ?")
args = append(args, t)
if t, err := parseAccessLogTime(startTime); err == nil {
filter.StartTime = &t
}
}
if endTime := c.Query("end_time"); endTime != "" {
if t, err := time.Parse(time.RFC3339, endTime); err == nil {
conditions = append(conditions, "created_at <= ?")
args = append(args, t)
} else if t, err := time.Parse("2006-01-02 15:04:05", endTime); err == nil {
conditions = append(conditions, "created_at <= ?")
args = append(args, t)
if t, err := parseAccessLogTime(endTime); err == nil {
filter.EndTime = &t
}
}
return conditions, args, userIDs, nil
return filter, nil
}
// fetchAccessLogDetails 查询 ClickHouse 访问日志明细并填充用户名
func fetchAccessLogDetails(ctx context.Context, whereClause string, args []interface{}, pageSize int, offset int) ([]accessLogItem, error) {
dataQuery := fmt.Sprintf(`
SELECT id, user_id, path, method, ip, user_agent, headers, status, latency, created_at
FROM w_user_access_logs
%s
ORDER BY created_at DESC, id DESC
LIMIT ? OFFSET ?
`, whereClause)
selectArgs := make([]interface{}, len(args), len(args)+queryExtraArgs)
copy(selectArgs, args)
selectArgs = append(selectArgs, pageSize, offset)
rows, err := db.ChConn.Query(ctx, dataQuery, selectArgs...)
if err != nil {
return nil, fmt.Errorf("查询 ClickHouse 日志明细失败: %w", err)
func parseAccessLogTime(value string) (time.Time, error) {
if t, err := time.Parse(time.RFC3339, value); err == nil {
return t, nil
}
defer func() { _ = rows.Close() }()
return time.Parse("2006-01-02 15:04:05", value)
}
var list []accessLogItem
var fetchUserIDs []uint64
for rows.Next() {
var item accessLogItem
var createdAt time.Time
if err := rows.Scan(&item.ID, &item.UserID, &item.Path, &item.Method, &item.IP, &item.UserAgent, &item.Headers, &item.Status, &item.Latency, &createdAt); err != nil {
return nil, fmt.Errorf("读取 ClickHouse 结果失败: %w", err)
}
item.CreatedAt = createdAt.Format(time.RFC3339)
list = append(list, item)
fetchUserIDs = append(fetchUserIDs, item.UserID)
func enrichAccessLogsWithUsers(ctx context.Context, list []accessLogItem) {
if len(list) == 0 {
return
}
// 反查 Postgres 关联 Username 和 Nickname
if len(fetchUserIDs) > 0 {
userMap := make(map[uint64]struct{ Username, Nickname string })
var users []model.User
if err := db.DB(ctx).Where("id IN ?", fetchUserIDs).Find(&users).Error; err == nil {
for _, u := range users {
userMap[u.ID] = struct{ Username, Nickname string }{Username: u.Username, Nickname: u.Nickname}
}
}
for i := range list {
if info, ok := userMap[list[i].UserID]; ok {
list[i].Username = info.Username
list[i].Nickname = info.Nickname
}
userIDs := make([]uint64, 0, len(list))
seen := make(map[uint64]struct{}, len(list))
for _, item := range list {
if _, ok := seen[item.UserID]; ok {
continue
}
seen[item.UserID] = struct{}{}
userIDs = append(userIDs, item.UserID)
}
return list, nil
userMap := make(map[uint64]struct{ Username, Nickname string })
var users []model.User
if err := db.DB(ctx).Where("id IN ?", userIDs).Find(&users).Error; err == nil {
for _, u := range users {
userMap[u.ID] = struct{ Username, Nickname string }{Username: u.Username, Nickname: u.Nickname}
}
}
for i := range list {
if info, ok := userMap[list[i].UserID]; ok {
list[i].Username = info.Username
list[i].Nickname = info.Nickname
}
}
}
// GetAccessLogs 获取 ClickHouse 异步采集的访问日志
@@ -287,12 +247,12 @@ func fetchAccessLogDetails(ctx context.Context, whereClause string, args []inter
// @Failure 403 {object} response.Any "无管理员权限"
// @Router /api/v1/admin/logs/access [get]
func GetAccessLogs(c *gin.Context) {
if db.ChConn == nil {
response.AbortWithError(c, http.StatusInternalServerError, "ClickHouse 未初始化，无法检索访问日志")
ctx := c.Request.Context()
if !config.Config.ClickHouse.Enabled || db.ChDB(ctx) == nil {
response.AbortWithError(c, http.StatusBadRequest, "ClickHouse 存储服务未启用，无法检索访问日志")
return
}
// 2. 解析分页参数
page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
if page < 1 {
page = 1
@@ -304,29 +264,20 @@ func GetAccessLogs(c *gin.Context) {
if pageSize > maxPageSize {
pageSize = maxPageSize
}
offset := (page - 1) * pageSize
// 3. 构建过滤条件
conditions, args, userIDs, err := buildAccessLogFilters(c.Request.Context(), c)
filter, err := buildAccessLogFilter(ctx, c)
if err != nil {
response.AbortWithError(c, http.StatusInternalServerError, err.Error())
return
}
if userIDs != nil && len(userIDs) == 0 {
if filter.UserIDs != nil && len(filter.UserIDs) == 0 {
c.JSON(http.StatusOK, response.OK(accessLogsResponse{Total: 0, List: []accessLogItem{}}))
return
}
whereClause := ""
if len(conditions) > 0 {
whereClause = "WHERE " + strings.Join(conditions, " AND ")
}
// 4. 查询日志总数
var total uint64
countQuery := fmt.Sprintf("SELECT count() FROM w_user_access_logs %s", whereClause)
if err := db.ChConn.QueryRow(c.Request.Context(), countQuery, args...).Scan(&total); err != nil {
response.AbortWithError(c, http.StatusInternalServerError, "查询 ClickHouse 日志统计失败: "+err.Error())
logs, total, err := analyticsrepo.ListAccessLogs(ctx, filter, page, pageSize)
if err != nil {
response.AbortWithError(c, http.StatusInternalServerError, err.Error())
return
}
if total == 0 {
@@ -334,12 +285,22 @@ func GetAccessLogs(c *gin.Context) {
return
}
// 5. 分页查询明细数据
list, err := fetchAccessLogDetails(c.Request.Context(), whereClause, args, pageSize, offset)
if err != nil {
response.AbortWithError(c, http.StatusInternalServerError, err.Error())
return
list := make([]accessLogItem, len(logs))
for i, logItem := range logs {
list[i] = accessLogItem{
ID: logItem.ID,
UserID: logItem.UserID,
Path: logItem.Path,
Method: logItem.Method,
IP: logItem.IP,
UserAgent: logItem.UserAgent,
Headers: logItem.Headers,
Status: logItem.Status,
Latency: logItem.Latency,
CreatedAt: logItem.CreatedAt.Format(time.RFC3339),
}
}
enrichAccessLogsWithUsers(ctx, list)
c.JSON(http.StatusOK, response.OK(accessLogsResponse{
Total: total,
@@ -386,133 +347,61 @@ type logsAnalyticsResponse struct {
// @Failure 403 {object} response.Any "无管理员权限"
// @Router /api/v1/admin/logs/analytics [get]
func GetLogsAnalytics(c *gin.Context) {
if db.ChConn == nil {
response.AbortWithError(c, http.StatusInternalServerError, "ClickHouse 未初始化，无法获取分析数据")
ctx := c.Request.Context()
if !config.Config.ClickHouse.Enabled || db.ChDB(ctx) == nil {
response.AbortWithError(c, http.StatusBadRequest, "ClickHouse 存储服务未启用，无法获取分析数据")
return
}
ctx := c.Request.Context()
// 7 天前 00:00:00
startTime := time.Now().AddDate(0, 0, -(analyticsDays - 1)).Truncate(hoursInDay * time.Hour)
trendList := queryAccessTrend(ctx, startTime)
browserList := queryBrowserDistribution(ctx, startTime)
topUsers := queryTopActiveUsers(ctx, startTime)
c.JSON(http.StatusOK, response.OK(logsAnalyticsResponse{
Trend: trendList,
Browsers: browserList,
TopUsers: topUsers,
}))
}
// queryAccessTrend 查询最近 7 天的访问趋势
func queryAccessTrend(ctx context.Context, startTime time.Time) []trendItem {
trendRows, err := db.ChConn.Query(ctx, `
SELECT toDate(created_at) as date, count() as count
FROM w_user_access_logs
WHERE created_at >= ?
GROUP BY date
ORDER BY date ASC
`, startTime)
trendMap := make(map[string]uint64)
for i := 0; i < analyticsDays; i++ {
dStr := time.Now().AddDate(0, 0, -i).Format("2006-01-02")
trendMap[dStr] = 0
trendPoints, err := analyticsrepo.GetDailyTrend(ctx, analyticsDays)
if err != nil {
response.AbortWithError(c, http.StatusInternalServerError, "查询访问趋势失败: "+err.Error())
return
}
if err == nil {
defer func() { _ = trendRows.Close() }()
for trendRows.Next() {
var dt time.Time
var cnt uint64
if errScan := trendRows.Scan(&dt, &cnt); errScan == nil {
dStr := dt.Format("2006-01-02")
trendMap[dStr] = cnt
}
trendList := make([]trendItem, len(trendPoints))
for i, point := range trendPoints {
trendList[i] = trendItem{
Date: point.Date,
Count: point.Count,
}
}
var trendList []trendItem
for i := analyticsDays - 1; i >= 0; i-- {
dStr := time.Now().AddDate(0, 0, -i).Format("2006-01-02")
trendList = append(trendList, trendItem{
Date: dStr,
Count: trendMap[dStr],
})
browserPoints, err := analyticsrepo.GetBrowserDistribution(ctx, startTime)
if err != nil {
response.AbortWithError(c, http.StatusInternalServerError, "查询浏览器分布失败: "+err.Error())
return
}
return trendList
}
// queryBrowserDistribution 查询浏览器分布排行
func queryBrowserDistribution(ctx context.Context, startTime time.Time) []browserItem {
uaRows, err := db.ChConn.Query(ctx, `
SELECT user_agent, count() as count
FROM w_user_access_logs
WHERE created_at >= ?
GROUP BY user_agent
`, startTime)
browserCounts := make(map[string]uint64)
if err == nil {
defer func() { _ = uaRows.Close() }()
for uaRows.Next() {
var ua string
var cnt uint64
if errScan := uaRows.Scan(&ua, &cnt); errScan == nil {
browser := parseBrowserName(ua)
browserCounts[browser] += cnt
}
browserList := make([]browserItem, len(browserPoints))
for i, point := range browserPoints {
browserList[i] = browserItem{
Browser: point.Browser,
Count: point.Count,
}
}
var browserList []browserItem
for b, cnt := range browserCounts {
browserList = append(browserList, browserItem{
Browser: b,
Count: cnt,
})
topUserPoints, err := analyticsrepo.GetTopActiveUsers(ctx, startTime, topActiveLimit)
if err != nil {
response.AbortWithError(c, http.StatusInternalServerError, "查询活跃用户失败: "+err.Error())
return
}
sort.Slice(browserList, func(i, j int) bool {
return browserList[i].Count > browserList[j].Count
})
return browserList
}
// queryTopActiveUsers 查询活跃用户 Top 10
func queryTopActiveUsers(ctx context.Context, startTime time.Time) []topUserItem {
userRows, err := db.ChConn.Query(ctx, `
SELECT user_id, count() as count
FROM w_user_access_logs
WHERE created_at >= ? AND user_id > 0
GROUP BY user_id
ORDER BY count DESC
LIMIT 10
`, startTime)
var topUsers []topUserItem
var userIDs []uint64
userCountMap := make(map[uint64]uint64)
if err == nil {
defer func() { _ = userRows.Close() }()
for userRows.Next() {
var uid uint64
var cnt uint64
if errScan := userRows.Scan(&uid, &cnt); errScan == nil {
userIDs = append(userIDs, uid)
userCountMap[uid] = cnt
}
topUsers := make([]topUserItem, len(topUserPoints))
userIDs := make([]uint64, len(topUserPoints))
for i, point := range topUserPoints {
topUsers[i] = topUserItem{
UserID: point.UserID,
Count: point.Count,
}
userIDs[i] = point.UserID
}
// 反查 Postgres 补全活跃用户的用户名和昵称
userProfileMap := make(map[uint64]struct {
Username string
Nickname string
})
if len(userIDs) > 0 {
userProfileMap := make(map[uint64]struct {
Username string
Nickname string
})
var users []model.User
if errProfile := db.DB(ctx).Where("id IN ?", userIDs).Find(&users).Error; errProfile == nil {
for _, u := range users {
@@ -525,40 +414,17 @@ func queryTopActiveUsers(ctx context.Context, startTime time.Time) []topUserItem
}
}
}
for i := range topUsers {
if profile, ok := userProfileMap[topUsers[i].UserID]; ok {
topUsers[i].Username = profile.Username
topUsers[i].Nickname = profile.Nickname
}
}
}
for _, uid := range userIDs {
profile := userProfileMap[uid]
topUsers = append(topUsers, topUserItem{
UserID: uid,
Username: profile.Username,
Nickname: profile.Nickname,
Count: userCountMap[uid],
})
}
return topUsers
}
// parseBrowserName 简易的 User-Agent 浏览器类型识别
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"
c.JSON(http.StatusOK, response.OK(logsAnalyticsResponse{
Trend: trendList,
Browsers: browserList,
TopUsers: topUsers,
}))
}
openflare-server/internal/apps/risk_control/logics.go
+11 -35
View File
@@ -8,11 +8,12 @@ import (
"time"
"github.com/Rain-kl/Wavelet/internal/config"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/model/analytics"
analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
"github.com/Rain-kl/Wavelet/pkg/logger"
)
var logChan chan *UserAccessLog
var logChan chan *analytics.UserAccessLog
const (
defaultQueueSize = 10000
@@ -26,7 +27,7 @@ func InitLogWriter(ctx context.Context) {
return
}
logChan = make(chan *UserAccessLog, defaultQueueSize)
logChan = make(chan *analytics.UserAccessLog, defaultQueueSize)
go startBatchWorker(context.WithoutCancel(ctx))
}
@@ -40,7 +41,7 @@ func IsBufferFull() bool {
}
// QueueAccessLog 异步非阻塞地将日志推入缓冲队列
func QueueAccessLog(logItem *UserAccessLog) {
func QueueAccessLog(logItem *analytics.UserAccessLog) {
if !config.Config.ClickHouse.Enabled || logChan == nil {
return
}
@@ -56,43 +57,18 @@ func startBatchWorker(ctx context.Context) {
ticker := time.NewTicker(flushInterval)
defer ticker.Stop()
var batch []*UserAccessLog
var batch []*analytics.UserAccessLog
flush := func() {
if len(batch) == 0 {
return
}
if db.ChConn == nil {
batch = nil
return
}
b, err := db.ChConn.PrepareBatch(ctx, "INSERT INTO w_user_access_logs (id, user_id, path, method, ip, user_agent, headers, status, latency, created_at)")
if err != nil {
logger.ErrorF(ctx, "[RiskControl] Prepare ClickHouse batch failed: %v", err)
batch = nil
return
items := make([]analytics.UserAccessLog, len(batch))
for i, item := range batch {
items[i] = *item
}
for _, item := range batch {
err = b.Append(
item.ID,
item.UserID,
item.Path,
item.Method,
item.IP,
item.UserAgent,
item.Headers,
item.Status,
item.Latency,
item.CreatedAt,
)
if err != nil {
logger.ErrorF(ctx, "[RiskControl] Append item to ClickHouse batch failed: %v", err)
}
}
if err := b.Send(); err != nil {
if err := analyticsrepo.BatchInsert(ctx, items); err != nil {
logger.ErrorF(ctx, "[RiskControl] Send ClickHouse batch failed: %v", err)
}
batch = nil
@@ -113,4 +89,4 @@ func startBatchWorker(ctx context.Context) {
flush()
}
}
}
}
openflare-server/internal/apps/risk_control/middleware.go
+2 -1
View File
@@ -14,6 +14,7 @@ import (
"github.com/Rain-kl/Wavelet/internal/config"
"github.com/Rain-kl/Wavelet/internal/db/idgen"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/model/analytics"
"github.com/gin-gonic/gin"
)
@@ -68,7 +69,7 @@ func RiskControlMiddleware() gin.HandlerFunc {
status = maxHTTPStatus
}
logItem := &UserAccessLog{
logItem := &analytics.UserAccessLog{
ID: idgen.NextUint64ID(),
UserID: userObj.ID, // 直接从 Context 获取已登录用户ID，避免数据库查询
Path: c.Request.URL.Path,
openflare-server/internal/apps/risk_control/middleware_test.go
+5 -4
View File
@@ -13,6 +13,7 @@ import (
"github.com/Rain-kl/Wavelet/internal/apps/oauth"
"github.com/Rain-kl/Wavelet/internal/config"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/model/analytics"
"github.com/Rain-kl/Wavelet/internal/testhelper"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/assert"
@@ -40,7 +41,7 @@ func TestRiskControlMiddleware(t *testing.T) {
t.Run("ClickHouse enabled - Normal Authenticated Request", func(t *testing.T) {
config.Config.ClickHouse.Enabled = true
logChan = make(chan *UserAccessLog, defaultQueueSize)
logChan = make(chan *analytics.UserAccessLog, defaultQueueSize)
defer func() {
config.Config.ClickHouse.Enabled = false
logChan = nil
@@ -84,7 +85,7 @@ func TestRiskControlMiddleware(t *testing.T) {
t.Run("ClickHouse enabled - Unauthenticated Request", func(t *testing.T) {
config.Config.ClickHouse.Enabled = true
logChan = make(chan *UserAccessLog, defaultQueueSize)
logChan = make(chan *analytics.UserAccessLog, defaultQueueSize)
defer func() {
config.Config.ClickHouse.Enabled = false
logChan = nil
@@ -113,7 +114,7 @@ func TestRiskControlMiddleware(t *testing.T) {
t.Run("ClickHouse enabled - Buffer Full Rate Limiting", func(t *testing.T) {
config.Config.ClickHouse.Enabled = true
logChan = make(chan *UserAccessLog, 2) // small capacity for quick fill
logChan = make(chan *analytics.UserAccessLog, 2) // small capacity for quick fill
defer func() {
config.Config.ClickHouse.Enabled = false
logChan = nil
@@ -121,7 +122,7 @@ func TestRiskControlMiddleware(t *testing.T) {
// fill logChan up to cap to simulate buffer full
for len(logChan) < cap(logChan) {
logChan <- &UserAccessLog{}
logChan <- &analytics.UserAccessLog{}
}
r := testhelper.NewTestGinEngine(RiskControlMiddleware())
openflare-server/internal/apps/risk_control/model.go
-22
View File
@@ -1,22 +0,0 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package risk_control
import (
"time"
)
// UserAccessLog 用户访问记录
type UserAccessLog struct {
ID uint64 `json:"id,string"`
UserID uint64 `json:"user_id,string"`
Path string `json:"path"`
Method string `json:"method"`
IP string `json:"ip"`
UserAgent string `json:"user_agent"`
Headers string `json:"headers"`
Status int32 `json:"status"`
Latency int64 `json:"latency"` // 耗时毫秒
CreatedAt time.Time `json:"created_at"`
}