write logs to database

Author: pk910

pkg/coordinator/coordinator.go

@@ -69,8 +69,8 @@ type testDescriptorEntry struct {
 func NewCoordinator(config *Config, log logrus.FieldLogger, metricsPort int) *Coordinator {
 	return &Coordinator{
 		log: logger.NewLogger(&logger.ScopeOptions{
-			Parent:      log,
-			HistorySize: 5000,
+			Parent:     log,
+			BufferSize: 5000,
 		}),
 		Config:      config,
 		metricsPort: metricsPort,
@@ -215,7 +215,7 @@ func (c *Coordinator) Logger() logrus.FieldLogger {
 	return c.log.GetLogger()
 }
 
-func (c *Coordinator) LogScope() *logger.LogScope {
+func (c *Coordinator) LogReader() logger.LogReader {
 	return c.log
 }
 
@@ -619,7 +619,6 @@ runLoop:
 		c.testRegistryMutex.Unlock()
 
 		if nextTest != nil {
-
 			// run next test
 			testFunc := func(nextTest types.TestRunner) {
 				defer func() {

pkg/coordinator/db/schema/pgsql/20240913135112_init.sql

@@ -53,18 +53,20 @@ CREATE TABLE IF NOT EXISTS public."task_states"
     "task_config" TEXT NOT NULL,
     "task_status" TEXT NOT NULL,
     "task_result" INTEGER NOT NULL,
+    "task_error" TEXT NOT NULL,
     CONSTRAINT "task_states_pkey" PRIMARY KEY ("run_id", "task_id")
 );
 
 CREATE TABLE IF NOT EXISTS public."task_logs"
 (
     "run_id" INTEGER NOT NULL,
     "task_id" INTEGER NOT NULL,
+    "log_idx" INTEGER NOT NULL,
     "log_time" BIGINT NOT NULL,
-    "log_level" VARCHAR(16) NOT NULL,
+    "log_level" SMALLINT NOT NULL,
     "log_fields" TEXT NOT NULL,
     "log_message" TEXT NOT NULL,
-    CONSTRAINT "task_logs_pkey" PRIMARY KEY ("run_id", "task_id", "log_time")
+    CONSTRAINT "task_logs_pkey" PRIMARY KEY ("run_id", "task_id", "log_idx")
 );
 
 -- +goose StatementEnd

pkg/coordinator/db/schema/sqlite/20240913135112_init.sql

@@ -53,18 +53,20 @@ CREATE TABLE IF NOT EXISTS "task_states"
     "task_config" TEXT NOT NULL,
     "task_status" TEXT NOT NULL,
     "task_result" INTEGER NOT NULL,
+    "task_error" TEXT NOT NULL,
     CONSTRAINT "task_states_pkey" PRIMARY KEY ("run_id", "task_id")
 );
 
 CREATE TABLE IF NOT EXISTS "task_logs"
 (
     "run_id" INTEGER NOT NULL,
     "task_id" INTEGER NOT NULL,
+    "log_idx" INTEGER NOT NULL,
     "log_time" INTEGER NOT NULL,
-    "log_level" TEXT NOT NULL,
+    "log_level" INTEGER NOT NULL,
     "log_fields" TEXT NOT NULL,
     "log_message" TEXT NOT NULL,
-    CONSTRAINT "task_logs_pkey" PRIMARY KEY ("run_id", "task_id", "log_time")
+    CONSTRAINT "task_logs_pkey" PRIMARY KEY ("run_id", "task_id", "log_idx")
 );
 
 -- +goose StatementEnd

pkg/coordinator/db/task_logs.go

@@ -5,8 +5,9 @@ import "github.com/jmoiron/sqlx"
 type TaskLog struct {
 	RunID      int    `db:"run_id"`
 	TaskID     int    `db:"task_id"`
+	LogIndex   int    `db:"log_idx"`
 	LogTime    int64  `db:"log_time"`
-	LogLevel   string `db:"log_level"`
+	LogLevel   int    `db:"log_level"`
 	LogFields  string `db:"log_fields"`
 	LogMessage string `db:"log_message"`
 }
@@ -15,21 +16,52 @@ func (db *Database) InsertTaskLog(tx *sqlx.Tx, log *TaskLog) error {
 	_, err := tx.Exec(db.EngineQuery(map[EngineType]string{
 		EnginePgsql: `
 			INSERT INTO task_logs (
-				run_id, task_id, log_time, log_level, log_fields, log_message
-			) VALUES ($1, $2, $3, $4, $5, $6)
+				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
+				log_time = excluded.log_time,
 				log_level = excluded.log_level,
 				log_fields = excluded.log_fields,
 				log_message = excluded.log_message`,
 		EngineSqlite: `
 			INSERT OR REPLACE INTO task_logs (
-				run_id, task_id, log_time, log_level, log_fields, log_message
-			) VALUES ($1, $2, $3, $4, $5, $6)`,
+				run_id, task_id, log_idx, log_time, log_level, log_fields, log_message
+			) VALUES ($1, $2, $3, $4, $5, $6, $7)`,
 	}),
-		log.RunID, log.TaskID, log.LogTime, log.LogLevel, log.LogFields, log.LogMessage)
+		log.RunID, log.TaskID, log.LogIndex, log.LogTime, log.LogLevel, log.LogFields, log.LogMessage)
 	if err != nil {
 		return err
 	}
 
 	return nil
 }
+
+func (db *Database) GetTaskLogs(runID, taskID, fromIdx, limit int) ([]*TaskLog, error) {
+	var logs []*TaskLog
+
+	err := db.reader.Select(&logs, `
+		SELECT * FROM task_logs
+		WHERE run_id = $1 AND task_id = $2 AND log_idx >= $3
+		ORDER BY log_idx ASC
+		LIMIT $4`,
+		runID, taskID, fromIdx, limit)
+	if err != nil {
+		return nil, err
+	}
+
+	return logs, nil
+}
+
+func (db *Database) GetLastLogIndex(runID, taskID int) (int, error) {
+	var logIdx int
+
+	err := db.reader.Get(&logIdx, `
+		SELECT COALESCE(MAX(log_idx), 0) FROM task_logs
+		WHERE run_id = $1 AND task_id = $2`,
+		runID, taskID)
+	if err != nil {
+		return 0, err
+	}
+
+	return logIdx, nil
+}

pkg/coordinator/db/task_states.go

@@ -1,6 +1,11 @@
 package db
 
-import "github.com/jmoiron/sqlx"
+import (
+	"fmt"
+	"strings"
+
+	"github.com/jmoiron/sqlx"
+)
 
 /*
 CREATE TABLE IF NOT EXISTS public."task_states"
@@ -42,15 +47,16 @@ type TaskState struct {
 	TaskConfig string `db:"task_config"`
 	TaskStatus string `db:"task_status"`
 	TaskResult int    `db:"task_result"`
+	TaskError  string `db:"task_error"`
 }
 
 func (db *Database) InsertTaskState(tx *sqlx.Tx, state *TaskState) error {
 	_, err := tx.Exec(db.EngineQuery(map[EngineType]string{
 		EnginePgsql: `
 			INSERT INTO task_states (
 				run_id, task_id, parent_task, name, title, timeout, ifcond, is_cleanup, is_started, is_running, is_skipped, is_timeout, 
-				start_time, stop_time, task_config, task_status, task_result
-			) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)
+				start_time, stop_time, task_config, task_status, task_result, task_error
+			) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18)
 			ON CONFLICT (run_id, task_id) DO UPDATE SET
 				parent_task = excluded.parent_task,
 				name = excluded.name,
@@ -66,16 +72,79 @@ func (db *Database) InsertTaskState(tx *sqlx.Tx, state *TaskState) error {
 				stop_time = excluded.stop_time,
 				task_config = excluded.task_config,
 				task_status = excluded.task_status,
-				task_result = excluded.task_result`,
+				task_result = excluded.task_result,
+				task_error = excluded.task_error`,
 		EngineSqlite: `
 			INSERT OR REPLACE INTO task_states (
 				run_id, task_id, parent_task, name, title, timeout, ifcond, is_cleanup, is_started, is_running, is_skipped, is_timeout, 
-				start_time, stop_time, task_config, task_status, task_result
-			) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)`,
+				start_time, stop_time, task_config, task_status, task_result, task_error
+			) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18)`,
 	}),
 		state.RunID, state.TaskID, state.ParentTask, state.Name, state.Title, state.Timeout, state.IfCond, state.IsCleanup, state.IsStarted, state.IsRunning,
 		state.IsSkipped, state.IsTimeout, state.StartTime, state.StopTime, state.TaskConfig, state.TaskStatus,
-		state.TaskResult)
+		state.TaskResult, state.TaskError)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (db *Database) UpdateTaskState(tx *sqlx.Tx, state *TaskState, updateFields []string) error {
+	var sql strings.Builder
+
+	args := []any{}
+
+	fmt.Fprint(&sql, `UPDATE task_states SET `)
+
+	for i, field := range updateFields {
+		if i > 0 {
+			fmt.Fprint(&sql, `, `)
+		}
+
+		switch field {
+		case "title":
+			fmt.Fprintf(&sql, `title = $%v`, len(args)+1)
+			args = append(args, state.Title)
+		case "is_started":
+			fmt.Fprintf(&sql, `is_started = $%v`, len(args)+1)
+			args = append(args, state.IsStarted)
+		case "is_running":
+			fmt.Fprintf(&sql, `is_running = $%v`, len(args)+1)
+			args = append(args, state.IsRunning)
+		case "is_skipped":
+			fmt.Fprintf(&sql, `is_skipped = $%v`, len(args)+1)
+			args = append(args, state.IsSkipped)
+		case "is_timeout":
+			fmt.Fprintf(&sql, `is_timeout = $%v`, len(args)+1)
+			args = append(args, state.IsTimeout)
+		case "start_time":
+			fmt.Fprintf(&sql, `start_time = $%v`, len(args)+1)
+			args = append(args, state.StartTime)
+		case "stop_time":
+			fmt.Fprintf(&sql, `stop_time = $%v`, len(args)+1)
+			args = append(args, state.StopTime)
+		case "task_config":
+			fmt.Fprintf(&sql, `task_config = $%v`, len(args)+1)
+			args = append(args, state.TaskConfig)
+		case "task_status":
+			fmt.Fprintf(&sql, `task_status = $%v`, len(args)+1)
+			args = append(args, state.TaskStatus)
+		case "task_result":
+			fmt.Fprintf(&sql, `task_result = $%v`, len(args)+1)
+			args = append(args, state.TaskResult)
+		case "task_error":
+			fmt.Fprintf(&sql, `task_error = $%v`, len(args)+1)
+			args = append(args, state.TaskError)
+		default:
+			return fmt.Errorf("unknown field %q", field)
+		}
+	}
+
+	fmt.Fprintf(&sql, ` WHERE run_id = $%v AND task_id = $%v`, len(args)+1, len(args)+2)
+	args = append(args, state.RunID, state.TaskID)
+
+	_, err := tx.Exec(sql.String(), args...)
 	if err != nil {
 		return err
 	}

pkg/coordinator/logger/dbreader.go

@@ -0,0 +1,44 @@
+package logger
+
+import "github.com/ethpandaops/assertoor/pkg/coordinator/db"
+
+type LogDBReader struct {
+	database  *db.Database
+	testRunID uint64
+	taskID    uint64
+	lastIdx   *int
+}
+
+func NewLogDBReader(database *db.Database, testRunID, taskID uint64) *LogDBReader {
+	return &LogDBReader{
+		database:  database,
+		testRunID: testRunID,
+		taskID:    taskID,
+	}
+}
+
+func (ls *LogDBReader) GetLogEntryCount() int {
+	if ls.lastIdx == nil {
+		lastIdx, err := ls.database.GetLastLogIndex(int(ls.testRunID), int(ls.taskID))
+		if err != nil {
+			return 0
+		}
+
+		ls.lastIdx = &lastIdx
+	}
+
+	if ls.lastIdx == nil {
+		return 0
+	}
+
+	return *ls.lastIdx
+}
+
+func (ls *LogDBReader) GetLogEntries(from, limit int) []*db.TaskLog {
+	dbEntries, err := ls.database.GetTaskLogs(int(ls.testRunID), int(ls.taskID), from, limit)
+	if err != nil {
+		return nil
+	}
+
+	return dbEntries
+}