mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-09-29 22:06:38 +08:00
ryan
00593742c3
Add a separate ClickHouse OLAP pipeline with goose/clickhouse DDL as the sole schema source, model/analytics for ORM mapping, and repository/analytics for reads (ChDB/GORM) and batch writes (ChConn). Refactor risk_control and admin/logs to use the repository layer instead of inline SQL. Remove the manual support-files DDL and document the workflow in database-migration skill.
92 lines
2.0 KiB
Go
92 lines
2.0 KiB
Go
// Copyright 2026 Arctel.net
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
package risk_control
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/Rain-kl/Wavelet/internal/config"
|
|
"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 *analytics.UserAccessLog
|
|
|
|
const (
|
|
defaultQueueSize = 10000
|
|
maxBatchSize = 1000
|
|
flushInterval = 1 * time.Second
|
|
)
|
|
|
|
// InitLogWriter 初始化日志写入通道和后台写入协程
|
|
func InitLogWriter(ctx context.Context) {
|
|
if !config.Config.ClickHouse.Enabled {
|
|
return
|
|
}
|
|
|
|
logChan = make(chan *analytics.UserAccessLog, defaultQueueSize)
|
|
go startBatchWorker(context.WithoutCancel(ctx))
|
|
}
|
|
|
|
// IsBufferFull 检查当前本地缓冲队列是否已满
|
|
// 如果没有启用 ClickHouse，默认返回 false，不触发限流
|
|
func IsBufferFull() bool {
|
|
if !config.Config.ClickHouse.Enabled || logChan == nil {
|
|
return false
|
|
}
|
|
return len(logChan) >= cap(logChan)
|
|
}
|
|
|
|
// QueueAccessLog 异步非阻塞地将日志推入缓冲队列
|
|
func QueueAccessLog(logItem *analytics.UserAccessLog) {
|
|
if !config.Config.ClickHouse.Enabled || logChan == nil {
|
|
return
|
|
}
|
|
|
|
select {
|
|
case logChan <- logItem:
|
|
default:
|
|
logger.WarnF(context.Background(), "[RiskControl] Log queue full, dropping log item for path: %s", logItem.Path)
|
|
}
|
|
}
|
|
|
|
func startBatchWorker(ctx context.Context) {
|
|
ticker := time.NewTicker(flushInterval)
|
|
defer ticker.Stop()
|
|
|
|
var batch []*analytics.UserAccessLog
|
|
|
|
flush := func() {
|
|
if len(batch) == 0 {
|
|
return
|
|
}
|
|
|
|
items := make([]analytics.UserAccessLog, len(batch))
|
|
for i, item := range batch {
|
|
items[i] = *item
|
|
}
|
|
if err := analyticsrepo.BatchInsert(ctx, items); err != nil {
|
|
logger.ErrorF(ctx, "[RiskControl] Send ClickHouse batch failed: %v", err)
|
|
}
|
|
batch = nil
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case item, ok := <-logChan:
|
|
if !ok {
|
|
flush()
|
|
return
|
|
}
|
|
batch = append(batch, item)
|
|
if len(batch) >= maxBatchSize {
|
|
flush()
|
|
}
|
|
case <-ticker.C:
|
|
flush()
|
|
}
|
|
}
|
|
} |