[*] run PROGRAM task for all parameters, fixes #741 (#742)

Previously program task returned on the first fail. This is not how
other kind of tasks behave. Make program task compatible with SQL and
built-in tasks. All errors will be joined and returned

Author: pashagolub

internal/scheduler/shell.go

@@ -28,7 +28,8 @@ var Cmd commander = realCommander{}
 
 // ExecuteProgramCommand executes program command and returns status code, output and error if any
 func (sch *Scheduler) ExecuteProgramCommand(ctx context.Context, task *pgengine.ChainTask, paramValues []string) error {
-
+	var err error
+	var exitCode int
 	command := strings.TrimSpace(task.Command)
 	if command == "" {
 		return errors.New("program command cannot be empty")
@@ -37,24 +38,22 @@ func (sch *Scheduler) ExecuteProgramCommand(ctx context.Context, task *pgengine.
 		paramValues = []string{""}
 	}
 	for _, val := range paramValues {
+		exitCode = 0
 		params := []string{}
 		if val > "" {
 			if err := json.Unmarshal([]byte(val), &params); err != nil {
 				return err
 			}
 		}
-		out, err := Cmd.CombinedOutput(ctx, command, params...) // #nosec
-		if err != nil {
-			//check if we're dealing with an ExitError - i.e. return code other than 0
+		out, e := Cmd.CombinedOutput(ctx, command, params...) // #nosec
+		if e != nil {
+			exitCode = -1
+			err = errors.Join(err, e) // accumulate errors for all param sets
 			if exitError, ok := err.(*exec.ExitError); ok {
-				exitCode := exitError.ExitCode()
-				sch.pgengine.LogTaskExecution(context.Background(), task, exitCode, string(out), val)
-				return exitError
+				exitCode = exitError.ExitCode()
 			}
-			sch.pgengine.LogTaskExecution(context.Background(), task, -1, string(out), val)
-			return err
 		}
-		sch.pgengine.LogTaskExecution(context.Background(), task, 0, string(out), val)
+		sch.pgengine.LogTaskExecution(context.Background(), task, exitCode, string(out), val)
 	}
-	return nil
+	return err
 }

internal/scheduler/shell_test.go

@@ -3,6 +3,7 @@ package scheduler_test
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os/exec"
 	"strings"
@@ -32,37 +33,37 @@ func TestShellCommand(t *testing.T) {
 
 	mock, err := pgxmock.NewPool() //
 	assert.NoError(t, err)
-	pge := pgengine.NewDB(mock, "scheduler_unit_test")
-	scheduler := scheduler.New(pge, log.Init(config.LoggingOpts{LogLevel: "panic", LogDBLevel: "none"}))
+	pge := pgengine.NewDB(mock, "--log-database-level=none")
+	sch := scheduler.New(pge, log.Init(config.LoggingOpts{LogLevel: "panic", LogDBLevel: "none"}))
 	ctx := context.Background()
 
-	err = scheduler.ExecuteProgramCommand(ctx, &pgengine.ChainTask{}, []string{""})
+	err = sch.ExecuteProgramCommand(ctx, &pgengine.ChainTask{}, []string{""})
 	assert.EqualError(t, err, "program command cannot be empty", "Empty command should out, fail")
 
-	err = scheduler.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping0"}, nil)
+	err = sch.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping0"}, nil)
 	assert.NoError(t, err, "Command with nil param is out, OK")
 
-	err = scheduler.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping1"}, []string{})
+	err = sch.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping1"}, []string{})
 	assert.NoError(t, err, "Command with empty array param is OK")
 
-	err = scheduler.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping2"}, []string{""})
+	err = sch.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping2"}, []string{""})
 	assert.NoError(t, err, "Command with empty string param is OK")
 
-	err = scheduler.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping3"}, []string{"[]"})
+	err = sch.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping3"}, []string{"[]"})
 	assert.NoError(t, err, "Command with empty json array param is OK")
 
-	err = scheduler.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping3"}, []string{"[null]"})
+	err = sch.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping3"}, []string{"[null]"})
 	assert.NoError(t, err, "Command with nil array param is OK")
 
-	err = scheduler.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping4"}, []string{`["localhost"]`})
+	err = sch.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping4"}, []string{`["localhost"]`})
 	assert.NoError(t, err, "Command with one param is OK")
 
-	err = scheduler.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping5"}, []string{`["localhost", "-4"]`})
+	err = sch.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping5"}, []string{`["localhost", "-4"]`})
 	assert.NoError(t, err, "Command with many params is OK")
 
-	err = scheduler.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "pong"}, nil)
-	assert.IsType(t, (*exec.Error)(nil), err, "Uknown command should produce error")
+	err = sch.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "pong"}, nil)
+	assert.True(t, errors.Is(err, exec.ErrNotFound), "Unknown command should produce exec.Error")
 
-	err = scheduler.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping5"}, []string{`{"param1": "localhost"}`})
-	assert.IsType(t, (*json.UnmarshalTypeError)(nil), err, "Command should fail with mailformed json parameter")
+	err = sch.ExecuteProgramCommand(ctx, &pgengine.ChainTask{Command: "ping5"}, []string{`{"param1": "localhost"}`})
+	assert.IsType(t, (*json.UnmarshalTypeError)(nil), err, "Command should fail with malformed json parameter")
 }

internal/scheduler/tasks_test.go

@@ -14,7 +14,7 @@ import (
 func TestExecuteTask(t *testing.T) {
 	mock, err := pgxmock.NewPool() //
 	assert.NoError(t, err)
-	pge := pgengine.NewDB(mock, "scheduler_unit_test")
+	pge := pgengine.NewDB(mock, "--log-database-level=none")
 	mocksch := New(pge, log.Init(config.LoggingOpts{LogLevel: "panic", LogDBLevel: "none"}))
 
 	et := func(task string, params []string) (err error) {