From bc6894b79a218ba64a2cc0b3d057c608a785919e Mon Sep 17 00:00:00 2001 From: NeilClack Date: Fri, 26 Jun 2026 18:06:19 -0500 Subject: [PATCH] Update logger to handle concurrent writes --- internal/logger/logger.go | 240 +++++++++++++++++++++++++++++--------- 1 file changed, 183 insertions(+), 57 deletions(-) diff --git a/internal/logger/logger.go b/internal/logger/logger.go index 4609f90..8214b06 100644 --- a/internal/logger/logger.go +++ b/internal/logger/logger.go @@ -10,6 +10,8 @@ import ( "rhemawavetech.com/rhemawave/internal/db" ) +// ─── Level ─────────────────────────────────────────────────────────────────── + type Level int const ( @@ -24,82 +26,206 @@ func (l Level) String() string { return [...]string{"DEBUG", "INFO", "WARNING", "ERROR", "FATAL"}[l] } +// ─── Internal entry type ────────────────────────────────────────────────────── + +type entry struct { + level Level + message string + timestamp time.Time + fatal bool +} + +// ─── Package state ──────────────────────────────────────────────────────────── + var ( - DB *gorm.DB - file *os.File - console = log.New(os.Stderr, "", 0) SetLevel Level = LevelInfo + console = log.New(os.Stderr, "", 0) + + ch chan entry // unbuffered send is a bug; always buffered + done chan struct{} + db_ *gorm.DB + file *os.File ) -// AddFile attaches or replaces the file backend +func init() { + ch = make(chan entry, 4096) + done = make(chan struct{}) + go drain() +} + +// ─── Configuration ──────────────────────────────────────────────────────────── + +// AddFile attaches or replaces the file backend. Safe to call at any time. func AddFile(path string) error { + f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) + if err != nil { + return fmt.Errorf("logger: open file: %w", err) + } + // Send a sentinel that carries the new file handle. + // We do this through the channel so the drain goroutine owns + // all file state — no mutex needed. + fileSwapCh <- f + return nil +} + +// AddDatabase attaches or replaces the database backend. Safe to call at any time. +func AddDatabase(d *gorm.DB) { + dbSwapCh <- d +} + +// Close drains remaining entries and shuts down the logger. +// Call this with defer in main(). +func Close() { + close(done) + <-drainDone // wait for drain goroutine to finish flushing if file != nil { file.Close() - } - - _, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) - if err != nil { - return err - } - - return nil -} - -// AddDatabase attaches or replaces the database backend -func AddDatabase(d *gorm.DB) { - console.Println("Adding database to logger") - DB = d -} - -// Close shuts down the file backend. -func Close() error { - if file != nil { - err := file.Close() file = nil - return err } - return nil } -func logMsg(level Level, format string, args ...any) { - if level == LevelDebug && SetLevel != LevelDebug { - return - } - message := fmt.Sprintf(format, args...) - now := time.Now() - timestamp := now.Format("2006-01-02 15:04:05.000") - line := fmt.Sprintf("%s [%s] %s", timestamp, level, message) +// ─── Swap channels (owned exclusively by drain goroutine) ──────────────────── - // 1. The standard log package is internally thread-safe - console.Println(line) +var ( + fileSwapCh = make(chan *os.File, 1) + dbSwapCh = make(chan *gorm.DB, 1) + drainDone = make(chan struct{}) +) - // 2. Write to file - if file != nil { - fmt.Fprintln(file, line) +// ─── Drain goroutine ────────────────────────────────────────────────────────── +// +// This is the ONLY goroutine that touches file, db_, or does DB writes. +// Everything else just sends to ch. + +func drain() { + defer close(drainDone) + + batch := make([]entry, 0, 64) + ticker := time.NewTicker(250 * time.Millisecond) + defer ticker.Stop() + + flush := func() { + if len(batch) == 0 { + return + } + writeBatchToDB(batch) + batch = batch[:0] } - // 3. Write to DB using GORM. It handles its own connection pooling and locking. - if DB != nil { - var entry db.Log - entry.Timestamp = now.UnixMilli() - entry.Level = level.String() - entry.Message = message + for { + select { - // Use Save() to insert the new record. - result := DB.Save(&entry) - if result.Error != nil { - console.Printf("Failed to write log to database: %v\n", result.Error) + // New log entry + case e := <-ch: + writeToFile(e) + if db_ != nil { + batch = append(batch, e) + if len(batch) >= 64 { + flush() + } + } + + // Periodic DB flush so low-volume logs don't sit in the batch forever + case <-ticker.C: + flush() + + // Hot-swap the file backend + case f := <-fileSwapCh: + if file != nil { + file.Close() + } + file = f + + // Hot-swap the DB backend + case d := <-dbSwapCh: + flush() // flush any pending entries to the old db first + db_ = d + console.Println("logger: database backend attached") + + // Shutdown: drain what's left, then exit + case <-done: + // Pull any entries still sitting in the channel + for len(ch) > 0 { + e := <-ch + writeToFile(e) + batch = append(batch, e) + } + flush() + return } } } -func Debug(format string, args ...any) { logMsg(LevelDebug, format, args...) } -func Info(format string, args ...any) { logMsg(LevelInfo, format, args...) } -func Warning(format string, args ...any) { logMsg(LevelWarning, format, args...) } -func Error(format string, args ...any) { logMsg(LevelError, format, args...) } +// ─── Writers (called only from drain goroutine) ─────────────────────────────── -func Fatal(format string, args ...any) { - logMsg(LevelFatal, format, args...) - time.Sleep(100 * time.Millisecond) - os.Exit(1) +func writeToFile(e entry) { + line := formatLine(e) + if file != nil { + fmt.Fprintln(file, line) + } } + +func writeBatchToDB(batch []entry) { + if db_ == nil || len(batch) == 0 { + return + } + records := make([]db.Log, len(batch)) + for i, e := range batch { + records[i] = db.Log{ + Timestamp: e.timestamp.UnixMilli(), + Level: e.level.String(), + Message: e.message, + } + } + // Single INSERT ... VALUES (...), (...), (...) instead of N individual saves. + if result := db_.CreateInBatches(records, 64); result.Error != nil { + console.Printf("logger: db batch write failed: %v\n", result.Error) + } +} + +func formatLine(e entry) string { + return fmt.Sprintf("%s [%s] %s", + e.timestamp.Format("2006-01-02 15:04:05.000"), + e.level, + e.message, + ) +} + +// ─── Public API ─────────────────────────────────────────────────────────────── + +func logMsg(level Level, fatal bool, format string, args ...any) { + if level == LevelDebug && SetLevel != LevelDebug { + return + } + e := entry{ + level: level, + message: fmt.Sprintf(format, args...), + timestamp: time.Now(), // captured here, in the caller's goroutine + fatal: fatal, + } + + // Console is synchronous — log.Logger is internally mutex-protected, + // so this is safe from any goroutine and gives you immediate output. + console.Println(formatLine(e)) + + // Non-blocking send: if the buffer is full, we drop rather than deadlock. + // 4096 entries of headroom means this should never happen in practice. + select { + case ch <- e: + default: + console.Println("logger: channel full, entry dropped") + } + + if fatal { + // Give the drain goroutine a moment to flush before we exit. + // Close() is cleaner but Fatal has to exit immediately. + time.Sleep(150 * time.Millisecond) + os.Exit(1) + } +} + +func Debug(format string, args ...any) { logMsg(LevelDebug, false, format, args...) } +func Info(format string, args ...any) { logMsg(LevelInfo, false, format, args...) } +func Warning(format string, args ...any) { logMsg(LevelWarning, false, format, args...) } +func Error(format string, args ...any) { logMsg(LevelError, false, format, args...) } +func Fatal(format string, args ...any) { logMsg(LevelFatal, true, format, args...) }