Update logger to handle concurrent writes
This commit is contained in:
NeilClack
+180
-54
@@ -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()
|
||||
file = nil
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Swap channels (owned exclusively by drain goroutine) ────────────────────
|
||||
|
||||
var (
|
||||
fileSwapCh = make(chan *os.File, 1)
|
||||
dbSwapCh = make(chan *gorm.DB, 1)
|
||||
drainDone = make(chan struct{})
|
||||
)
|
||||
|
||||
// ─── 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]
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
|
||||
// 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
|
||||
|
||||
_, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
|
||||
if err != nil {
|
||||
return err
|
||||
// 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)
|
||||
}
|
||||
|
||||
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 {
|
||||
flush()
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// 1. The standard log package is internally thread-safe
|
||||
console.Println(line)
|
||||
// ─── Writers (called only from drain goroutine) ───────────────────────────────
|
||||
|
||||
// 2. Write to file
|
||||
func writeToFile(e entry) {
|
||||
line := formatLine(e)
|
||||
if file != nil {
|
||||
fmt.Fprintln(file, line)
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
|
||||
// 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)
|
||||
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 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...) }
|
||||
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,
|
||||
)
|
||||
}
|
||||
|
||||
func Fatal(format string, args ...any) {
|
||||
logMsg(LevelFatal, format, args...)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
// ─── 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...) }
|
||||
|
||||
Reference in New Issue
Block a user