From 32e04d5ea277d0b1bd865c059181843c92babf83 Mon Sep 17 00:00:00 2001 From: duorameng <2997944583@qq.com> Date: Tue, 28 Apr 2026 17:15:01 +0800 Subject: [PATCH] chore: buffer opt --- internal/services/tasks/executor_service.go | 3 + internal/services/tasks/tiny_log.go | 73 +++++++++++++-------- internal/utils/buffer.go | 15 ++++- 3 files changed, 64 insertions(+), 27 deletions(-) diff --git a/internal/services/tasks/executor_service.go b/internal/services/tasks/executor_service.go index e9322bb..b09e9ca 100644 --- a/internal/services/tasks/executor_service.go +++ b/internal/services/tasks/executor_service.go @@ -67,6 +67,9 @@ func NewExecutorService( settingsService SettingsService, envService EnvService, ) *ExecutorService { + // 0. 清理旧临时日志 + CleanupOrphanedTinyLogs() + es := &ExecutorService{ taskService: taskService, taskLogService: taskLogService, diff --git a/internal/services/tasks/tiny_log.go b/internal/services/tasks/tiny_log.go index a381058..fe9f271 100644 --- a/internal/services/tasks/tiny_log.go +++ b/internal/services/tasks/tiny_log.go @@ -7,10 +7,11 @@ import ( "fmt" "io" "os" + "path/filepath" "sync" - "unicode/utf8" "github.com/engigu/baihu-panel/internal/constant" + "github.com/engigu/baihu-panel/internal/logger" "github.com/engigu/baihu-panel/internal/utils" ) @@ -57,7 +58,7 @@ type TinyLog struct { path string writer *bufio.Writer subscribers []chan []byte - remainder []byte // Leftover bytes from previous write (partial multi-byte characters) + remainder []byte // Leftover bytes from previous write (partial lines) masks []string // Secrets to mask closed bool } @@ -90,7 +91,6 @@ func (l *TinyLog) Write(p []byte) (n int, err error) { return 0, os.ErrClosed } - // 1. 合并上次调用剩余的字节(可能是半个 UTF-8 字符) originalInputLen := len(p) payload := p if len(l.remainder) > 0 { @@ -98,42 +98,43 @@ func (l *TinyLog) Write(p []byte) (n int, err error) { l.remainder = nil } - // 2. 识别结尾不完整的 UTF-8 序列 - lastSafe := len(payload) - // UTF-8 字符最多 4 字节,检查最后几个字节 - for i := len(payload) - 1; i >= 0 && i >= len(payload)-4; i-- { - if utf8.RuneStart(payload[i]) { - if !utf8.FullRune(payload[i:]) { - // 发现末尾存在不完整字符 - lastSafe = i - l.remainder = make([]byte, len(payload)-i) - copy(l.remainder, payload[i:]) - } - break + // 1. 寻找最后一个换行符 + lastNewline := bytes.LastIndexByte(payload, '\n') + if lastNewline == -1 { + // 没有换行符,且如果长度超过 4KB,强制截断并输出,防止内存无限制增长 + if len(payload) > 4096 { + lastNewline = len(payload) - 1 + } else { + // 保留当前所有内容到下一轮 + l.remainder = payload + return originalInputLen, nil } } - // 如果整个负载都不完整且不超过一个 UTF-8 字符的最大长度, - // 则全部保留到下次调用 - if lastSafe == 0 && len(l.remainder) > 0 { - return originalInputLen, nil + // 2. 提取出完整的行 + completeBytes := payload[:lastNewline+1] + + // 3. 剥离并保存剩余的部分 + if lastNewline+1 < len(payload) { + l.remainder = make([]byte, len(payload)-(lastNewline+1)) + copy(l.remainder, payload[lastNewline+1:]) } - // 3. 仅将完整的部分转换为 UTF-8,并调用封装的函数进行脱敏处理 - text := utils.MaskSecrets(utils.ToUTF8(payload[:lastSafe]), l.masks) - data := []byte(text) + // 4. 将完整行转换为 UTF-8 并脱敏 + text := utils.MaskSecrets(utils.ToUTF8(completeBytes), l.masks) + outData := []byte(text) - // 4. 写入文件缓冲区 - _, err = l.writer.Write(data) + // 5. 输出安全部分 + _, err = l.writer.Write(outData) if err != nil { return 0, err } - // 5. 广播给所有订阅者 + // 6. 广播给所有订阅者 if len(l.subscribers) > 0 { for _, ch := range l.subscribers { select { - case ch <- data: + case ch <- outData: default: // 如果订阅者处理太慢,丢弃消息以避免阻塞写入 } @@ -324,3 +325,23 @@ func (l *TinyLog) ReadLastLines(n int) ([]byte, error) { func (l *TinyLog) GetPath() string { return l.path } + +// CleanupOrphanedTinyLogs 启动时清理残留的临时日志文件 +func CleanupOrphanedTinyLogs() { + tmpDir := os.TempDir() + files, err := os.ReadDir(tmpDir) + if err != nil { + return + } + + count := 0 + for _, file := range files { + if !file.IsDir() && len(file.Name()) > 9 && file.Name()[:9] == "task_log_" && filepath.Ext(file.Name()) == ".log" { + os.Remove(filepath.Join(tmpDir, file.Name())) + count++ + } + } + if count > 0 { + logger.Infof("[System] 清理了 %d 个残留的任务日志临时文件", count) + } +} diff --git a/internal/utils/buffer.go b/internal/utils/buffer.go index 8b10e59..e1ad508 100644 --- a/internal/utils/buffer.go +++ b/internal/utils/buffer.go @@ -2,10 +2,12 @@ package utils import ( "fmt" + "sync" ) // TailBuffer 是一个只保留最后 N 字节数据的缓冲区 type TailBuffer struct { + mu sync.Mutex limit int data []byte size int // 当前实际存储的大小 @@ -22,6 +24,9 @@ func NewTailBuffer(limit int) *TailBuffer { // Write 实现 io.Writer 接口 func (b *TailBuffer) Write(p []byte) (n int, err error) { + b.mu.Lock() + defer b.mu.Unlock() + n = len(p) if n >= b.limit { // 如果单次写入就超过了限制,直接取最后 limit 字节 @@ -43,16 +48,24 @@ func (b *TailBuffer) Write(p []byte) (n int, err error) { // Bytes 返回缓冲区内的所有数据 func (b *TailBuffer) Bytes() []byte { - return b.data + b.mu.Lock() + defer b.mu.Unlock() + res := make([]byte, len(b.data)) + copy(res, b.data) + return res } // String 返回缓冲区内的字符串表示 func (b *TailBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() return string(b.data) } // Len 返回当前存储的数据长度 func (b *TailBuffer) Len() int { + b.mu.Lock() + defer b.mu.Unlock() return len(b.data) }