diff --git a/example/gin_server.go b/example/gin_server/main.go similarity index 83% rename from example/gin_server.go rename to example/gin_server/main.go index e8f2fd9..478efaa 100644 --- a/example/gin_server.go +++ b/example/gin_server/main.go @@ -8,11 +8,14 @@ import ( "github.com/yuninks/loggerx/middleware" ) +// 运行: go run ./example/gin_server // curl --location '127.0.0.1:8080/ping' func main() { ctx := context.Background() log := loggerx.NewLogger(ctx, loggerx.SetToConsole()) + // 进程退出前落盘并释放文件句柄 + defer log.Close() g := gin.Default() @@ -25,5 +28,4 @@ func main() { }) g.Run(":8080") - } diff --git a/example/main.go b/example/main.go index ad26f46..cd54158 100644 --- a/example/main.go +++ b/example/main.go @@ -17,6 +17,7 @@ func main() { loggerx.SetEscapeHTML(false), // loggerx.SetExpandData("ddd", "dddd"), ) + defer log.Close() log.WriteAsync().Info(ctx, "{ \"a\": 1, \"b\": 2 }") log.Info(ctx, "哈哈哈2") log.Info(ctx, "哈哈哈2") diff --git a/filePath.go b/filePath.go index 29f74fe..a14d423 100644 --- a/filePath.go +++ b/filePath.go @@ -1,7 +1,9 @@ package loggerx import ( + "errors" "fmt" + "io" "os" "path/filepath" "time" @@ -9,99 +11,200 @@ import ( // 文件操作 +// 每个文件的写缓冲大小：越大越省系统调用，代价是崩溃时可能丢最后一批日志 +const fileBufSize = 32 * 1024 + +// logFile 一个已打开的日志文件句柄 +// 自己管理写缓冲（不用 bufio）：bufio.Writer 在底层短写时会永久置位内部错误状态， +// 之后所有写入和 Flush 都会失败，已攒下的一批日志会整批丢掉 +type logFile struct { + file *os.File + buf []byte + pending int + fileName string +} + +// Write 写入缓冲，满了就落盘一次 +func (f *logFile) Write(b []byte) (int, error) { + written := 0 + for len(b) > 0 { + if f.pending == len(f.buf) { + if err := f.Flush(); err != nil { + return written, err + } + } + n := copy(f.buf[f.pending:], b) + f.pending += n + written += n + b = b[n:] + } + return written, nil +} + +// writeFull 保证把整块数据写进文件（允许底层短写，循环补齐） +func (f *logFile) writeFull(p []byte) error { + for len(p) > 0 { + n, err := f.file.Write(p) + p = p[n:] + if err != nil { + return err + } + if n == 0 { + return io.ErrShortWrite + } + } + return nil +} + +// Flush 把缓冲写入文件（不做 fsync） +func (f *logFile) Flush() error { + if f.pending == 0 { + return nil + } + if err := f.writeFull(f.buf[:f.pending]); err != nil { + // 保留未写成功的部分，下次继续 + return err + } + f.pending = 0 + return nil +} + +// Sync 把缓冲刷到文件并 fsync +func (f *logFile) Sync() error { + var errs []error + if err := f.Flush(); err != nil { + errs = append(errs, err) + } + if f.file != nil { + if err := f.file.Sync(); err != nil { + errs = append(errs, err) + } + } + return joinErrors(errs) +} + +// Close 落盘并关闭文件 +func (f *logFile) Close() error { + return joinErrors([]error{f.Sync(), f.file.Close()}) +} + +// joinErrors 汇总错误，Go 1.20+ 用 errors.Join +func joinErrors(errs []error) error { + var real []error + for _, err := range errs { + if err != nil { + real = append(real, err) + } + } + if len(real) == 0 { + return nil + } + return errors.Join(real...) +} + // 获取最新的文件名 func (l *Logger) nowFileName(event string) string { - // ioc, _ := time.LoadLocation("Asia/Shanghai") - // timeDir := fmt.Sprint(time.Now().In(ioc).Format("2006/01/02/15")) // 2006-01-02 15:04:05 - prefix := "" switch l.option.fileSplit { case FileSplitTimeA: // （年/月/日/时） - prefix = fmt.Sprint(time.Now().In(l.option.timeZone).Format("2006/01/02/15")) // 2006-01-02 15:04:05 + prefix = time.Now().In(l.option.timeZone).Format("2006/01/02/15") case FileSplitTimeB: // （年/月/日） - prefix = fmt.Sprint(time.Now().In(l.option.timeZone).Format("2006/01/02")) // 2006-01-02 15:04:05 + prefix = time.Now().In(l.option.timeZone).Format("2006/01/02") case FileSplitTimeC: // （年/月-日） - prefix = fmt.Sprint(time.Now().In(l.option.timeZone).Format("2006/01-02")) // 2006-01-02 15:04:05 + prefix = time.Now().In(l.option.timeZone).Format("2006/01-02") case FileSplitTimeD: // （年-月-日-时） - prefix = fmt.Sprint(time.Now().In(l.option.timeZone).Format("2006-01-02-15")) // 2006-01-02 15:04:05 + prefix = time.Now().In(l.option.timeZone).Format("2006-01-02-15") case FileSplitTimeE: // （年-月-日） - prefix = fmt.Sprint(time.Now().In(l.option.timeZone).Format("2006-01-02")) // 2006-01-02 15:04:05 + prefix = time.Now().In(l.option.timeZone).Format("2006-01-02") } if prefix != "" { prefix = prefix + "_" } - // timeDir := fmt.Sprint(time.Now().Local().Format("2006/01/02")) // 2006-01-02 15:04:05 if l.channel != "" { prefix = l.channel + "/" + prefix } - path := l.option.dir + "/" + prefix + event + ".log" - // fmt.Println(filepath.Abs(path)) - return path + return l.option.dir + "/" + prefix + event + ".log" } // 新建文件 -func (l *Logger) getFile(event string, isRefresh bool) (*os.File, error) { - f := l.loadFile(event) - if f != nil && !isRefresh { +func (l *Logger) getFile(event string) (*logFile, error) { + key := fileKey{channel: l.channel, event: event} + + if f := l.loadFile(key); f != nil { return f, nil } l.mu.Lock() defer l.mu.Unlock() + // 双检：可能在等锁期间已经被别的 goroutine 建好了 + if f := l.loadFileLocked(key); f != nil { + return f, nil + } + fileName := l.nowFileName(event) - dir, _ := filepath.Split(fileName) // 识别目录与文件 - os.MkdirAll(dir, os.ModePerm) // 创建多层目录，如果存在不会报错 + // 识别目录与文件，创建多层目录（已存在不报错） + if dir := filepath.Dir(fileName); dir != "" { + if err := os.MkdirAll(dir, 0755); err != nil { + return nil, err + } + } - // 打开该文件，如果不存在则创建 file, err := os.OpenFile(fileName, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644) if err != nil { - // 打开失败，尝试创建 - fmt.Println("打开日志文件失败") - return nil, err - } - // 关闭原来的 - if f != nil { - closeFile(f) + return nil, fmt.Errorf("打开日志文件失败 %s: %w", fileName, err) } - l.filePath.Store(l.channel, &filePath{ + lf := &logFile{ file: file, + buf: make([]byte, fileBufSize), fileName: fileName, - }) + } + l.filePath[key] = lf - return file, nil + return lf, nil } -// 加载文件 -func (l *Logger) loadFile(event string) *os.File { - val, ok := l.filePath.Load(l.channel) - if !ok { - return nil - } - f := val.(*filePath) - if f == nil { - return nil - } - if f.fileName != l.nowFileName(event) { - // 原来的文件需关闭 - closeFile(f.file) - return nil - } - return f.file +// 加载文件（读锁） +func (l *Logger) loadFile(key fileKey) *logFile { + l.mu.RLock() + defer l.mu.RUnlock() + return l.loadFileLocked(key) } -// 关闭文件 -func closeFile(f *os.File) error { - f.Sync() - return f.Close() +// loadFileLocked 调用方必须已持有锁 +func (l *Logger) loadFileLocked(key fileKey) *logFile { + f, ok := l.filePath[key] + if !ok || f == nil { + return nil + } + // 时间切割导致文件名变化：关闭旧文件，让调用方按新名字重建 + if f.fileName != l.nowFileName(key.event) { + delete(l.filePath, key) + f.Close() + return nil + } + return f +} + +// close 关闭所有文件句柄（Close 专用） +func (l *Logger) close() error { + l.mu.Lock() + defer l.mu.Unlock() + + var errs []error + for key, f := range l.filePath { + errs = append(errs, f.Close()) + delete(l.filePath, key) + } + return joinErrors(errs) } diff --git a/format.go b/format.go index 64ee1f9..4eba301 100644 --- a/format.go +++ b/format.go @@ -7,84 +7,142 @@ import ( "fmt" "path/filepath" "runtime" + "strconv" "strings" "time" ) func (l *Logger) logger(ctx context.Context, event string, v ...any) { - pc, file, line, _ := runtime.Caller(2) - // fmt.Println("runtime.Caller", pc, file, line, ok) + // 调用方可能是 log 包（Logger.Write -> logger），所以这里取第 2 层 + pc, file, line, ok := runtime.Caller(2) - basePath, _ := filepath.Abs("./") - basePath = strings.ReplaceAll(basePath, "\\", "/") - // fmt.Println("basePath", basePath) - - file = strings.TrimPrefix(file, basePath) - - funcName := runtime.FuncForPC(pc).Name() - funcName = filepath.Ext(funcName) - funcName = strings.TrimPrefix(funcName, ".") - - // by, _ := json.Marshal(v) + var funcName string + if ok && pc != 0 { + if fn := runtime.FuncForPC(pc); fn != nil { + funcName = strings.TrimPrefix(filepath.Ext(fn.Name()), ".") + } + if basePath, err := filepath.Abs("./"); err == nil { + basePath = strings.ReplaceAll(basePath, "\\", "/") + file = strings.TrimPrefix(strings.ReplaceAll(file, "\\", "/"), basePath) + } + } nowTime := time.Now().In(l.option.timeZone).Format("2006-01-02 15:04:05.000000") - traceId, _ := ctx.Value(l.option.traceField).(string) - - // writeStr := "[" + event + "]" + nowTime + " " + file + ":" + fmt.Sprintf("%d", line) + " " + funcName + " gid:" + getGID() + " " + traceId + " @data@: " + string(by) + "\n\n" + // ctx 允许为 nil，不能直接调 Value + var traceId string + if ctx != nil { + traceId, _ = ctx.Value(l.option.traceField).(string) + } + // error 转成带堆栈的字符串（%+v 会带上 pkg/errors 的调用栈） for idx, val := range v { - if _, ok := val.(error); ok { + if _, isErr := val.(error); isErr { v[idx] = fmt.Sprintf("%+v", val) } } + var gid string + if l.option.isGid { + gid = getGID() + } + fd := FormatData{ Time: nowTime, - File: file + ":" + fmt.Sprintf("%d", line), + File: file + ":" + strconv.Itoa(line), Func: funcName, - Gid: getGID(), + Gid: gid, Content: v, TraceId: traceId, - Expand: l.option.expandData, + Expand: l.option.expandData, } - if event == "error" { - // fd.Stack = string(debug.Stack()) - } + fdb := marshalLog(fd, nowTime, file, line, l.option.escapeHTML, gid, traceId) - var fdb []byte - if l.option.escapeHTML { - fdb, _ = json.Marshal(fd) + fdb = append([]byte("\n["+event+"]"), fdb...) + + _, _ = l.write(event, fdb) + + if l.option.errorToInfo && event == "error" { + _, _ = l.write("info", fdb) + } +} + +// marshalLog 序列化日志内容 +// json 序列化失败时（chan/func 等不可序列化类型，或循环引用）不能静默丢弃， +// 否则只会留下一行没有内容的 "["+event+"]"，需要降级成可读的文本 +func marshalLog(fd FormatData, nowTime, file string, line int, escapeHTML bool, gid, traceId string) []byte { + var ( + b []byte + err error + ) + if escapeHTML { + b, err = json.Marshal(fd) } else { - // 非转义 var buf bytes.Buffer encoder := json.NewEncoder(&buf) encoder.SetEscapeHTML(false) - encoder.Encode(fd) - fdb = buf.Bytes() + err = encoder.Encode(fd) + b = bytes.TrimRight(buf.Bytes(), "\n") } - fdb = bytes.TrimRight(fdb, "\n") - - ff := []byte("\n[" + event + "]") - fdb = append(ff, fdb...) - - l.write(event, fdb) - - if l.option.errorToInfo && event == "error" { - l.write("info", fdb) + if err == nil { + return b } - // log.Println("" + string(by)) + // 降级：逐字段尝试，失败的内容用 %+v 兜底 + fallback := FormatData{ + Time: nowTime, + File: file + ":" + strconv.Itoa(line), + Func: fd.Func, + Gid: gid, + TraceId: traceId, + Content: stringifyValues(fd.Content), + Expand: fd.Expand, + } + if fb, ferr := json.Marshal(fallback); ferr == nil { + return fb + } + + return []byte(fmt.Sprintf(`{"time":%q,"content":%q,"marshal_error":%q}`, + nowTime, fmt.Sprintf("%+v", fd.Content), err.Error())) +} + +// stringifyValues 把不可序列化的值转成字符串 +func stringifyValues(v any) []string { + switch val := v.(type) { + case nil: + return nil + case []any: + out := make([]string, 0, len(val)) + for _, item := range val { + out = append(out, fmt.Sprintf("%+v", item)) + } + return out + default: + return []string{fmt.Sprintf("%+v", val)} + } +} + +// getGID 获取当前 goroutine 的 ID +func getGID() string { + b := make([]byte, 64) + b = b[:runtime.Stack(b, false)] + b = bytes.TrimPrefix(b, []byte("goroutine ")) + idx := bytes.IndexByte(b, ' ') + if idx < 0 { + // 解析失败时不要 panic + return "0" + } + return string(b[:idx]) } type FormatData struct { - Time string `json:"time,omitempty"` - File string `json:"file,omitempty"` - Func string `json:"func,omitempty"` - Gid string `json:"gid,omitempty"` - Content interface{} `json:"content,omitempty"` - TraceId string `json:"traceId,omitempty"` - Stack string `json:"stack,omitempty"` - Expand map[string]string `json:"expand,omitempty"` + Time string `json:"time,omitempty"` + File string `json:"file,omitempty"` + Func string `json:"func,omitempty"` + Gid string `json:"gid,omitempty"` + Content interface{} `json:"content,omitempty"` + TraceId string `json:"traceId,omitempty"` + Stack string `json:"stack,omitempty"` + Expand map[string]string `json:"expand,omitempty"` } diff --git a/loggerx.go b/loggerx.go index bc61948..97c0b3b 100644 --- a/loggerx.go +++ b/loggerx.go @@ -5,13 +5,11 @@ package loggerx // desc: 日志封装类 import ( - "bytes" "context" "fmt" "io" "log" "os" - "runtime" "sync" // "sync_log/global" @@ -22,16 +20,37 @@ import ( // 需要实现io.Writer接口 type Logger struct { ctx context.Context - filePath *sync.Map // filePath - mu *sync.Mutex + filePath map[fileKey]*logFile // 每个 (channel,event) 一个句柄 + mu *sync.RWMutex // 保护 filePath + writeMu *sync.Mutex // 串行化文件写入（bufio 不是并发安全的） option loggerOption channel string writeType writeType // 是否异步落盘，这里作用范围是本条，优先判断这里 + + closeOnce *sync.Once + done chan struct{} // 通用关闭信号（delete / ctx 相关 goroutine 用） + workerDone chan struct{} // 异步消费 goroutine 退出信号 + closing chan struct{} // 异步投递关闭信号：只关一次 + + // 异步队列相关的整块状态都挂在指针后面：Logger 会被 Channel()/WriteAsync() + // 浅拷贝，如果状态直接放在 Logger 里，拷贝出的实例就会各看各的 + // （曾经因此让 Close 看不到真正的队列，白等一场、最后一条日志丢失） + async *asyncCtl } -type filePath struct { - file *os.File - fileName string +// asyncCtl 异步队列的共享控制块：所有拷贝共享同一份 +type asyncCtl struct { + mu sync.Mutex + ch chan cacheData + closed bool // 只能读写于 mu 保护下 +} + +// fileKey 文件缓存的键 +// 必须同时包含 channel 与 event：文件名由两者共同决定，只按 channel 缓存会 +// 导致 info/error 互相顶掉句柄，每次写入都 close+open 一次文件 +type fileKey struct { + channel string + event string } func NewLogger(ctx context.Context, opts ...Option) *Logger { @@ -46,11 +65,16 @@ func NewLogger(ctx context.Context, opts ...Option) *Logger { } l := &Logger{ - ctx: ctx, - filePath: &sync.Map{}, - mu: &sync.Mutex{}, - option: opt, - writeType: writeTypeDefault, + ctx: ctx, + filePath: make(map[fileKey]*logFile), + mu: &sync.RWMutex{}, + writeMu: &sync.Mutex{}, + option: opt, + writeType: writeTypeDefault, + closeOnce: &sync.Once{}, + done: make(chan struct{}), + workerDone: make(chan struct{}), + async: &asyncCtl{}, } log.SetOutput(l) @@ -66,21 +90,54 @@ func NewLogger(ctx context.Context, opts ...Option) *Logger { go l.delete() // 强制刷盘 + // 用 done 而不是只监听 ctx：ctx 为 Background 时 Close 也要能把 goroutine 收回去 go func() { - <-ctx.Done() - l.MustSync() + select { + case <-ctx.Done(): + l.MustSync() + case <-l.done: + } }() return l } -// 强制刷盘 -func (l *Logger) MustSync() { - l.filePath.Range(func(key, value any) bool { - f := value.(*filePath) - f.file.Sync() - return true +// 关闭日志：关闭异步队列、落盘并关闭所有文件句柄 +// 重复调用安全，调用后本实例的后台 goroutine 会全部退出 +func (l *Logger) Close() error { + var err error + l.closeOnce.Do(func() { + // 1. 停投递 + 等消费协程把队列里已入队的任务全部写完 + l.drainAsync() + + // 2. 在写锁内完成最后一次刷盘与关闭 + // 此后不会再有写入（投递已关闭，同步写入也要先抢到这把锁） + l.writeMu.Lock() + syncErr := l.MustSync() + closeErr := l.close() + l.writeMu.Unlock() + err = joinErrors([]error{syncErr, closeErr}) + + // 3. 最后再发通用关闭信号，让 delete / ctx 相关 goroutine 退出 + close(l.done) }) + return err +} + +// 强制刷盘：只 flush 缓存数据，文件继续可写 +func (l *Logger) MustSync() error { + l.mu.RLock() + files := make([]*logFile, 0, len(l.filePath)) + for _, f := range l.filePath { + files = append(files, f) + } + l.mu.RUnlock() + + var errs []error + for _, f := range files { + errs = append(errs, f.Sync()) + } + return joinErrors(errs) } func (l *Logger) Channel(ch string) (r *Logger) { @@ -147,32 +204,20 @@ func (l *Logger) Warnf(ctx context.Context, format string, v ...any) { l.logger(ctx, "warn", s) } -// 添加固定的内容 -// func (l *Logger) ContextWithFields(ctx context.Context, v ...any) { -// l.logger(ctx, "add", v...) -// } -// func (l *Logger) Field(key,val string) { -// l.logger(nil, "add", key,val) -// } - -func getGID() string { - b := make([]byte, 64) - b = b[:runtime.Stack(b, false)] - b = bytes.TrimPrefix(b, []byte("goroutine ")) - b = b[:bytes.IndexByte(b, ' ')] - return string(b) -} - // 验证文件夹权限 // 根文件夹如果不存在则创建 func checkDir(dir string) bool { if _, err := os.Stat(dir); err != nil { if os.IsNotExist(err) { - if err := os.MkdirAll(dir, os.ModePerm); err != nil { + if err := os.MkdirAll(dir, 0755); err != nil { log.Println("创建文件夹失败", err) return false } + return true } + // 目标存在但不可访问（权限不足等），此时返回 false 才是这个函数的本意 + log.Println("日志文件夹不可用", dir, err) + return false } return true } diff --git a/loggerx_fixes_test.go b/loggerx_fixes_test.go new file mode 100644 index 0000000..b0628e6 --- /dev/null +++ b/loggerx_fixes_test.go @@ -0,0 +1,380 @@ +package loggerx_test + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log" + "os" + "path/filepath" + "runtime" + "strings" + "sync" + "testing" + "time" + + "github.com/yuninks/loggerx" +) + +// newTestLogger 统一构造测试实例，测试结束关闭句柄 +func newTestLogger(t *testing.T, opts ...loggerx.Option) (*loggerx.Logger, string) { + t.Helper() + dir := t.TempDir() + opts = append([]loggerx.Option{loggerx.SetDir(dir)}, opts...) + l := loggerx.NewLogger(context.Background(), opts...) + t.Cleanup(func() { + if err := l.Close(); err != nil { + t.Errorf("Close 返回错误: %v", err) + } + }) + return l, dir +} + +// syncAndRead 先刷盘再读取（写入走 bufio，落盘后才能看到） +func syncAndRead(t *testing.T, l *loggerx.Logger, dir, event string) string { + t.Helper() + if err := l.MustSync(); err != nil { + t.Fatalf("MustSync: %v", err) + } + return readLog(t, dir, event) +} + +// tailOf 取字符串末尾 n 个字节，用于失败时打印现场 +func tailOf(s string, n int) string { + if len(s) <= n { + return s + } + return s[len(s)-n:] +} + +// readLog 读取切割后的日志文件（文件名可能带日期前缀） +func readLog(t *testing.T, dir, event string) string { + t.Helper() + files, err := filepath.Glob(filepath.Join(dir, "*_"+event+".log")) + if err != nil { + t.Fatalf("glob %s: %v", event, err) + } + if len(files) == 0 { + files, _ = filepath.Glob(filepath.Join(dir, event+".log")) + } + var sb strings.Builder + for _, f := range files { + b, err := os.ReadFile(f) + if err != nil { + t.Fatalf("读取 %s: %v", f, err) + } + sb.Write(b) + } + return sb.String() +} + +// ---------------- 修复1：文件句柄缓存键必须包含 event ---------------- + +// info/error 交替写入后，各自文件的内容必须完整且不串台 +func TestEventHandlerIsolation(t *testing.T) { + l, dir := newTestLogger(t) + + const n = 50 + for i := 0; i < n; i++ { + l.Infof(context.Background(), "INFO-%d", i) + l.Errorf(context.Background(), "ERR-%d", i) + } + + info, errl := syncAndRead(t, l, dir, "info"), syncAndRead(t, l, dir, "error") + if got := strings.Count(info, "INFO-"); got != n { + t.Errorf("info.log 条数 = %d, 期望 %d", got, n) + } + if got := strings.Count(errl, "ERR-"); got != n { + t.Errorf("error.log 条数 = %d, 期望 %d", got, n) + } + if strings.Contains(info, "ERR-") { + t.Error("info.log 中混入了 error 日志") + } + if strings.Contains(errl, "INFO-") { + t.Error("error.log 中混入了 info 日志") + } +} + +// 不同 channel 也必须各自独立 +func TestChannelIsolation(t *testing.T) { + l, dir := newTestLogger(t) + l.Channel("c1").Info(context.Background(), "IN-C1") + l.Channel("c2").Info(context.Background(), "IN-C2") + + if s := syncAndRead(t, l, dir, "info"); strings.Contains(s, "IN-C2") { + t.Error("根 channel 的日志里混入了 c2 的内容") + } + // channel 的日志在子目录里，同样带时间前缀 + c1, err := filepath.Glob(filepath.Join(dir, "c1", "*_info.log")) + if err != nil || len(c1) == 0 { + t.Fatalf("没有找到 c1 的日志文件: %v", err) + } + b, err := os.ReadFile(c1[0]) + if err != nil { + t.Fatalf("读取 c1 日志: %v", err) + } + if !strings.Contains(string(b), "IN-C1") { + t.Errorf("c1 日志内容不正确: %s", string(b)) + } +} + +// 事件交替写入的性能：不能因为切换事件而反复重开文件 +func BenchmarkInfoOnly(b *testing.B) { + l := loggerx.NewLogger(context.Background(), loggerx.SetDir(b.TempDir())) + defer l.Close() + ctx := context.Background() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + l.Infof(ctx, "hello %d", i) + } +} + +func BenchmarkInfoErrorInterleaved(b *testing.B) { + l := loggerx.NewLogger(context.Background(), loggerx.SetDir(b.TempDir())) + defer l.Close() + ctx := context.Background() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + l.Infof(ctx, "hello %d", i) + l.Errorf(ctx, "hello %d", i) + } +} + +// ---------------- 修复2：io.Writer 契约 ---------------- + +func TestWriteSatisfiesIOWriterContract(t *testing.T) { + cases := []struct { + name string + opts []loggerx.Option + }{ + {"默认(写文件)", nil}, + {"只写驱动(不写文件)", []loggerx.Option{loggerx.SetPrintFile(false), loggerx.SetToConsole()}}, + {"文件+附加驱动", []loggerx.Option{loggerx.SetExtraDriver(io.Discard)}}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + l, _ := newTestLogger(t, c.opts...) + b := []byte("hello io.Writer\n") + n, err := l.Write(b) + if err != nil { + t.Fatalf("Write 返回错误: %v", err) + } + if n != len(b) { + t.Fatalf("Write 返回 n=%d, 期望 len(b)=%d（违反 io.Writer 契约）", n, len(b)) + } + }) + } +} + +// 通过标准库 log 写入也必须成功：log 在短写时会报 io.ErrShortWrite +func TestStdLogWriteNoShortWriteError(t *testing.T) { + l, dir := newTestLogger(t) + + // log.SetOutput 是全局状态，恢复现场 + prev := log.Writer() + defer log.SetOutput(prev) + log.SetOutput(l) + + if err := log.Output(2, "via standard log"); err != nil { + t.Fatalf("标准 log 写入失败: %v", err) + } + if s := syncAndRead(t, l, dir, "info"); !strings.Contains(s, "via standard log") { + t.Errorf("日志内容未落盘: %s", s) + } +} + +// ---------------- 修复3：句柄释放与刷盘 ---------------- + +// Close 之后文件必须能被删除（Windows 上句柄没关是删不掉的） +func TestCloseReleasesFileHandles(t *testing.T) { + dir := t.TempDir() + l := loggerx.NewLogger(context.Background(), loggerx.SetDir(dir)) + l.Info(context.Background(), "before close") + + if err := l.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + files, _ := filepath.Glob(filepath.Join(dir, "*.log")) + if len(files) == 0 { + t.Fatal("没有生成日志文件") + } + for _, f := range files { + if err := os.Remove(f); err != nil { + t.Errorf("Close 后仍无法删除 %s: %v", filepath.Base(f), err) + } + } +} + +// Close 必须先把缓存里的内容刷到磁盘 +func TestCloseFlushesBufferedData(t *testing.T) { + dir := t.TempDir() + l := loggerx.NewLogger(context.Background(), loggerx.SetDir(dir)) + l.Info(context.Background(), "buffered-content") + + if err := l.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + files, _ := filepath.Glob(filepath.Join(dir, "*.log")) + if len(files) != 1 { + t.Fatalf("期望 1 个日志文件, 实际 %d", len(files)) + } + b, err := os.ReadFile(files[0]) + if err != nil { + t.Fatalf("读取: %v", err) + } + if !strings.Contains(string(b), "buffered-content") { + t.Errorf("Close 后内容丢失: %q", string(b)) + } +} + +// 重复 Close 必须安全 +func TestCloseIsIdempotent(t *testing.T) { + l := loggerx.NewLogger(context.Background(), loggerx.SetDir(t.TempDir())) + if err := l.Close(); err != nil { + t.Fatalf("首次 Close: %v", err) + } + if err := l.Close(); err != nil { + t.Fatalf("重复 Close: %v", err) + } +} + +// 异步写入 + Close：队列里的内容不能丢 +func TestCloseDrainsAsyncQueue(t *testing.T) { + for _, n := range []int{10, 100, 197, 198, 199, 200, 201, 250} { + t.Run(fmt.Sprintf("n=%d", n), func(t *testing.T) { + dir := t.TempDir() + l := loggerx.NewLogger(context.Background(), loggerx.SetDir(dir), loggerx.SetWriteAsync()) + + al := l.WriteAsync() + for i := 0; i < n; i++ { + al.Infof(context.Background(), "ASYNC-%d", i) + } + if err := l.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + var sb strings.Builder + files, _ := filepath.Glob(filepath.Join(dir, "*.log")) + for _, f := range files { + b, err := os.ReadFile(f) + if err != nil { + t.Fatalf("读取 %s: %v", f, err) + } + sb.Write(b) + } + content := sb.String() + + missing := make([]string, 0) + for i := 0; i < n; i++ { + if !strings.Contains(content, fmt.Sprintf(`ASYNC-%d"`, i)) { + missing = append(missing, fmt.Sprint(i)) + } + } + if len(missing) > 0 { + cnt := strings.Count(content, "[info]") + t.Errorf("丢 %d 条: %s | 实际日志行数=%d n=%d 结尾=%q", len(missing), strings.Join(missing, ","), cnt, n, tailOf(content, 160)) + } + }) + } +} + +// 用 context.Background 创建实例，Close 后后台 goroutine 必须退出 +func TestCloseStopsGoroutines(t *testing.T) { + before := runtime.NumGoroutine() + + loggers := make([]*loggerx.Logger, 0, 10) + for i := 0; i < 10; i++ { + loggers = append(loggers, loggerx.NewLogger(context.Background(), loggerx.SetDir(t.TempDir()))) + } + for _, l := range loggers { + if err := l.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + } + + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if runtime.NumGoroutine() <= before { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Errorf("Close 后 goroutine 未回收: before=%d after=%d", before, runtime.NumGoroutine()) +} + +// ---------------- 修复4：nil ctx / 序列化失败 ---------------- + +func TestNilContextDoesNotPanic(t *testing.T) { + l, dir := newTestLogger(t) + // nolint:staticcheck // 故意传 nil，验证不 panic + l.Info(nil, "nil ctx") + if s := syncAndRead(t, l, dir, "info"); !strings.Contains(s, "nil ctx") { + t.Errorf("nil ctx 时日志未写入: %q", s) + } +} + +// 不可序列化的内容不能产出空行 / 坏 JSON +func TestUnmarshalableContentProducesValidJSON(t *testing.T) { + l, dir := newTestLogger(t) + l.Info(context.Background(), make(chan int), func() {}, map[string]any{"ok": 1}) + l.Infof(context.Background(), "后面这条必须还在") + + content := syncAndRead(t, l, dir, "info") + lines := strings.Split(strings.TrimSpace(content), "\n") + if len(lines) != 2 { + t.Fatalf("期望 2 行日志, 实际 %d 行: %q", len(lines), content) + } + for i, line := range lines { + body := strings.TrimPrefix(line, "[info]") + if body == "" { + t.Fatalf("第 %d 行是空的 [info] 裸行（序列化失败被静默丢弃）", i+1) + } + var v map[string]any + if err := json.Unmarshal([]byte(body), &v); err != nil { + t.Fatalf("第 %d 行不是合法 JSON: %v\n内容: %s", i+1, err, body) + } + if i == 0 && v["content"] == nil { + t.Error("序列化失败时 content 丢失，无法定位问题") + } + } +} + +// 并发写入：数据不能串、不能丢 +func TestConcurrentWriteIntegrity(t *testing.T) { + l, dir := newTestLogger(t) + + var wg sync.WaitGroup + const perLevel = 300 + for i := 0; i < perLevel; i++ { + wg.Add(3) + go func(i int) { defer wg.Done(); l.Infof(context.Background(), "INFO-%d", i) }(i) + go func(i int) { defer wg.Done(); l.Errorf(context.Background(), "ERR-%d", i) }(i) + go func(i int) { defer wg.Done(); l.Channel("cc").Infof(context.Background(), "CC-%d", i) }(i) + } + wg.Wait() + + if got := strings.Count(syncAndRead(t, l, dir, "info"), "INFO-"); got != perLevel { + t.Errorf("info 条数 = %d, 期望 %d", got, perLevel) + } + if got := strings.Count(syncAndRead(t, l, dir, "error"), "ERR-"); got != perLevel { + t.Errorf("error 条数 = %d, 期望 %d", got, perLevel) + } + // cc 是子 channel，日志落在 cc/ 子目录 + ccFiles, _ := filepath.Glob(filepath.Join(dir, "cc", "*_info.log")) + if len(ccFiles) == 0 { + t.Fatal("没有找到 cc 的日志文件") + } + var ccb strings.Builder + for _, f := range ccFiles { + b, _ := os.ReadFile(f) + ccb.Write(b) + } + if got := strings.Count(ccb.String(), "CC-"); got != perLevel { + t.Errorf("cc 条数 = %d, 期望 %d", got, perLevel) + } +} diff --git a/remove.go b/remove.go index be46866..b388967 100644 --- a/remove.go +++ b/remove.go @@ -10,35 +10,48 @@ import ( // 监听dir文件夹的所有文件，循环查找是否有超过days天的文件，删除掉 func (l *Logger) delete() { - tick := time.NewTicker(time.Hour) + defer tick.Stop() for { select { case <-tick.C: - err := l.walkAndDel() - if err != nil { - fmt.Println(err) - } + // 后台清理不能把整个进程带崩 + func() { + defer func() { + if r := recover(); r != nil { + fmt.Println("清理日志文件异常:", r) + } + }() + if err := l.walkAndDel(); err != nil { + fmt.Println(err) + } + }() case <-l.ctx.Done(): return + case <-l.done: + return } } } func (l *Logger) walkAndDel() error { + // 没设置删除天数（<=0 表示不删除） + if l.option.days <= 0 { + return nil + } + + deadline := time.Now().AddDate(0, 0, -l.option.days) + err := filepath.Walk(l.option.dir, func(path string, info os.FileInfo, err error) error { if err != nil { + // 单个文件出错不应该中断整个遍历 fmt.Println(err) - return err - } - // 没设置删除天数 - if l.option.days <= 0 { return nil } - // 判断最后修改时间是否大于3天 - if info.ModTime().After(time.Now().AddDate(0, 0, -l.option.days)) { + // 最后修改时间在保留期内，跳过 + if info.ModTime().After(deadline) { return nil } @@ -46,14 +59,18 @@ func (l *Logger) walkAndDel() error { if !isEmptyDir(path) { return nil } + } else if filepath.Ext(path) != ".log" { + return nil } - ext := filepath.Ext(path) - if ext != ".log" { + + // 删除文件 / 空目录 + if rerr := os.Remove(path); rerr != nil { + // 正被写入的文件在 Windows 上删不掉，等下一轮再试，不要终止遍历 + fmt.Println("删除文件失败", path, rerr) return nil } - // 删除文件 fmt.Println("删除文件", path) - return os.Remove(path) + return nil }) return err diff --git a/storage.go b/storage.go index df7083f..6e007c0 100644 --- a/storage.go +++ b/storage.go @@ -2,71 +2,158 @@ package loggerx import ( "io" - "sync" + "log" + "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) { - // fmt.Println("异步写入") 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) { - if l.option.isPrintFile { - f, err := l.getFile(event, false) + // 串行化写入：句柄缓冲区不是并发安全的， + // 异步消费协程与同步调用可能同时写同一个句柄 + l.writeMu.Lock() + n, err = l.storeFile(event, b) + l.writeMu.Unlock() if err != nil { return 0, err } - - n, err = f.Write(b) - if err == nil && n < len(b) { - err = io.ErrShortWrite - } - if err != nil { - // 强制更新 & 再次写入 - f, err := l.getFile(event, true) - if err == nil { - f.Write(b) - } - } } - if len(l.option.drivers) > 0 { - io.MultiWriter(l.option.drivers...).Write(b) - } - return n, err + // 驱动（控制台等）的写入失败不改变对调用方的契约 + _, _ = l.writeDrivers(b) + + return len(b), nil } -var chanStore = make(chan cacheData, 1000) -var chanOnce = sync.Once{} +// storeFile 写入日志文件 +func (l *Logger) storeFile(event string, b []byte) (int, error) { + f, err := l.getFile(event) + if err != nil { + return 0, err + } + n, err := f.Write(b) + if err == nil && n < len(b) { + err = io.ErrShortWrite + } + if err == nil { + return n, nil + } + + // 写入失败：丢弃这个句柄，落到磁盘后按最新文件名重开一次再写 + // 只重试一次，避免原实现在短写时反复重开文件 + l.discardFile(event, f) + if nf, nerr := l.getFile(event); nerr == nil { + if n2, err2 := nf.Write(b); err2 == nil && n2 == len(b) { + return n2, nil + } + } + + return 0, err +} + +// 写入额外的驱动（控制台 / 自定义 writer） +func (l *Logger) writeDrivers(b []byte) (int, error) { + if len(l.option.drivers) == 0 { + return 0, nil + } + return io.MultiWriter(l.option.drivers...).Write(b) +} + +// discardFile 关闭文件并从缓存中移除，使下次写入重新打开 +func (l *Logger) discardFile(event string, f *logFile) { + _ = f.Close() + + l.mu.Lock() + defer l.mu.Unlock() + key := fileKey{channel: l.channel, event: event} + if cur, ok := l.filePath[key]; ok && cur == f { + delete(l.filePath, key) + } +} + +// 异步队列的任务 type cacheData struct { - logger *Logger - Event string - Data []byte + Event string + Data []byte } +// toAsync 尝试异步写入 +// 返回 true 表示已经交给异步队列，返回 false 表示需要同步写入 func (l *Logger) toAsync(event string, b []byte) bool { - chanOnce.Do(func() { - go func() { - for val := range chanStore { - val.logger.store(val.Event, val.Data) - } - }() - }) - if l.writeType == writeTypeSync || // 指定同步模式 (l.writeType == writeTypeDefault && l.option.writeType != writeTypeAsync) { // 默认同步模式 return false } - // 为了避免丢失，还是要阻塞等待 - chanStore <- cacheData{l, event, b} + // 整段「检查开关 + 入队」都在同一把锁内：Close 也拿这把锁来停止投递， + // 这样 Close 拿到锁时就能确定「要么这条还没入队、要么已经完整入队」。 + // 否则消费协程可能先看到空队列就退出，把还在路上的这条日志整条丢掉。 + // 队列满时这里会阻塞，但消费者是独立 goroutine 且不需要这把锁，不会死锁 + l.async.mu.Lock() + defer l.async.mu.Unlock() + if l.async.closed { + // 已开始关闭：退化为同步写入 + return false + } + if l.async.ch == nil { + l.async.ch = make(chan cacheData, asyncQueueSize) + go l.asyncWorker(l.async.ch) + } + l.async.ch <- cacheData{Event: event, Data: b} return true } + +// asyncWorker 消费异步队列，直到队列被关闭 +func (l *Logger) asyncWorker(q chan cacheData) { + defer close(l.workerDone) + defer func() { + if r := recover(); r != nil { + log.Println("loggerx: 异步写入协程异常:", r) + } + }() + for val := range q { + _, _ = l.store(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 + } + + // 关闭队列并等消费者把缓冲里的任务全部处理完。 + // 这一步必须真的等到 workerDone：否则 Close 会在消费协程还在写缓冲时 + // 就刷盘并关闭文件，最后几条日志会连着句柄一起丢掉 + close(q) + select { + case <-l.workerDone: + case <-time.After(closeDrainTimeout): + log.Println("loggerx: 等待异步日志落盘超时") + } +}