feat: change the Test Workflow orchestration mechanism to atomic instructions

cmd/kubectl-testkube/commands/testworkflows/run.go

@@ -13,7 +13,7 @@ import (
 	"github.com/kubeshop/testkube/cmd/kubectl-testkube/commands/common"
 	"github.com/kubeshop/testkube/cmd/kubectl-testkube/commands/common/render"
 	"github.com/kubeshop/testkube/cmd/kubectl-testkube/commands/testworkflows/renderer"
-	"github.com/kubeshop/testkube/cmd/testworkflow-init/data"
+	"github.com/kubeshop/testkube/cmd/testworkflow-init/instructions"
 	apiclientv1 "github.com/kubeshop/testkube/pkg/api/v1/client"
 	"github.com/kubeshop/testkube/pkg/api/v1/testkube"
 	"github.com/kubeshop/testkube/pkg/testworkflows/testworkflowprocessor/constants"
@@ -304,7 +304,7 @@ func printRawLogLines(logs []byte, steps []testkube.TestWorkflowSignature, resul
 			line = line[getTimestampLength(line)+1:]
 		}
 
-		start := data.StartHintRe.FindStringSubmatch(line)
+		start := instructions.StartHintRe.FindStringSubmatch(line)
 		if len(start) == 0 {
 			line += "\x07"
 			fmt.Println(line)

cmd/tcl/testworkflow-toolkit/commands/execute.go

@@ -22,6 +22,7 @@ import (
 	commontcl "github.com/kubeshop/testkube/cmd/tcl/testworkflow-toolkit/common"
 	"github.com/kubeshop/testkube/cmd/tcl/testworkflow-toolkit/spawn"
 	"github.com/kubeshop/testkube/cmd/testworkflow-init/data"
+	"github.com/kubeshop/testkube/cmd/testworkflow-init/instructions"
 	"github.com/kubeshop/testkube/cmd/testworkflow-toolkit/env"
 	"github.com/kubeshop/testkube/cmd/testworkflow-toolkit/transfer"
 	"github.com/kubeshop/testkube/internal/common"
@@ -92,7 +93,7 @@ func buildTestExecution(test testworkflowsv1.StepExecuteTest, async bool) (func(
 			return
 		}
 
-		data.PrintOutput(env.Ref(), "test-start", &testExecutionDetails{
+		instructions.PrintOutput(env.Ref(), "test-start", &testExecutionDetails{
 			Id:          exec.Id,
 			Name:        exec.Name,
 			TestName:    exec.TestName,
@@ -126,7 +127,7 @@ func buildTestExecution(test testworkflowsv1.StepExecuteTest, async bool) (func(
 					break loop
 				}
 				if prevStatus != status {
-					data.PrintOutput(env.Ref(), "test-status", &executionResult{Id: exec.Id, Status: string(status)})
+					instructions.PrintOutput(env.Ref(), "test-status", &executionResult{Id: exec.Id, Status: string(status)})
 				}
 				prevStatus = status
 			}
@@ -140,7 +141,7 @@ func buildTestExecution(test testworkflowsv1.StepExecuteTest, async bool) (func(
 			color = ui.Red
 		}
 
-		data.PrintOutput(env.Ref(), "test-end", &executionResult{Id: exec.Id, Status: string(status)})
+		instructions.PrintOutput(env.Ref(), "test-end", &executionResult{Id: exec.Id, Status: string(status)})
 		fmt.Printf("%s • %s\n", color(execName), string(status))
 		return
 	}, nil
@@ -161,7 +162,7 @@ func buildWorkflowExecution(workflow testworkflowsv1.StepExecuteWorkflow, async
 			return
 		}
 
-		data.PrintOutput(env.Ref(), "testworkflow-start", &testWorkflowExecutionDetails{
+		instructions.PrintOutput(env.Ref(), "testworkflow-start", &testWorkflowExecutionDetails{
 			Id:               exec.Id,
 			Name:             exec.Name,
 			TestWorkflowName: exec.Workflow.Name,
@@ -195,7 +196,7 @@ func buildWorkflowExecution(workflow testworkflowsv1.StepExecuteWorkflow, async
 					break loop
 				}
 				if prevStatus != status {
-					data.PrintOutput(env.Ref(), "testworkflow-status", &executionResult{Id: exec.Id, Status: string(status)})
+					instructions.PrintOutput(env.Ref(), "testworkflow-status", &executionResult{Id: exec.Id, Status: string(status)})
 				}
 				prevStatus = status
 			}
@@ -209,7 +210,7 @@ func buildWorkflowExecution(workflow testworkflowsv1.StepExecuteWorkflow, async
 			color = ui.Red
 		}
 
-		data.PrintOutput(env.Ref(), "testworkflow-end", &executionResult{Id: exec.Id, Status: string(status)})
+		instructions.PrintOutput(env.Ref(), "testworkflow-end", &executionResult{Id: exec.Id, Status: string(status)})
 		fmt.Printf("%s • %s\n", color(execName), string(status))
 		return
 	}, nil

cmd/tcl/testworkflow-toolkit/commands/kill.go

@@ -19,6 +19,7 @@ import (
 	commontcl "github.com/kubeshop/testkube/cmd/tcl/testworkflow-toolkit/common"
 	"github.com/kubeshop/testkube/cmd/tcl/testworkflow-toolkit/spawn"
 	"github.com/kubeshop/testkube/cmd/testworkflow-init/data"
+	"github.com/kubeshop/testkube/cmd/testworkflow-init/instructions"
 	"github.com/kubeshop/testkube/cmd/testworkflow-toolkit/artifacts"
 	"github.com/kubeshop/testkube/cmd/testworkflow-toolkit/env"
 	"github.com/kubeshop/testkube/pkg/expressions"
@@ -104,7 +105,7 @@ func NewKillCmd() *cobra.Command {
 
 					logsFilePath, err := spawn.SaveLogs(context.Background(), clientSet, storage, env.Namespace(), id, service+"/", index)
 					if err == nil {
-						data.PrintOutput(env.Ref(), "service", ServiceInfo{Group: groupRef, Name: service, Index: index, Logs: storage.FullPath(logsFilePath)})
+						instructions.PrintOutput(env.Ref(), "service", ServiceInfo{Group: groupRef, Name: service, Index: index, Logs: storage.FullPath(logsFilePath)})
 						log("saved logs")
 					} else {
 						log("warning", "problem saving the logs", err.Error())

cmd/tcl/testworkflow-toolkit/commands/parallel.go

@@ -23,17 +23,17 @@ import (
 	testworkflowsv1 "github.com/kubeshop/testkube-operator/api/testworkflows/v1"
 	commontcl "github.com/kubeshop/testkube/cmd/tcl/testworkflow-toolkit/common"
 	"github.com/kubeshop/testkube/cmd/tcl/testworkflow-toolkit/spawn"
-	"github.com/kubeshop/testkube/cmd/testworkflow-init/data"
+	"github.com/kubeshop/testkube/cmd/testworkflow-init/instructions"
 	"github.com/kubeshop/testkube/cmd/testworkflow-toolkit/artifacts"
 	"github.com/kubeshop/testkube/cmd/testworkflow-toolkit/env"
 	"github.com/kubeshop/testkube/cmd/testworkflow-toolkit/transfer"
 	"github.com/kubeshop/testkube/internal/common"
 	"github.com/kubeshop/testkube/pkg/api/v1/testkube"
 	"github.com/kubeshop/testkube/pkg/expressions"
 	"github.com/kubeshop/testkube/pkg/testworkflows/testworkflowcontroller"
-	"github.com/kubeshop/testkube/pkg/testworkflows/testworkflowprocessor"
 	"github.com/kubeshop/testkube/pkg/testworkflows/testworkflowprocessor/constants"
 	"github.com/kubeshop/testkube/pkg/testworkflows/testworkflowprocessor/presets"
+	"github.com/kubeshop/testkube/pkg/testworkflows/testworkflowprocessor/stage"
 	"github.com/kubeshop/testkube/pkg/ui"
 )
 
@@ -163,7 +163,7 @@ func NewParallelCmd() *cobra.Command {
 
 			// Send initial output
 			for index := range specs {
-				data.PrintOutput(env.Ref(), "parallel", ParallelStatus{
+				instructions.PrintOutput(env.Ref(), "parallel", ParallelStatus{
 					Index:       index,
 					Description: descriptions[index],
 				})
@@ -200,7 +200,7 @@ func NewParallelCmd() *cobra.Command {
 				}
 
 				// Compute the bundle instructions
-				sig := testworkflowprocessor.MapSignatureListToInternal(bundle.Signature)
+				sig := stage.MapSignatureListToInternal(bundle.Signature)
 				namespace := bundle.Job.Namespace
 				if namespace == "" {
 					namespace = env.Namespace()
@@ -228,7 +228,7 @@ func NewParallelCmd() *cobra.Command {
 					if shouldSaveLogs {
 						logsFilePath, err := spawn.SaveLogs(context.Background(), clientSet, storage, namespace, id, "", index)
 						if err == nil {
-							data.PrintOutput(env.Ref(), "parallel", ParallelStatus{Index: int(index), Logs: storage.FullPath(logsFilePath)})
+							instructions.PrintOutput(env.Ref(), "parallel", ParallelStatus{Index: int(index), Logs: storage.FullPath(logsFilePath)})
 							log("saved logs")
 						} else {
 							log("warning", "problem saving the logs", err.Error())
@@ -246,7 +246,7 @@ func NewParallelCmd() *cobra.Command {
 				}()
 
 				// Inform about the step structure
-				data.PrintOutput(env.Ref(), "parallel", ParallelStatus{Index: int(index), Signature: sig})
+				instructions.PrintOutput(env.Ref(), "parallel", ParallelStatus{Index: int(index), Signature: sig})
 
 				// Control the execution
 				// TODO: Consider aggregated controller to limit number of watchers
@@ -291,11 +291,11 @@ func NewParallelCmd() *cobra.Command {
 						prevStep = v.Current
 						prevStatus = v.Status
 						if v.Result.IsFinished() {
-							data.PrintOutput(env.Ref(), "parallel", ParallelStatus{Index: int(index), Status: v.Status, Result: v.Result})
+							instructions.PrintOutput(env.Ref(), "parallel", ParallelStatus{Index: int(index), Status: v.Status, Result: v.Result})
 							ctxCancel()
 							return v.Result.IsPassed()
 						} else {
-							data.PrintOutput(env.Ref(), "parallel", ParallelStatus{Index: int(index), Status: v.Status, Current: v.Current})
+							instructions.PrintOutput(env.Ref(), "parallel", ParallelStatus{Index: int(index), Status: v.Status, Current: v.Current})
 						}
 					}
 				}

cmd/tcl/testworkflow-toolkit/commands/services.go

@@ -24,7 +24,7 @@ import (
 	testworkflowsv1 "github.com/kubeshop/testkube-operator/api/testworkflows/v1"
 	commontcl "github.com/kubeshop/testkube/cmd/tcl/testworkflow-toolkit/common"
 	"github.com/kubeshop/testkube/cmd/tcl/testworkflow-toolkit/spawn"
-	"github.com/kubeshop/testkube/cmd/testworkflow-init/data"
+	"github.com/kubeshop/testkube/cmd/testworkflow-init/instructions"
 	"github.com/kubeshop/testkube/cmd/testworkflow-toolkit/env"
 	"github.com/kubeshop/testkube/cmd/testworkflow-toolkit/transfer"
 	"github.com/kubeshop/testkube/internal/common"
@@ -111,7 +111,7 @@ func NewServicesCmd() *cobra.Command {
 				}
 
 				// Initialize empty array of details for each of the services
-				data.PrintHintDetails(env.Ref(), fmt.Sprintf("services.%s", name), []ServiceState{})
+				instructions.PrintHintDetails(env.Ref(), fmt.Sprintf("services.%s", name), []ServiceState{})
 			}
 
 			// Analyze instances to run
@@ -189,12 +189,12 @@ func NewServicesCmd() *cobra.Command {
 				for i := range svcInstances {
 					state[name][i].Description = svcInstances[i].Description
 				}
-				data.PrintHintDetails(env.Ref(), fmt.Sprintf("services.%s", name), state)
+				instructions.PrintHintDetails(env.Ref(), fmt.Sprintf("services.%s", name), state)
 			}
 
 			// Inform about each service instance
 			for _, instance := range instances {
-				data.PrintOutput(env.Ref(), "service", ServiceInfo{
+				instructions.PrintOutput(env.Ref(), "service", ServiceInfo{
 					Group:       groupRef,
 					Index:       instance.Index,
 					Name:        instance.Name,
@@ -274,7 +274,8 @@ func NewServicesCmd() *cobra.Command {
 				if namespace == "" {
 					namespace = env.Namespace()
 				}
-				mainRef := bundle.Job.Spec.Template.Spec.Containers[0].Name
+
+				mainRef := bundle.Actions().GetLastRef()
 
 				// Deploy the resources
 				// TODO: Avoid using Job
@@ -329,7 +330,7 @@ func NewServicesCmd() *cobra.Command {
 						state[instance.Name][index].Ip = v.PodIP
 						log(fmt.Sprintf("assigned to %s IP", ui.LightBlue(v.PodIP)))
 						info.Status = ServiceStatusRunning
-						data.PrintOutput(env.Ref(), "service", info)
+						instructions.PrintOutput(env.Ref(), "service", info)
 					}
 
 					if v.Current == mainRef {
@@ -349,7 +350,7 @@ func NewServicesCmd() *cobra.Command {
 				if !started {
 					info.Status = ServiceStatusFailed
 					log("container failed")
-					data.PrintOutput(env.Ref(), "service", info)
+					instructions.PrintOutput(env.Ref(), "service", info)
 					return false
 				}
 
@@ -377,7 +378,7 @@ func NewServicesCmd() *cobra.Command {
 					log("container ready")
 					info.Status = ServiceStatusReady
 				}
-				data.PrintOutput(env.Ref(), "service", info)
+				instructions.PrintOutput(env.Ref(), "service", info)
 
 				return ready
 			}
@@ -387,7 +388,7 @@ func NewServicesCmd() *cobra.Command {
 
 			// Inform about the services state
 			for k := range state {
-				data.PrintHintDetails(env.Ref(), fmt.Sprintf("services.%s", k), state[k])
+				instructions.PrintHintDetails(env.Ref(), fmt.Sprintf("services.%s", k), state[k])
 			}
 
 			// Notify the results

cmd/tcl/testworkflow-toolkit/spawn/utils.go

@@ -33,8 +33,8 @@ import (
 	"github.com/kubeshop/testkube/pkg/api/v1/testkube"
 	"github.com/kubeshop/testkube/pkg/expressions"
 	"github.com/kubeshop/testkube/pkg/testworkflows/testworkflowcontroller"
-	"github.com/kubeshop/testkube/pkg/testworkflows/testworkflowprocessor"
 	"github.com/kubeshop/testkube/pkg/testworkflows/testworkflowprocessor/constants"
+	"github.com/kubeshop/testkube/pkg/testworkflows/testworkflowprocessor/stage"
 )
 
 func MapDynamicListToStringList(list []interface{}) []string {
@@ -174,7 +174,7 @@ func ProcessFetch(transferSrv transfer.Server, fetch []testworkflowsv1.StepParal
 					Env: []corev1.EnvVar{
 						{Name: "TK_NS", Value: env.Namespace()},
 						{Name: "TK_REF", Value: env.Ref()},
-						testworkflowprocessor.BypassToolkitCheck,
+						stage.BypassToolkitCheck,
 					},
 					Args: &result,
 				},

cmd/testworkflow-init/commands/run.go

@@ -0,0 +1,74 @@
+package commands
+
+import (
+	"slices"
+
+	"github.com/kubeshop/testkube/cmd/testworkflow-init/constants"
+	"github.com/kubeshop/testkube/cmd/testworkflow-init/data"
+	"github.com/kubeshop/testkube/cmd/testworkflow-init/orchestration"
+	"github.com/kubeshop/testkube/cmd/testworkflow-init/output"
+	"github.com/kubeshop/testkube/pkg/expressions"
+	"github.com/kubeshop/testkube/pkg/testworkflows/testworkflowprocessor/action/actiontypes/lite"
+)
+
+func Run(run lite.ActionExecute, container lite.LiteActionContainer) {
+	machine := data.GetInternalTestWorkflowMachine()
+	state := data.GetState()
+	step := state.GetStep(run.Ref)
+
+	// Abandon executing if the step was finished before
+	if step.IsFinished() {
+		return
+	}
+
+	// Obtain command to run
+	command := make([]string, 0)
+	if container.Config.Command != nil {
+		command = slices.Clone(*container.Config.Command)
+	}
+	if container.Config.Args != nil {
+		command = append(command, *container.Config.Args...)
+	}
+
+	// Resolve the command to run
+	for i := range command {
+		value, err := expressions.CompileAndResolveTemplate(command[i], machine, expressions.FinalizerFail)
+		if err != nil {
+			output.ExitErrorf(data.CodeInternal, "failed to compute argument '%d': %s", i, err.Error())
+		}
+		command[i], _ = value.Static().StringValue()
+	}
+
+	// Run the operation
+	execution := orchestration.Executions.Create(command[0], command[1:])
+	result, err := execution.Run()
+	if err != nil {
+		output.ExitErrorf(data.CodeInternal, "failed to execute: %v", err)
+	}
+
+	// Initialize local state
+	var status data.StepStatus
+
+	success := result.ExitCode == 0
+
+	// Compute the result
+	if run.Negative {
+		success = !success
+	}
+	if result.Aborted {
+		status = data.StepStatusAborted
+	} else if success {
+		status = data.StepStatusPassed
+	} else {
+		status = data.StepStatusFailed
+	}
+
+	// Abandon saving execution data if the step has been finished before
+	if step.IsFinished() {
+		return
+	}
+
+	// Notify about the status
+	step.SetStatus(status).SetExitCode(result.ExitCode)
+	orchestration.FinishExecution(step, constants.ExecutionResult{ExitCode: result.ExitCode, Iteration: int(step.Iteration)})
+}