diff --git a/internal/pgengine/transaction.go b/internal/pgengine/transaction.go index 4655ace5..a3688c3a 100644 --- a/internal/pgengine/transaction.go +++ b/internal/pgengine/transaction.go @@ -123,7 +123,9 @@ func (pge *PgEngine) ExecuteSQLCommand(ctx context.Context, executor executor, t return errors.New("SQL command cannot be empty") } if len(paramValues) == 0 { //mimic empty param + task.MarkStart() ct, e := executor.Exec(ctx, task.Command) + task.MarkDone() pge.LogTaskExecution(context.Background(), task, errCodes[e != nil], ct.String(), "") return e } @@ -135,8 +137,10 @@ func (pge *PgEngine) ExecuteSQLCommand(ctx context.Context, executor executor, t err = errors.Join(err, fmt.Errorf("failed to parse parameter %s: %w", val, parseErr)) return } + task.MarkStart() ct, e := executor.Exec(ctx, task.Command, params...) err = errors.Join(err, e) + task.MarkDone() pge.LogTaskExecution(context.Background(), task, errCodes[e != nil], ct.String(), val) } return diff --git a/internal/pgengine/transaction_test.go b/internal/pgengine/transaction_test.go index f6575392..22c11d2f 100644 --- a/internal/pgengine/transaction_test.go +++ b/internal/pgengine/transaction_test.go @@ -185,12 +185,16 @@ func TestExecuteSQLCommand(t *testing.T) { assert.Error(t, err) mockPool.ExpectExec("correct json").WillReturnResult(pgxmock.NewResult("EXECUTE", 0)) - err = pge.ExecuteSQLCommand(ctx, mockPool, &pgengine.ChainTask{Command: "correct json"}, []string{}) + taskEmptyArgs := &pgengine.ChainTask{Command: "correct json"} + err = pge.ExecuteSQLCommand(ctx, mockPool, taskEmptyArgs, []string{}) assert.NoError(t, err) + assert.False(t, taskEmptyArgs.StartedAt.IsZero()) mockPool.ExpectExec("correct json").WithArgs("John", 30.0, nil).WillReturnResult(pgxmock.NewResult("EXECUTE", 0)) - err = pge.ExecuteSQLCommand(ctx, mockPool, &pgengine.ChainTask{Command: "correct json"}, []string{`["John", 30, null]`}) + taskWithArgs := &pgengine.ChainTask{Command: "correct json"} + err = pge.ExecuteSQLCommand(ctx, mockPool, taskWithArgs, []string{`["John", 30, null]`}) assert.NoError(t, err) + assert.False(t, taskWithArgs.StartedAt.IsZero()) mockPool.ExpectExec("incorrect json").WillReturnError(json.Unmarshal([]byte("foo"), &struct{}{})) err = pge.ExecuteSQLCommand(ctx, mockPool, &pgengine.ChainTask{Command: "incorrect json"}, []string{"foo"}) diff --git a/internal/pgengine/types.go b/internal/pgengine/types.go index dd3bcddd..3cc935f5 100644 --- a/internal/pgengine/types.go +++ b/internal/pgengine/types.go @@ -75,6 +75,16 @@ func (task *ChainTask) IsRemote() bool { return strings.TrimSpace(task.ConnectString) != "" } +// MarkStart records the moment command execution begins. +func (task *ChainTask) MarkStart() { + task.StartedAt = time.Now() +} + +// MarkDone records the command execution duration (in microseconds) since the last MarkStart. +func (task *ChainTask) MarkDone() { + task.Duration = time.Since(task.StartedAt).Microseconds() +} + // String returns a log-friendly identifier, e.g. "49|Check_if_file_exist". func (task ChainTask) String() string { return logIdent(task.TaskID, task.TaskName) diff --git a/internal/scheduler/chain.go b/internal/scheduler/chain.go index aa218f94..d59d59f4 100644 --- a/internal/scheduler/chain.go +++ b/internal/scheduler/chain.go @@ -307,7 +307,6 @@ func (sch *Scheduler) executeTask(ctx context.Context, tx pgx.Tx, task *pgengine defer cancel() } - task.StartedAt = time.Now() switch task.Kind { case "SQL": err = sch.pgengine.ExecuteSQLTask(ctx, tx, task, paramValues) @@ -320,7 +319,6 @@ func (sch *Scheduler) executeTask(ctx context.Context, tx pgx.Tx, task *pgengine case "BUILTIN": err = sch.executeBuiltinTask(ctx, task, paramValues) } - task.Duration = time.Since(task.StartedAt).Microseconds() returnCode := 0 if err != nil { returnCode = -1 diff --git a/internal/scheduler/shell.go b/internal/scheduler/shell.go index 4602f674..717699a3 100644 --- a/internal/scheduler/shell.go +++ b/internal/scheduler/shell.go @@ -45,6 +45,7 @@ func (sch *Scheduler) ExecuteProgramCommand(ctx context.Context, task *pgengine. return err } } + task.MarkStart() out, e := Cmd.CombinedOutput(ctx, command, params...) // #nosec if e != nil { exitCode = -1 @@ -53,6 +54,7 @@ func (sch *Scheduler) ExecuteProgramCommand(ctx context.Context, task *pgengine. exitCode = exitError.ExitCode() } } + task.MarkDone() sch.pgengine.LogTaskExecution(context.Background(), task, exitCode, string(out), val) } return err diff --git a/internal/scheduler/shell_test.go b/internal/scheduler/shell_test.go index a4da8c7a..4b6e60da 100644 --- a/internal/scheduler/shell_test.go +++ b/internal/scheduler/shell_test.go @@ -8,6 +8,7 @@ import ( "os/exec" "strings" "testing" + "time" "github.com/cybertec-postgresql/pg_timetable/internal/config" "github.com/cybertec-postgresql/pg_timetable/internal/log" @@ -18,6 +19,23 @@ import ( "github.com/stretchr/testify/assert" ) +func TestShellCommandDuration(t *testing.T) { + mock, err := pgxmock.NewPool() + assert.NoError(t, err) + pge := pgengine.NewDB(mock, "--log-database-level=none") + sch := scheduler.New(pge, log.Init(config.LoggingOpts{LogLevel: "panic", LogDBLevel: "none"}), otel.NewNoop()) + ctx := context.Background() + + task := &pgengine.ChainTask{Command: "sleep"} + + err = sch.ExecuteProgramCommand(ctx, task, []string{`["2"]`}) + assert.NoError(t, err, "sleep command should succeed") + + assert.False(t, task.StartedAt.IsZero()) + assert.GreaterOrEqual(t, time.Duration(task.Duration * int64(time.Microsecond)), 1 * time.Second) + assert.LessOrEqual(t, task.StartedAt, time.Now().Add(-2 * time.Second)) +} + type testCommander struct{} // overwrite CombinedOutput function of os/exec so only parameter syntax and return codes are checked... diff --git a/internal/scheduler/tasks.go b/internal/scheduler/tasks.go index 542210b3..998ab43a 100644 --- a/internal/scheduler/tasks.go +++ b/internal/scheduler/tasks.go @@ -37,12 +37,16 @@ func (sch *Scheduler) executeBuiltinTask(ctx context.Context, task *pgengine.Cha l := log.GetLogger(ctx) l.WithField("name", name).Debugf("Executing builtin task with parameters %+q", paramValues) if len(paramValues) == 0 { + task.MarkStart() stdout, err = f(ctx, sch, "") + task.MarkDone() sch.pgengine.LogTaskExecution(context.Background(), task, errCodes[err == nil], stdout, "") return err } for _, val := range paramValues { + task.MarkStart() stdout, err = f(ctx, sch, val) + task.MarkDone() sch.pgengine.LogTaskExecution(context.Background(), task, errCodes[err == nil], stdout, val) if err != nil { return diff --git a/internal/scheduler/tasks_test.go b/internal/scheduler/tasks_test.go index 1e982737..2834e181 100644 --- a/internal/scheduler/tasks_test.go +++ b/internal/scheduler/tasks_test.go @@ -3,6 +3,7 @@ package scheduler import ( "context" "testing" + "time" "github.com/cybertec-postgresql/pg_timetable/internal/config" "github.com/cybertec-postgresql/pg_timetable/internal/log" @@ -18,16 +19,24 @@ func TestExecuteTask(t *testing.T) { a.NoError(err) pge := pgengine.NewDB(mock, "--log-database-level=none") mocksch := New(pge, log.Init(config.LoggingOpts{LogLevel: "panic", LogDBLevel: "none"}), otel.NewNoop()) + task := &pgengine.ChainTask{} - et := func(task string, params []string) (err error) { - err = mocksch.executeBuiltinTask(context.TODO(), &pgengine.ChainTask{Command: task}, params) + et := func(command string, params []string) (err error) { + task.Command = command + err = mocksch.executeBuiltinTask(context.TODO(), task, params) return } - a.Error(et("foo", []string{})) + // Tests the len(param) == 0 case + a.NoError(et("NoOp", []string{})) + a.False(task.StartedAt.IsZero()) + a.NoError(et("Sleep", []string{"2"})) + a.GreaterOrEqual(time.Duration(task.Duration * int64(time.Microsecond)), 2 * time.Second) + a.LessOrEqual(task.StartedAt, time.Now().Add(-2 * time.Second)) + + a.Error(et("foo", []string{})) a.Error(et("Sleep", []string{"foo"})) - a.NoError(et("Sleep", []string{"1"})) a.NoError(et("NoOp", []string{})) a.NoError(et("NoOp", []string{"foo", "bar"}))