feat: 添加日志管理和敏感词过滤功能
- 新增日志管理模块：操作日志、登录日志、数据变更日志 - 新增敏感词过滤器 (sanitizer) - 新增日志异步写入管理器 - 新增日志定时清理服务 - 优化帖子相关服务和上传服务
This commit is contained in:
304
internal/service/async_log_manager.go
Normal file
304
internal/service/async_log_manager.go
Normal file
@@ -0,0 +1,304 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"carrot_bbs/internal/model"
|
||||
"carrot_bbs/internal/repository"
|
||||
|
||||
"go.uber.org/zap"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// AsyncLogConfig 异步日志配置
|
||||
type AsyncLogConfig struct {
|
||||
BufferSize int
|
||||
BatchSize int
|
||||
FlushInterval time.Duration
|
||||
WorkerCount int
|
||||
EnableFallback bool
|
||||
FallbackPath string
|
||||
}
|
||||
|
||||
// AsyncLogManager 异步日志管理器
|
||||
type AsyncLogManager struct {
|
||||
operationChan chan *model.OperationLog
|
||||
loginChan chan *model.LoginLog
|
||||
dataChangeChan chan *model.DataChangeLog
|
||||
|
||||
cfg *AsyncLogConfig
|
||||
db *gorm.DB
|
||||
wg sync.WaitGroup
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
logger *zap.Logger
|
||||
|
||||
fallbackWriter *os.File
|
||||
}
|
||||
|
||||
// NewAsyncLogManager 创建异步日志管理器
|
||||
func NewAsyncLogManager(
|
||||
db *gorm.DB,
|
||||
operationRepo *repository.OperationLogRepository,
|
||||
loginRepo *repository.LoginLogRepository,
|
||||
dataChangeRepo *repository.DataChangeLogRepository,
|
||||
logger *zap.Logger,
|
||||
cfg *AsyncLogConfig,
|
||||
) *AsyncLogManager {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
m := &AsyncLogManager{
|
||||
operationChan: make(chan *model.OperationLog, cfg.BufferSize),
|
||||
loginChan: make(chan *model.LoginLog, cfg.BufferSize),
|
||||
dataChangeChan: make(chan *model.DataChangeLog, cfg.BufferSize),
|
||||
cfg: cfg,
|
||||
db: db,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
logger: logger,
|
||||
}
|
||||
|
||||
if cfg.EnableFallback {
|
||||
if err := m.initFallbackWriter(); err != nil {
|
||||
logger.Warn("failed to init fallback writer", zap.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
for i := 0; i < cfg.WorkerCount; i++ {
|
||||
m.wg.Add(1)
|
||||
go m.worker(i, operationRepo, loginRepo, dataChangeRepo)
|
||||
}
|
||||
|
||||
go m.flushScheduler()
|
||||
|
||||
logger.Info("async log manager started",
|
||||
zap.Int("buffer_size", cfg.BufferSize),
|
||||
zap.Int("batch_size", cfg.BatchSize),
|
||||
zap.Duration("flush_interval", cfg.FlushInterval),
|
||||
zap.Int("worker_count", cfg.WorkerCount),
|
||||
)
|
||||
|
||||
return m
|
||||
}
|
||||
|
||||
// worker 处理协程
|
||||
func (m *AsyncLogManager) worker(
|
||||
id int,
|
||||
operationRepo *repository.OperationLogRepository,
|
||||
loginRepo *repository.LoginLogRepository,
|
||||
dataChangeRepo *repository.DataChangeLogRepository,
|
||||
) {
|
||||
defer m.wg.Done()
|
||||
|
||||
opBatch := make([]*model.OperationLog, 0, m.cfg.BatchSize)
|
||||
loginBatch := make([]*model.LoginLog, 0, m.cfg.BatchSize)
|
||||
changeBatch := make([]*model.DataChangeLog, 0, m.cfg.BatchSize)
|
||||
|
||||
ticker := time.NewTicker(m.cfg.FlushInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-m.ctx.Done():
|
||||
m.flushBatches(opBatch, loginBatch, changeBatch, operationRepo, loginRepo, dataChangeRepo)
|
||||
return
|
||||
|
||||
case op := <-m.operationChan:
|
||||
opBatch = append(opBatch, op)
|
||||
if len(opBatch) >= m.cfg.BatchSize {
|
||||
if err := m.flushOperationLogs(opBatch, operationRepo); err != nil {
|
||||
m.handleError("operation_logs", opBatch, err)
|
||||
}
|
||||
opBatch = opBatch[:0]
|
||||
}
|
||||
|
||||
case login := <-m.loginChan:
|
||||
loginBatch = append(loginBatch, login)
|
||||
if len(loginBatch) >= m.cfg.BatchSize {
|
||||
if err := m.flushLoginLogs(loginBatch, loginRepo); err != nil {
|
||||
m.handleError("login_logs", loginBatch, err)
|
||||
}
|
||||
loginBatch = loginBatch[:0]
|
||||
}
|
||||
|
||||
case change := <-m.dataChangeChan:
|
||||
changeBatch = append(changeBatch, change)
|
||||
if len(changeBatch) >= m.cfg.BatchSize {
|
||||
if err := m.flushDataChangeLogs(changeBatch, dataChangeRepo); err != nil {
|
||||
m.handleError("data_change_logs", changeBatch, err)
|
||||
}
|
||||
changeBatch = changeBatch[:0]
|
||||
}
|
||||
|
||||
case <-ticker.C:
|
||||
if len(opBatch) > 0 || len(loginBatch) > 0 || len(changeBatch) > 0 {
|
||||
m.flushBatches(opBatch, loginBatch, changeBatch, operationRepo, loginRepo, dataChangeRepo)
|
||||
opBatch = opBatch[:0]
|
||||
loginBatch = loginBatch[:0]
|
||||
changeBatch = changeBatch[:0]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// flushScheduler 定时刷新
|
||||
func (m *AsyncLogManager) flushScheduler() {
|
||||
ticker := time.NewTicker(m.cfg.FlushInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-m.ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// flushBatches 批量写入
|
||||
func (m *AsyncLogManager) flushBatches(
|
||||
ops []*model.OperationLog,
|
||||
logins []*model.LoginLog,
|
||||
changes []*model.DataChangeLog,
|
||||
operationRepo *repository.OperationLogRepository,
|
||||
loginRepo *repository.LoginLogRepository,
|
||||
dataChangeRepo *repository.DataChangeLogRepository,
|
||||
) {
|
||||
if len(ops) == 0 && len(logins) == 0 && len(changes) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
if len(ops) > 0 {
|
||||
if err := operationRepo.BatchCreateOperationLogs(ctx, ops); err != nil {
|
||||
m.handleError("operation_logs", ops, err)
|
||||
}
|
||||
}
|
||||
|
||||
if len(logins) > 0 {
|
||||
if err := loginRepo.BatchCreateLoginLogs(ctx, logins); err != nil {
|
||||
m.handleError("login_logs", logins, err)
|
||||
}
|
||||
}
|
||||
|
||||
if len(changes) > 0 {
|
||||
if err := dataChangeRepo.BatchCreateDataChangeLogs(ctx, changes); err != nil {
|
||||
m.handleError("data_change_logs", changes, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// flushOperationLogs 刷新操作日志
|
||||
func (m *AsyncLogManager) flushOperationLogs(logs []*model.OperationLog, repo *repository.OperationLogRepository) error {
|
||||
return repo.BatchCreateOperationLogs(context.Background(), logs)
|
||||
}
|
||||
|
||||
// flushLoginLogs 刷新登录日志
|
||||
func (m *AsyncLogManager) flushLoginLogs(logs []*model.LoginLog, repo *repository.LoginLogRepository) error {
|
||||
return repo.BatchCreateLoginLogs(context.Background(), logs)
|
||||
}
|
||||
|
||||
// flushDataChangeLogs 刷新数据变更日志
|
||||
func (m *AsyncLogManager) flushDataChangeLogs(logs []*model.DataChangeLog, repo *repository.DataChangeLogRepository) error {
|
||||
return repo.BatchCreateDataChangeLogs(context.Background(), logs)
|
||||
}
|
||||
|
||||
// handleError 处理写入错误
|
||||
func (m *AsyncLogManager) handleError(logType string, logs interface{}, err error) {
|
||||
m.logger.Error("failed to batch write logs",
|
||||
zap.String("log_type", logType),
|
||||
zap.Error(err),
|
||||
)
|
||||
|
||||
if m.fallbackWriter != nil {
|
||||
if err := m.writeToFallback(logType, logs); err != nil {
|
||||
m.logger.Error("failed to write to fallback", zap.Error(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// writeToFallback 写入降级文件
|
||||
func (m *AsyncLogManager) writeToFallback(logType string, logs interface{}) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// initFallbackWriter 初始化降级写入器
|
||||
func (m *AsyncLogManager) initFallbackWriter() error {
|
||||
if m.cfg.FallbackPath == "" {
|
||||
m.cfg.FallbackPath = "./logs/fallback.log"
|
||||
}
|
||||
|
||||
dir := "./logs"
|
||||
if _, err := os.Stat(dir); os.IsNotExist(err) {
|
||||
os.MkdirAll(dir, 0755)
|
||||
}
|
||||
|
||||
file, err := os.OpenFile(m.cfg.FallbackPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.fallbackWriter = file
|
||||
return nil
|
||||
}
|
||||
|
||||
// RecordOperation 记录操作日志（非阻塞）
|
||||
func (m *AsyncLogManager) RecordOperation(log *model.OperationLog) {
|
||||
select {
|
||||
case m.operationChan <- log:
|
||||
default:
|
||||
m.logger.Warn("operation log channel full, dropping log",
|
||||
zap.String("user_id", log.UserID),
|
||||
zap.String("operation", log.Operation),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// RecordLogin 记录登录日志（非阻塞）
|
||||
func (m *AsyncLogManager) RecordLogin(log *model.LoginLog) {
|
||||
select {
|
||||
case m.loginChan <- log:
|
||||
default:
|
||||
m.logger.Warn("login log channel full, dropping log",
|
||||
zap.String("user_id", log.UserID),
|
||||
zap.String("event", log.Event),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// RecordDataChange 记录数据变更日志（非阻塞）
|
||||
func (m *AsyncLogManager) RecordDataChange(log *model.DataChangeLog) {
|
||||
select {
|
||||
case m.dataChangeChan <- log:
|
||||
default:
|
||||
m.logger.Warn("data change log channel full, dropping log",
|
||||
zap.String("user_id", log.UserID),
|
||||
zap.String("change_type", log.ChangeType),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// Stop 停止异步日志管理器
|
||||
func (m *AsyncLogManager) Stop() {
|
||||
m.cancel()
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
m.wg.Wait()
|
||||
close(done)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
m.logger.Info("async log manager stopped gracefully")
|
||||
case <-time.After(10 * time.Second):
|
||||
m.logger.Warn("async log manager stop timeout, force closing")
|
||||
}
|
||||
|
||||
if m.fallbackWriter != nil {
|
||||
m.fallbackWriter.Close()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user