diff --git a/pkg/db/task_logs.go b/pkg/db/task_logs.go index 16d600ff..f95ebffc 100644 --- a/pkg/db/task_logs.go +++ b/pkg/db/task_logs.go @@ -18,7 +18,7 @@ func (db *Database) InsertTaskLog(tx *sqlx.Tx, log *TaskLog) error { INSERT INTO task_logs ( run_id, task_id, log_idx, log_time, log_level, log_fields, log_message ) VALUES ($1, $2, $3, $4, $5, $6, $7) - ON CONFLICT (run_id, task_id, log_time) DO UPDATE SET + ON CONFLICT (run_id, task_id, log_idx) DO UPDATE SET log_time = excluded.log_time, log_level = excluded.log_level, log_fields = excluded.log_fields, diff --git a/pkg/db/task_logs_test.go b/pkg/db/task_logs_test.go new file mode 100644 index 00000000..67da1487 --- /dev/null +++ b/pkg/db/task_logs_test.go @@ -0,0 +1,75 @@ +package db + +import ( + "io" + "testing" + + "github.com/jmoiron/sqlx" + "github.com/sirupsen/logrus" +) + +func newSqliteTestDB(t *testing.T) *Database { + t.Helper() + + quiet := logrus.New() + quiet.SetOutput(io.Discard) + + database := NewDatabase(quiet) + if err := database.InitDB(&DatabaseConfig{ + Engine: "sqlite", + Sqlite: &SqliteDatabaseConfig{File: t.TempDir() + "/test.sqlite"}, + }); err != nil { + t.Fatalf("init db: %v", err) + } + + if err := database.ApplySchema(-2); err != nil { + t.Fatalf("apply schema: %v", err) + } + + return database +} + +// TestInsertTaskLogPersistsAndUpserts verifies that a task log is persisted and +// that re-inserting the same primary key (run_id, task_id, log_idx) updates the +// row in place rather than duplicating or failing. The upsert conflict target +// must be the primary key; any other target is rejected by Postgres at execution +// time, which previously made every insert fail on that engine. +func TestInsertTaskLogPersistsAndUpserts(t *testing.T) { + database := newSqliteTestDB(t) + + insert := func(l *TaskLog) { + if err := database.RunTransaction(func(tx *sqlx.Tx) error { + return database.InsertTaskLog(tx, l) + }); err != nil { + t.Fatalf("insert task log: %v", err) + } + } + + insert(&TaskLog{RunID: 1, TaskID: 1, LogIndex: 1, LogTime: 100, LogLevel: 1, LogMessage: "hello"}) + insert(&TaskLog{RunID: 1, TaskID: 1, LogIndex: 2, LogTime: 110, LogLevel: 1, LogMessage: "world"}) + + logs, err := database.GetTaskLogs(1, 1, 0, 100) + if err != nil { + t.Fatalf("get task logs: %v", err) + } + + if len(logs) != 2 { + t.Fatalf("expected 2 persisted logs, got %d", len(logs)) + } + + // re-insert log index 1 with new content + insert(&TaskLog{RunID: 1, TaskID: 1, LogIndex: 1, LogTime: 200, LogLevel: 2, LogMessage: "updated"}) + + logs, err = database.GetTaskLogs(1, 1, 0, 100) + if err != nil { + t.Fatalf("get task logs: %v", err) + } + + if len(logs) != 2 { + t.Fatalf("after upsert: expected 2 logs, got %d", len(logs)) + } + + if logs[0].LogMessage != "updated" { + t.Fatalf("after upsert: expected log index 1 to read \"updated\", got %q", logs[0].LogMessage) + } +} diff --git a/pkg/logger/dbwriter.go b/pkg/logger/dbwriter.go index 28669bc3..11bec3bb 100644 --- a/pkg/logger/dbwriter.go +++ b/pkg/logger/dbwriter.go @@ -16,14 +16,11 @@ type logDBWriter struct { bufferSize uint64 flushDelay time.Duration - lastIdx uint64 - flushIdx uint64 - bufIdx uint64 - bufMtx sync.Mutex - buf []*db.TaskLog - flushMtx sync.Mutex - flushing bool - flushingChan chan bool + lastIdx uint64 + bufIdx uint64 + bufMtx sync.Mutex + buf []*db.TaskLog + flushScheduled bool } func newLogDBWriter(logger *LogScope, bufferSize uint64, flushDelay time.Duration) *logDBWriter { @@ -44,7 +41,7 @@ func (lh *logDBWriter) Fire(entry *logrus.Entry) error { defer lh.bufMtx.Unlock() if lh.bufIdx >= lh.bufferSize { - lh.flushToDB() + lh.flushToDBLocked() } logIdx := lh.lastIdx + 1 @@ -69,59 +66,59 @@ func (lh *logDBWriter) Fire(entry *logrus.Entry) error { lh.buf = append(lh.buf, taskLog) lh.bufIdx++ - lh.flushDelayed() + lh.scheduleFlushLocked() return nil } -func (lh *logDBWriter) flushDelayed() { - if lh.flushing { +// scheduleFlushLocked arranges for the buffer to be flushed after flushDelay. +// The caller must hold bufMtx. At most one delayed flush is pending at a time; +// a flush triggered earlier by the buffer filling up simply leaves the pending +// timer to find an empty buffer and do nothing. +func (lh *logDBWriter) scheduleFlushLocked() { + if lh.flushScheduled { return } - lh.flushing = true - lh.flushingChan = make(chan bool) + lh.flushScheduled = true - go func() { - defer func() { - lh.flushing = false - }() - - select { - case <-lh.flushingChan: - lh.flushingChan = nil - return - case <-time.After(2 * time.Second): - } + delay := lh.flushDelay + if delay <= 0 { + delay = 2 * time.Second + } - lh.flushingChan = nil + go func() { + time.Sleep(delay) lh.bufMtx.Lock() defer lh.bufMtx.Unlock() - lh.flushToDB() + lh.flushScheduled = false + lh.flushToDBLocked() }() } -func (lh *logDBWriter) flushToDB() { - lh.flushMtx.Lock() - - defer func() { - lh.flushMtx.Unlock() - }() +// flush writes any buffered entries to the database. +func (lh *logDBWriter) flush() { + lh.bufMtx.Lock() + defer lh.bufMtx.Unlock() - if flushingChan := lh.flushingChan; flushingChan != nil { - close(flushingChan) - } + lh.flushToDBLocked() +} +// flushToDBLocked writes the buffered entries to the database. The caller must +// hold bufMtx. The buffer is always cleared afterwards, even on error, so a +// persistent database failure cannot grow it without bound. Errors are reported +// through the parent logger, not this writer's own logger, whose db hook would +// re-enter Fire and deadlock on bufMtx. +func (lh *logDBWriter) flushToDBLocked() { if len(lh.buf) == 0 { return } err := lh.logger.options.Database.RunTransaction(func(tx *sqlx.Tx) error { for _, entry := range lh.buf { - err := lh.logger.options.Database.InsertTaskLog(tx, entry) - if err != nil { + if err := lh.logger.options.Database.InsertTaskLog(tx, entry); err != nil { return err } } @@ -129,13 +126,20 @@ func (lh *logDBWriter) flushToDB() { return nil }) if err != nil { - lh.logger.logger.Errorf("failed to write log entries to db: %v", err) - return + lh.logFlushError(err) } lh.buf = lh.buf[:0] lh.bufIdx = 0 - lh.flushIdx = lh.lastIdx +} + +func (lh *logDBWriter) logFlushError(err error) { + if lh.logger.parentLogger != nil { + lh.logger.parentLogger.WithError(err).Error("failed to write task log entries to db") + return + } + + logrus.WithError(err).Error("failed to write task log entries to db") } func (lh *logDBWriter) getBufferEntries() []*db.TaskLog { diff --git a/pkg/logger/dbwriter_test.go b/pkg/logger/dbwriter_test.go new file mode 100644 index 00000000..5453fffa --- /dev/null +++ b/pkg/logger/dbwriter_test.go @@ -0,0 +1,111 @@ +package logger + +import ( + "io" + "testing" + "time" + + "github.com/ethpandaops/assertoor/pkg/db" + "github.com/sirupsen/logrus" +) + +func newSqliteDB(t *testing.T, applySchema bool) *db.Database { + t.Helper() + + quiet := logrus.New() + quiet.SetOutput(io.Discard) + + database := db.NewDatabase(quiet) + if err := database.InitDB(&db.DatabaseConfig{ + Engine: "sqlite", + Sqlite: &db.SqliteDatabaseConfig{File: t.TempDir() + "/logger.sqlite"}, + }); err != nil { + t.Fatalf("init db: %v", err) + } + + if applySchema { + if err := database.ApplySchema(-2); err != nil { + t.Fatalf("apply schema: %v", err) + } + } + + return database +} + +// TestFlushDoesNotDeadlockOrGrowOnDBError drives the real logger with a database +// whose task_logs table does not exist, so every flush fails. Previously the flush +// error was logged through the writer's own logger, whose db hook re-entered Fire +// and deadlocked on the buffer mutex, and the buffer was never trimmed on error so +// it grew without bound. Both must be gone: logging must make progress and the +// buffer must stay bounded. +func TestFlushDoesNotDeadlockOrGrowOnDBError(t *testing.T) { + quiet := logrus.New() + quiet.SetOutput(io.Discard) + + database := newSqliteDB(t, false) // no schema -> every InsertTaskLog fails + + ls := NewLogger(&ScopeOptions{ + Parent: quiet, + Database: database, + BufferSize: 4, + TestRunID: 1, + TaskID: 1, + }) + log := ls.GetLogger() + + done := make(chan struct{}) + + go func() { + for i := 0; i < 1000; i++ { + log.Infof("entry %d", i) + } + + close(done) + }() + + select { + case <-done: + case <-time.After(15 * time.Second): + t.Fatal("logging deadlocked against a failing database") + } + + ls.dbWriter.bufMtx.Lock() + buffered := len(ls.dbWriter.buf) + ls.dbWriter.bufMtx.Unlock() + + if uint64(buffered) > ls.dbWriter.bufferSize { + t.Fatalf("buffer grew to %d entries on persistent db error (bufferSize %d)", buffered, ls.dbWriter.bufferSize) + } +} + +// TestLogsPersistThroughWriter verifies the happy path still works after the flush +// rewrite: entries logged through the scope are written to the database on flush. +func TestLogsPersistThroughWriter(t *testing.T) { + quiet := logrus.New() + quiet.SetOutput(io.Discard) + + database := newSqliteDB(t, true) + + ls := NewLogger(&ScopeOptions{ + Parent: quiet, + Database: database, + TestRunID: 7, + TaskID: 3, + }) + log := ls.GetLogger() + + log.Info("first") + log.Info("second") + log.Info("third") + + ls.Flush() + + logs, err := database.GetTaskLogs(7, 3, 0, 100) + if err != nil { + t.Fatalf("get task logs: %v", err) + } + + if len(logs) != 3 { + t.Fatalf("expected 3 persisted logs, got %d", len(logs)) + } +} diff --git a/pkg/logger/logscope.go b/pkg/logger/logscope.go index 19437438..7635bf53 100644 --- a/pkg/logger/logscope.go +++ b/pkg/logger/logscope.go @@ -98,7 +98,7 @@ func (ls *LogScope) GetLogger() *logrus.Logger { func (ls *LogScope) Flush() { if ls.dbWriter != nil { - ls.dbWriter.flushToDB() + ls.dbWriter.flush() } }