mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-09-30 06:16:37 +08:00
ryan
9491b2a744
Migrate risk_control audit logs to internal/db/batchwriter and add openflare/chwriter with per-table async flush for observability timeseries and node access logs. Replace single-row ClickHouse inserts and pre-insert SELECT count() dedup with repository BatchInsert* APIs, in-process TTL dedup for metric/report snapshots, and bootstrap initialization on API startup.
100 lines
2.6 KiB
Go
100 lines
2.6 KiB
Go
// Copyright 2026 Arctel.net
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
package risk_control
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
|
|
"github.com/Rain-kl/Wavelet/internal/config"
|
|
"github.com/Rain-kl/Wavelet/internal/db/batchwriter"
|
|
"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 (
|
|
logWriterMu sync.RWMutex
|
|
logWriter *batchwriter.Writer[*analytics.UserAccessLog]
|
|
)
|
|
|
|
// InitLogWriter initializes the ClickHouse access-log batch writer.
|
|
func InitLogWriter(ctx context.Context) {
|
|
if !config.Config.ClickHouse.Enabled {
|
|
return
|
|
}
|
|
|
|
logWriterMu.Lock()
|
|
defer logWriterMu.Unlock()
|
|
if logWriter != nil {
|
|
return
|
|
}
|
|
|
|
cfg := batchwriter.DefaultConfig()
|
|
writer, err := batchwriter.New[*analytics.UserAccessLog](cfg, func(ctx context.Context, items []*analytics.UserAccessLog) error {
|
|
rows := make([]analytics.UserAccessLog, 0, len(items))
|
|
for _, item := range items {
|
|
if item == nil {
|
|
continue
|
|
}
|
|
rows = append(rows, *item)
|
|
}
|
|
return analyticsrepo.BatchInsert(ctx, rows)
|
|
},
|
|
batchwriter.WithDropHandler[*analytics.UserAccessLog](func(item *analytics.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[*analytics.UserAccessLog](func(ctx context.Context, batchSize int, err error) {
|
|
logger.ErrorF(ctx, "[RiskControl] Send ClickHouse batch failed (batch=%d): %v", batchSize, err)
|
|
}),
|
|
)
|
|
if err != nil {
|
|
logger.ErrorF(ctx, "[RiskControl] init log writer failed: %v", err)
|
|
return
|
|
}
|
|
|
|
writer.Start(ctx)
|
|
logWriter = writer
|
|
}
|
|
|
|
// 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 *analytics.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[*analytics.UserAccessLog]) func() {
|
|
logWriterMu.Lock()
|
|
previous := logWriter
|
|
logWriter = writer
|
|
logWriterMu.Unlock()
|
|
return func() {
|
|
logWriterMu.Lock()
|
|
logWriter = previous
|
|
logWriterMu.Unlock()
|
|
}
|
|
}
|
|
|
|
func currentLogWriter() *batchwriter.Writer[*analytics.UserAccessLog] {
|
|
logWriterMu.RLock()
|
|
defer logWriterMu.RUnlock()
|
|
return logWriter
|
|
} |