334 lines
11 KiB
Go
334 lines
11 KiB
Go
package loggerx
|
||
|
||
import (
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"time"
|
||
)
|
||
|
||
// 异步队列容量，跟随实例
|
||
const asyncQueueSize = 1000
|
||
|
||
// Close 等待异步日志落盘的最长时间
|
||
const closeDrainTimeout = 5 * time.Second
|
||
|
||
// 写入，需要判断同步还是异步
|
||
func (l *Logger) write(event string, b []byte) (n int, err error) {
|
||
if l.toAsync(event, b) {
|
||
return len(b), nil
|
||
}
|
||
return l.store(event, b)
|
||
}
|
||
|
||
// 实际的存储
|
||
// 必须满足 io.Writer 契约：成功时 n == len(b)，否则调用方（log / io.MultiWriter）
|
||
// 会认为发生了短写并把日志吞掉
|
||
func (l *Logger) store(event string, b []byte) (n int, err error) {
|
||
return l.storeTo(l.channel, event, b)
|
||
}
|
||
|
||
// storeTo 按指定 channel 落盘（异步消费协程用，它的 l.channel 是空的）
|
||
func (l *Logger) storeTo(channel, event string, b []byte) (n int, err error) {
|
||
if l.option.isPrintFile {
|
||
// 串行化写入：句柄缓冲区不是并发安全的，
|
||
// 异步消费协程与同步调用可能同时写同一个句柄。
|
||
// 这里必须用 defer 解锁：一旦未来写入路径里出现 panic，
|
||
// 非 defer 的 Unlock 会被跳过，锁永久不释放，整个进程的日志全卡死
|
||
n, err = func() (int, error) {
|
||
l.writeMu.Lock()
|
||
defer l.writeMu.Unlock()
|
||
return l.storeFileTo(channel, event, b)
|
||
}()
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
}
|
||
|
||
// 驱动（控制台等）的写入失败不改变对调用方的契约，但要上报
|
||
if _, derr := l.writeDrivers(b); derr != nil {
|
||
l.reportError(fmt.Errorf("loggerx: 写入额外驱动失败: %w", derr))
|
||
}
|
||
|
||
return len(b), nil
|
||
}
|
||
|
||
// storeFile 写入日志文件
|
||
func (l *Logger) storeFile(event string, b []byte) (int, error) {
|
||
return l.storeFileTo(l.channel, event, b)
|
||
}
|
||
|
||
// storeFileTo 把日志写进指定 channel 对应的文件
|
||
func (l *Logger) storeFileTo(channel, event string, b []byte) (int, error) {
|
||
f, err := l.getFileTo(channel, event)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
|
||
// 按大小切割：这条日志会把当前文件写超上限时，先归档当前文件再换新文件
|
||
if limit := l.option.sizeSplit; limit > 0 && f.full(limit, len(b)) {
|
||
nf, rerr := l.rollFileTo(channel, event, f)
|
||
if rerr == nil {
|
||
f = nf
|
||
}
|
||
// 归档失败就继续往原文件写：宁可文件超一点，也不能把日志丢了
|
||
}
|
||
|
||
n, err := f.Write(b)
|
||
if err == nil && n < len(b) {
|
||
err = io.ErrShortWrite
|
||
}
|
||
if err == nil {
|
||
return n, nil
|
||
}
|
||
|
||
// 写入失败：丢弃这个句柄，落到磁盘后按最新文件名重开一次再写
|
||
// 只重试一次，避免原实现在短写时反复重开文件
|
||
l.discardFileTo(channel, event, f)
|
||
if nf, nerr := l.getFileTo(channel, event); nerr == nil {
|
||
if n2, err2 := nf.Write(b); err2 == nil && n2 == len(b) {
|
||
return n2, nil
|
||
}
|
||
}
|
||
|
||
return 0, err
|
||
}
|
||
|
||
// rollFileTo 归档当前文件并返回新文件句柄
|
||
//
|
||
// 任何失败路径都必须保证：filePath[key] 不会留下一个「已关闭」的句柄。
|
||
// 否则后续写入只会进那个句柄的内存缓冲，而它已不在句柄表里，
|
||
// 刷新和关闭都遍历不到 —— 数据静默丢失且接口返回成功
|
||
func (l *Logger) rollFileTo(channel, event string, f *logFile) (*logFile, error) {
|
||
key := fileKey{channel: channel, event: event}
|
||
|
||
// 先把缓冲清空再关句柄，保证归档内容完整
|
||
if err := f.Close(); err != nil {
|
||
l.repairHandle(key, f)
|
||
return nil, err
|
||
}
|
||
|
||
// 归档基名直接用句柄记下的 baseName：绝不去猜文件名尾部的 _N 是不是归档序号
|
||
// （按小时切割出来的名字本身就长这样：2026/09/13/06_info.log）
|
||
path, err := l.archive(f.baseName, f.fileName)
|
||
if err != nil {
|
||
l.repairHandle(key, f)
|
||
return nil, err
|
||
}
|
||
|
||
l.mu.Lock()
|
||
if cur, ok := l.filePath[key]; ok && cur == f {
|
||
delete(l.filePath, key)
|
||
}
|
||
l.mu.Unlock()
|
||
|
||
// 新文件带走递增序号，避免覆盖刚归档出去的同名文件
|
||
nf, err := l.openNumberedFile(key, l.nextIndex())
|
||
if err != nil {
|
||
l.repairHandle(key, nil)
|
||
return nil, err
|
||
}
|
||
|
||
l.mu.Lock()
|
||
l.filePath[key] = nf
|
||
l.mu.Unlock()
|
||
|
||
// 后台压缩，不阻塞写入
|
||
l.scheduleCompress(path)
|
||
return nf, nil
|
||
}
|
||
|
||
// repairHandle 给某个 key 装回一个「可写」的句柄
|
||
//
|
||
// 滚动失败时现场可能残留一个已关闭的句柄（还在表里或已被摘掉），
|
||
// 这里统一替换成一个新打开的同名文件句柄，保证后续写入有地方落盘；
|
||
// 实在打不开就把表项摘掉，让下一次写入重新走完整的打开流程
|
||
func (l *Logger) repairHandle(key fileKey, stale *logFile) {
|
||
l.mu.Lock()
|
||
if stale != nil {
|
||
if cur, ok := l.filePath[key]; ok && cur == stale {
|
||
delete(l.filePath, key)
|
||
}
|
||
} else {
|
||
delete(l.filePath, key)
|
||
}
|
||
l.mu.Unlock()
|
||
|
||
nf, err := l.openNewFile(key)
|
||
if err != nil {
|
||
l.reportError(fmt.Errorf("loggerx: 滚动失败后无法重新打开日志文件 %s/%s: %w", key.channel, key.event, err))
|
||
return
|
||
}
|
||
l.mu.Lock()
|
||
// 期间可能有别的 goroutine 已经装好了句柄，别覆盖
|
||
if _, ok := l.filePath[key]; !ok {
|
||
l.filePath[key] = nf
|
||
} else {
|
||
_ = nf.Close()
|
||
}
|
||
l.mu.Unlock()
|
||
}
|
||
|
||
// 写入额外的驱动（控制台 / 自定义 writer）
|
||
//
|
||
// 用户传进来的 driver 往往是 *bytes.Buffer / *bufio.Writer 这类非并发安全的对象，
|
||
// 所以这里必须和文件写入一样串行化；同时逐个写、单个失败不影响其它 driver
|
||
// （io.MultiWriter 会在第一个错误处短路，把后面的输出一起吞掉）
|
||
func (l *Logger) writeDrivers(b []byte) (int, error) {
|
||
if len(l.option.drivers) == 0 {
|
||
return 0, nil
|
||
}
|
||
|
||
l.driverMu.Lock()
|
||
defer l.driverMu.Unlock()
|
||
|
||
var errs []error
|
||
for _, d := range l.option.drivers {
|
||
if d == nil {
|
||
continue
|
||
}
|
||
if _, err := d.Write(b); err != nil {
|
||
errs = append(errs, err)
|
||
}
|
||
}
|
||
return len(b), joinErrors(errs)
|
||
}
|
||
|
||
// discardFile 关闭文件并从缓存中移除，使下次写入重新打开
|
||
func (l *Logger) discardFile(event string, f *logFile) {
|
||
l.discardFileTo(l.channel, event, f)
|
||
}
|
||
|
||
// discardFileTo 关闭并移除指定 channel 的文件句柄
|
||
func (l *Logger) discardFileTo(channel, event string, f *logFile) {
|
||
_ = f.Close()
|
||
|
||
l.mu.Lock()
|
||
defer l.mu.Unlock()
|
||
key := fileKey{channel: channel, event: event}
|
||
if cur, ok := l.filePath[key]; ok && cur == f {
|
||
delete(l.filePath, key)
|
||
}
|
||
}
|
||
|
||
// 异步队列的任务
|
||
// Channel 必须跟着任务走：消费协程持有的是「创建队列时的那个 logger」，
|
||
// 它的 channel 是空的。如果消费时用 logger 自己的 channel，
|
||
// 所有 channel 的日志都会落进根目录文件（曾经就是这么错的）
|
||
type cacheData struct {
|
||
Channel string
|
||
Event string
|
||
Data []byte
|
||
}
|
||
|
||
// toAsync 尝试异步写入
|
||
// 返回 true 表示已经交给异步队列，返回 false 表示需要同步写入
|
||
func (l *Logger) toAsync(event string, b []byte) bool {
|
||
if l.writeType == writeTypeSync || // 指定同步模式
|
||
(l.writeType == writeTypeDefault && l.option.writeType != writeTypeAsync) { // 默认同步模式
|
||
return false
|
||
}
|
||
|
||
// 整段「检查开关 + 入队」都在同一把锁内：Close 也拿这把锁来停止投递，
|
||
// 这样 Close 拿到锁时就能确定「要么这条还没入队、要么已经完整入队」。
|
||
// 否则消费协程可能先看到空队列就退出，把还在路上的这条日志整条丢掉。
|
||
//
|
||
// wg 用来告诉 drainAsync「投递临界区里还有人」：Close 必须等这些写入
|
||
// 真正入队之后才能关闭队列，否则向已关闭的 channel 发送会 panic。
|
||
// 队列满时这里会阻塞，但消费者是独立 goroutine 且不需要这把锁，不会死锁
|
||
l.async.mu.Lock()
|
||
if l.async.closed {
|
||
l.async.mu.Unlock()
|
||
// 已开始关闭：退化为同步写入
|
||
return false
|
||
}
|
||
if l.async.ch == nil {
|
||
l.async.ch = make(chan cacheData, asyncQueueSize)
|
||
go l.asyncWorker(l.async.ch)
|
||
}
|
||
l.async.begin()
|
||
ch := l.async.ch
|
||
l.async.mu.Unlock()
|
||
|
||
defer l.async.end() // 必须在解锁之后 Done，保证 begin/end 覆盖整段投递
|
||
|
||
// 必须复制一份再入队：b 可能是标准库 log 从 sync.Pool 借来的行缓冲，
|
||
// io.Writer 契约明确禁止保留传入的切片 —— log 在 Write 返回后立刻把
|
||
// 缓冲还池并复用，异步消费协程读到的就会是被改写的内存
|
||
// （曾经导致 1374/2000 条日志内容错乱）
|
||
data := make([]byte, len(b))
|
||
copy(data, b)
|
||
|
||
ch <- cacheData{Channel: l.channel, Event: event, Data: data}
|
||
return true
|
||
}
|
||
|
||
// asyncWorker 消费异步队列，直到队列被关闭
|
||
//
|
||
// 关键：单条任务 panic 绝不能让消费循环退出。
|
||
// 否则队列没人消费、channel 很快写满，之后所有写入方都会永久阻塞在
|
||
// `ch <- ...` 上（应用整体卡死），Close 也会永远等不到 workerDone。
|
||
func (l *Logger) asyncWorker(q chan cacheData) {
|
||
defer close(l.workerDone)
|
||
|
||
for val := range q {
|
||
l.consumeOne(val)
|
||
}
|
||
}
|
||
|
||
// consumeOne 处理一条异步任务，把 panic 限制在这一条之内
|
||
func (l *Logger) consumeOne(val cacheData) {
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
// 回调要包一层 recover：用户的错误处理函数自己也可能 panic
|
||
l.reportError(fmt.Errorf("loggerx: 异步写入单条日志时 panic: %v", r))
|
||
}
|
||
}()
|
||
// 按任务里记的 channel 落盘，避免多个 channel 的日志混到根目录
|
||
_, _ = l.storeTo(val.Channel, val.Event, val.Data)
|
||
}
|
||
|
||
// drainAsync 停止投递、关闭队列，并等消费协程把剩余任务全部写完
|
||
func (l *Logger) drainAsync() {
|
||
// 持锁关闭投递：此后 toAsync 一律退化为同步写入
|
||
l.async.mu.Lock()
|
||
l.async.closed = true
|
||
q := l.async.ch
|
||
l.async.mu.Unlock()
|
||
|
||
if q == nil {
|
||
// 从未启用过异步写入，没有后台 goroutine 需要等
|
||
return
|
||
}
|
||
|
||
// 等「已进入投递临界区」的写入全部入队，之后才能关队列。
|
||
//
|
||
// 这里刻意【不加超时】：wg 的非零计数正说明有 goroutine 卡在
|
||
// 临界区里（通常是队列满导致 `ch <-` 阻塞）。若超时后就 close(q)，
|
||
// 那些还停在 `ch <-` 上的生产者会被唤醒并 panic: send on closed channel，
|
||
// 而它们跑在应用自己的 goroutine 上（Info/Write 的调用方），
|
||
// 没有 recover，直接把进程打挂。
|
||
//
|
||
// 不设超时也不会死等：只有消费者倒下才会让队列永久满，
|
||
// 而消费者现在对每条任务单独 recover（见 consumeOne），不会死。
|
||
l.async.wg.Wait()
|
||
|
||
// 关闭队列并等消费者把缓冲里的任务全部处理完。
|
||
// 这一步必须真的等到 workerDone：否则 Close 会在消费协程还在写缓冲时
|
||
// 就刷盘并关闭文件，最后几条日志会连着句柄一起丢掉
|
||
close(q)
|
||
if !waitChanTimeout(l.workerDone, closeDrainTimeout) {
|
||
l.reportError(errors.New("loggerx: 等待异步日志落盘超时，队列中剩余日志可能丢失"))
|
||
}
|
||
}
|
||
|
||
// waitChanTimeout 等通道关闭，超时返回 false
|
||
func waitChanTimeout(ch <-chan struct{}, d time.Duration) bool {
|
||
select {
|
||
case <-ch:
|
||
return true
|
||
case <-time.After(d):
|
||
return false
|
||
}
|
||
}
|