feat: add options to control Test Workflows orchestration

api/v1/testkube.yaml

@@ -8478,6 +8478,14 @@ components:
       required:
       - from
 
+    TestWorkflowSystem:
+      type: object
+      properties:
+        pureByDefault:
+          $ref: "#/components/schemas/BoxedBoolean"
+        isolatedContainers:
+          $ref: "#/components/schemas/BoxedBoolean"
+
     TestWorkflowSpec:
       type: object
       properties:
@@ -8487,6 +8495,8 @@ components:
             $ref: "#/components/schemas/TestWorkflowTemplateRef"
         config:
           $ref: "#/components/schemas/TestWorkflowConfigSchema"
+        system:
+          $ref: "#/components/schemas/TestWorkflowSystem"
         content:
           $ref: "#/components/schemas/TestWorkflowContent"
         services:
@@ -8521,6 +8531,8 @@ components:
       properties:
         config:
           $ref: "#/components/schemas/TestWorkflowConfigSchema"
+        system:
+          $ref: "#/components/schemas/TestWorkflowSystem"
         content:
           $ref: "#/components/schemas/TestWorkflowContent"
         services:
@@ -8595,6 +8607,8 @@ components:
         condition:
           type: string
           description: expression to declare under which conditions the step should be run; defaults to "passed", except artifacts where it defaults to "always"
+        pure:
+          $ref: "#/components/schemas/BoxedBoolean"
         paused:
           type: boolean
           description: should the step be paused initially
@@ -8655,6 +8669,8 @@ components:
         condition:
           type: string
           description: expression to declare under which conditions the step should be run; defaults to "passed", except artifacts where it defaults to "always"
+        pure:
+          $ref: "#/components/schemas/BoxedBoolean"
         paused:
           type: boolean
           description: should the step be paused initially

cmd/testworkflow-init/orchestration/executions.go

@@ -177,6 +177,7 @@ func (e *execution) Run() (*executionResult, error) {
 		e.group.pauseMu.Unlock()
 		e.cmdMu.Unlock()
 		_, exitCode = getProcessStatus(err)
+		e.cmd.Stderr.Write(append([]byte(err.Error()), '\n'))
 	}
 
 	// Clean up

go.mod

@@ -34,7 +34,7 @@ require (
 	github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/kubepug/kubepug v1.7.1
-	github.com/kubeshop/testkube-operator v1.15.2-beta1.0.20240729112855-d7bd9c5f64a9
+	github.com/kubeshop/testkube-operator v1.15.2-beta1.0.20240808123612-1bfae781ec4c
 	github.com/minio/minio-go/v7 v7.0.47
 	github.com/montanaflynn/stats v0.6.6
 	github.com/moogar0880/problems v0.1.1

go.sum

@@ -371,8 +371,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kubepug/kubepug v1.7.1 h1:LKhfSxS8Y5mXs50v+3Lpyec+cogErDLcV7CMUuiaisw=
 github.com/kubepug/kubepug v1.7.1/go.mod h1:lv+HxD0oTFL7ZWjj0u6HKhMbbTIId3eG7aWIW0gyF8g=
-github.com/kubeshop/testkube-operator v1.15.2-beta1.0.20240729112855-d7bd9c5f64a9 h1:Zg+hTPkHi504Nwp3M9kM9+DPX7OdiYJBJL2hJCE5Jhg=
-github.com/kubeshop/testkube-operator v1.15.2-beta1.0.20240729112855-d7bd9c5f64a9/go.mod h1:P47tw1nKQFufdsZndyq2HG2MSa0zK/lU0XpRfZtEmIk=
+github.com/kubeshop/testkube-operator v1.15.2-beta1.0.20240808123612-1bfae781ec4c h1:lMNmYvazHkx+yDxxbNUg0Vfk+5tvexLwhUGZAZ9mM6Y=
+github.com/kubeshop/testkube-operator v1.15.2-beta1.0.20240808123612-1bfae781ec4c/go.mod h1:P47tw1nKQFufdsZndyq2HG2MSa0zK/lU0XpRfZtEmIk=
 github.com/leodido/go-urn v1.2.1 h1:BqpAaACuzVSgi/VLzGZIobT2z4v53pjosyNd9Yv6n/w=
 github.com/leodido/go-urn v1.2.1/go.mod h1:zt4jvISO2HfUBqxjfIshjdMTYS56ZS/qv49ictyFfxY=
 github.com/lithammer/fuzzysearch v1.1.8 h1:/HIuJnjHuXS8bKaiTMeeDlW2/AyIWk2brx1V8LFgLN4=

pkg/api/v1/testkube/model_test_workflow_independent_service_spec.go

@@ -25,12 +25,6 @@ type TestWorkflowIndependentServiceSpec struct {
 	SecurityContext *SecurityContext       `json:"securityContext,omitempty"`
 	// volumes to mount to the container
 	VolumeMounts []VolumeMount `json:"volumeMounts,omitempty"`
-	Count        *BoxedString  `json:"count,omitempty"`
-	MaxCount     *BoxedString  `json:"maxCount,omitempty"`
-	// matrix of parameters to spawn instances
-	Matrix map[string]interface{} `json:"matrix,omitempty"`
-	// parameters that should be distributed across sharded instances
-	Shards map[string]interface{} `json:"shards,omitempty"`
 	// service description to display
 	Description string `json:"description,omitempty"`
 	// maximum time until reaching readiness
@@ -42,4 +36,10 @@ type TestWorkflowIndependentServiceSpec struct {
 	Logs           *BoxedString                       `json:"logs,omitempty"`
 	RestartPolicy  string                             `json:"restartPolicy,omitempty"`
 	ReadinessProbe *Probe                             `json:"readinessProbe,omitempty"`
+	Count          *BoxedString                       `json:"count,omitempty"`
+	MaxCount       *BoxedString                       `json:"maxCount,omitempty"`
+	// matrix of parameters to spawn instances
+	Matrix map[string]interface{} `json:"matrix,omitempty"`
+	// parameters that should be distributed across sharded instances
+	Shards map[string]interface{} `json:"shards,omitempty"`
 }

pkg/api/v1/testkube/model_test_workflow_independent_step.go

@@ -13,7 +13,8 @@ type TestWorkflowIndependentStep struct {
 	// readable name for the step
 	Name string `json:"name,omitempty"`
 	// expression to declare under which conditions the step should be run; defaults to \"passed\", except artifacts where it defaults to \"always\"
-	Condition string `json:"condition,omitempty"`
+	Condition string        `json:"condition,omitempty"`
+	Pure      *BoxedBoolean `json:"pure,omitempty"`
 	// should the step be paused initially
 	Paused bool `json:"paused,omitempty"`
 	// is the step expected to fail

pkg/api/v1/testkube/model_test_workflow_independent_step_parallel.go

@@ -28,11 +28,21 @@ type TestWorkflowIndependentStepParallel struct {
 	// delay before the step
 	Delay string `json:"delay,omitempty"`
 	// script to run in a default shell for the container
-	Shell     string                                        `json:"shell,omitempty"`
-	Run       *TestWorkflowStepRun                          `json:"run,omitempty"`
-	Execute   *TestWorkflowStepExecute                      `json:"execute,omitempty"`
-	Artifacts *TestWorkflowStepArtifacts                    `json:"artifacts,omitempty"`
+	Shell     string                     `json:"shell,omitempty"`
+	Run       *TestWorkflowStepRun       `json:"run,omitempty"`
+	Execute   *TestWorkflowStepExecute   `json:"execute,omitempty"`
+	Artifacts *TestWorkflowStepArtifacts `json:"artifacts,omitempty"`
+	// how many resources could be scheduled in parallel
+	Parallelism int32 `json:"parallelism,omitempty"`
+	// worker description to display
+	Description string       `json:"description,omitempty"`
+	Logs        *BoxedString `json:"logs,omitempty"`
+	// list of files to send to parallel steps
+	Transfer []TestWorkflowStepParallelTransfer `json:"transfer,omitempty"`
+	// list of files to fetch from parallel steps
+	Fetch     []TestWorkflowStepParallelFetch               `json:"fetch,omitempty"`
 	Config    map[string]TestWorkflowParameterSchema        `json:"config,omitempty"`
+	System    *TestWorkflowSystem                           `json:"system,omitempty"`
 	Content   *TestWorkflowContent                          `json:"content,omitempty"`
 	Services  map[string]TestWorkflowIndependentServiceSpec `json:"services,omitempty"`
 	Container *TestWorkflowContainerConfig                  `json:"container,omitempty"`
@@ -42,13 +52,4 @@ type TestWorkflowIndependentStepParallel struct {
 	Steps     []TestWorkflowIndependentStep                 `json:"steps,omitempty"`
 	After     []TestWorkflowIndependentStep                 `json:"after,omitempty"`
 	Events    []TestWorkflowEvent                           `json:"events,omitempty"`
-	// how many resources could be scheduled in parallel
-	Parallelism int32 `json:"parallelism,omitempty"`
-	// worker description to display
-	Description string       `json:"description,omitempty"`
-	Logs        *BoxedString `json:"logs,omitempty"`
-	// list of files to send to parallel steps
-	Transfer []TestWorkflowStepParallelTransfer `json:"transfer,omitempty"`
-	// list of files to fetch from parallel steps
-	Fetch []TestWorkflowStepParallelFetch `json:"fetch,omitempty"`
 }

pkg/api/v1/testkube/model_test_workflow_result_extended.go

@@ -157,7 +157,46 @@ func (r *TestWorkflowResult) UpdateStepResult(sig []TestWorkflowSignature, ref s
 func (r *TestWorkflowResult) RecomputeDuration() {
 	if !r.FinishedAt.IsZero() {
 		r.PausedMs = 0
+
+		// Get unique pause periods
+		pauses := make([]TestWorkflowPause, 0)
+	loop:
 		for _, p := range r.Pauses {
+			// Finalize the pause if it's not
+			step := r.Steps[p.Ref]
+			if !step.FinishedAt.IsZero() && p.ResumedAt.IsZero() {
+				p.ResumedAt = step.FinishedAt
+			}
+
+			for i := range pauses {
+				// They don't overlap
+				if p.PausedAt.After(pauses[i].ResumedAt) || p.ResumedAt.Before(pauses[i].PausedAt) {
+					continue
+				}
+
+				// The existing pause period may take some period before
+				if pauses[i].PausedAt.After(p.PausedAt) {
+					pauses[i].PausedAt = p.PausedAt
+					p.PausedAt = pauses[i].ResumedAt
+					if p.ResumedAt.Before(p.PausedAt) {
+						p.ResumedAt = p.PausedAt
+					}
+				}
+
+				// The existing pause period may take some period after
+				if pauses[i].ResumedAt.Before(p.ResumedAt) {
+					pauses[i].ResumedAt = p.ResumedAt
+					p.ResumedAt = pauses[i].PausedAt
+				}
+
+				// The pause is already enclosed in existing list
+				continue loop
+			}
+
+			pauses = append(pauses, p)
+		}
+
+		for _, p := range pauses {
 			resumedAt := p.ResumedAt
 			if resumedAt.IsZero() {
 				resumedAt = r.FinishedAt
@@ -167,6 +206,7 @@ func (r *TestWorkflowResult) RecomputeDuration() {
 				r.PausedMs += milli
 			}
 		}
+
 		totalDuration := r.FinishedAt.Sub(r.QueuedAt)
 		duration := totalDuration - time.Duration(1e3*r.PausedMs)
 		if totalDuration < 0 {
@@ -451,7 +491,7 @@ func adjustMinimumTime(dst, min time.Time) time.Time {
 func predictTestWorkflowStepStatus(v TestWorkflowStepResult, sig TestWorkflowSignature, r *TestWorkflowResult) (TestWorkflowStepStatus, bool) {
 	children := sig.Children
 	if len(children) == 0 {
-		if getTestWorkflowStepStatus(v) == QUEUED_TestWorkflowStepStatus || getTestWorkflowStepStatus(v) == RUNNING_TestWorkflowStepStatus || getTestWorkflowStepStatus(v) == PAUSED_TestWorkflowStepStatus {
+		if !getTestWorkflowStepStatus(v).Finished() {
 			return PASSED_TestWorkflowStepStatus, false
 		}
 		return *v.Status, true
@@ -473,7 +513,7 @@ func predictTestWorkflowStepStatus(v TestWorkflowStepResult, sig TestWorkflowSig
 		if !ch.Optional && (status == FAILED_TestWorkflowStepStatus || status == TIMEOUT_TestWorkflowStepStatus) {
 			failed = true
 		}
-		if status == QUEUED_TestWorkflowStepStatus || status == RUNNING_TestWorkflowStepStatus || status == PAUSED_TestWorkflowStepStatus {
+		if !status.Finished() {
 			finished = false
 		}
 	}
@@ -530,7 +570,7 @@ func recomputeTestWorkflowStepResult(v TestWorkflowStepResult, sig TestWorkflowS
 	// It is finished already
 	if !v.FinishedAt.IsZero() {
 		predicted, finished := predictTestWorkflowStepStatus(v, sig, r)
-		if finished {
+		if finished && (v.Status == nil || !(*v.Status).Finished()) {
 			v.Status = common.Ptr(predicted)
 		}
 		return v

pkg/api/v1/testkube/model_test_workflow_service_spec.go

@@ -36,11 +36,11 @@ type TestWorkflowServiceSpec struct {
 	Logs           *BoxedString                       `json:"logs,omitempty"`
 	RestartPolicy  string                             `json:"restartPolicy,omitempty"`
 	ReadinessProbe *Probe                             `json:"readinessProbe,omitempty"`
+	Use            []TestWorkflowTemplateRef          `json:"use,omitempty"`
 	Count          *BoxedString                       `json:"count,omitempty"`
 	MaxCount       *BoxedString                       `json:"maxCount,omitempty"`
 	// matrix of parameters to spawn instances
 	Matrix map[string]interface{} `json:"matrix,omitempty"`
 	// parameters that should be distributed across sharded instances
-	Shards map[string]interface{}    `json:"shards,omitempty"`
-	Use    []TestWorkflowTemplateRef `json:"use,omitempty"`
+	Shards map[string]interface{} `json:"shards,omitempty"`
 }

pkg/api/v1/testkube/model_test_workflow_spec.go

@@ -12,6 +12,7 @@ package testkube
 type TestWorkflowSpec struct {
 	Use       []TestWorkflowTemplateRef              `json:"use,omitempty"`
 	Config    map[string]TestWorkflowParameterSchema `json:"config,omitempty"`
+	System    *TestWorkflowSystem                    `json:"system,omitempty"`
 	Content   *TestWorkflowContent                   `json:"content,omitempty"`
 	Services  map[string]TestWorkflowServiceSpec     `json:"services,omitempty"`
 	Container *TestWorkflowContainerConfig           `json:"container,omitempty"`

pkg/api/v1/testkube/model_test_workflow_step.go

@@ -13,7 +13,8 @@ type TestWorkflowStep struct {
 	// readable name for the step
 	Name string `json:"name,omitempty"`
 	// expression to declare under which conditions the step should be run; defaults to \"passed\", except artifacts where it defaults to \"always\"
-	Condition string `json:"condition,omitempty"`
+	Condition string        `json:"condition,omitempty"`
+	Pure      *BoxedBoolean `json:"pure,omitempty"`
 	// should the step be paused initially
 	Paused bool `json:"paused,omitempty"`
 	// is the step expected to fail

pkg/api/v1/testkube/model_test_workflow_step_execute_test_ref.go

@@ -10,16 +10,16 @@
 package testkube
 
 type TestWorkflowStepExecuteTestRef struct {
-	Count    *BoxedString `json:"count,omitempty"`
-	MaxCount *BoxedString `json:"maxCount,omitempty"`
-	// matrix of parameters to spawn instances
-	Matrix map[string]interface{} `json:"matrix,omitempty"`
-	// parameters that should be distributed across sharded instances
-	Shards map[string]interface{} `json:"shards,omitempty"`
 	// test name to schedule
 	Name string `json:"name,omitempty"`
 	// test execution description to display
 	Description      string                                       `json:"description,omitempty"`
+	Count            *BoxedString                                 `json:"count,omitempty"`
+	MaxCount         *BoxedString                                 `json:"maxCount,omitempty"`
 	ExecutionRequest *TestWorkflowStepExecuteTestExecutionRequest `json:"executionRequest,omitempty"`
 	Tarball          map[string]TestWorkflowTarballRequest        `json:"tarball,omitempty"`
+	// matrix of parameters to spawn instances
+	Matrix map[string]interface{} `json:"matrix,omitempty"`
+	// parameters that should be distributed across sharded instances
+	Shards map[string]interface{} `json:"shards,omitempty"`
 }

pkg/api/v1/testkube/model_test_workflow_step_execute_test_workflow_ref.go

@@ -10,12 +10,6 @@
 package testkube
 
 type TestWorkflowStepExecuteTestWorkflowRef struct {
-	Count    *BoxedString `json:"count,omitempty"`
-	MaxCount *BoxedString `json:"maxCount,omitempty"`
-	// matrix of parameters to spawn instances
-	Matrix map[string]interface{} `json:"matrix,omitempty"`
-	// parameters that should be distributed across sharded instances
-	Shards map[string]interface{} `json:"shards,omitempty"`
 	// TestWorkflow name to include
 	Name string `json:"name,omitempty"`
 	// TestWorkflow execution description to display
@@ -24,4 +18,10 @@ type TestWorkflowStepExecuteTestWorkflowRef struct {
 	ExecutionName string                                `json:"executionName,omitempty"`
 	Tarball       map[string]TestWorkflowTarballRequest `json:"tarball,omitempty"`
 	Config        map[string]string                     `json:"config,omitempty"`
+	Count         *BoxedString                          `json:"count,omitempty"`
+	MaxCount      *BoxedString                          `json:"maxCount,omitempty"`
+	// matrix of parameters to spawn instances
+	Matrix map[string]interface{} `json:"matrix,omitempty"`
+	// parameters that should be distributed across sharded instances
+	Shards map[string]interface{} `json:"shards,omitempty"`
 }

pkg/api/v1/testkube/model_test_workflow_step_parallel.go

@@ -28,12 +28,23 @@ type TestWorkflowStepParallel struct {
 	// delay before the step
 	Delay string `json:"delay,omitempty"`
 	// script to run in a default shell for the container
-	Shell     string                                 `json:"shell,omitempty"`
-	Run       *TestWorkflowStepRun                   `json:"run,omitempty"`
-	Execute   *TestWorkflowStepExecute               `json:"execute,omitempty"`
-	Artifacts *TestWorkflowStepArtifacts             `json:"artifacts,omitempty"`
+	Shell     string                     `json:"shell,omitempty"`
+	Run       *TestWorkflowStepRun       `json:"run,omitempty"`
+	Execute   *TestWorkflowStepExecute   `json:"execute,omitempty"`
+	Artifacts *TestWorkflowStepArtifacts `json:"artifacts,omitempty"`
+	// how many resources could be scheduled in parallel
+	Parallelism int32 `json:"parallelism,omitempty"`
+	// worker description to display
+	Description string       `json:"description,omitempty"`
+	Logs        *BoxedString `json:"logs,omitempty"`
+	// list of files to send to parallel steps
+	Transfer []TestWorkflowStepParallelTransfer `json:"transfer,omitempty"`
+	// list of files to fetch from parallel steps
+	Fetch     []TestWorkflowStepParallelFetch        `json:"fetch,omitempty"`
+	Template  *TestWorkflowTemplateRef               `json:"template,omitempty"`
 	Use       []TestWorkflowTemplateRef              `json:"use,omitempty"`
 	Config    map[string]TestWorkflowParameterSchema `json:"config,omitempty"`
+	System    *TestWorkflowSystem                    `json:"system,omitempty"`
 	Content   *TestWorkflowContent                   `json:"content,omitempty"`
 	Services  map[string]TestWorkflowServiceSpec     `json:"services,omitempty"`
 	Container *TestWorkflowContainerConfig           `json:"container,omitempty"`
@@ -43,14 +54,4 @@ type TestWorkflowStepParallel struct {
 	Steps     []TestWorkflowStep                     `json:"steps,omitempty"`
 	After     []TestWorkflowStep                     `json:"after,omitempty"`
 	Events    []TestWorkflowEvent                    `json:"events,omitempty"`
-	// how many resources could be scheduled in parallel
-	Parallelism int32 `json:"parallelism,omitempty"`
-	// worker description to display
-	Description string       `json:"description,omitempty"`
-	Logs        *BoxedString `json:"logs,omitempty"`
-	// list of files to send to parallel steps
-	Transfer []TestWorkflowStepParallelTransfer `json:"transfer,omitempty"`
-	// list of files to fetch from parallel steps
-	Fetch    []TestWorkflowStepParallelFetch `json:"fetch,omitempty"`
-	Template *TestWorkflowTemplateRef        `json:"template,omitempty"`
 }

pkg/api/v1/testkube/model_test_workflow_step_status_extended.go

@@ -0,0 +1,5 @@
+package testkube
+
+func (s TestWorkflowStepStatus) Finished() bool {
+	return s != "" && s != QUEUED_TestWorkflowStepStatus && s != PAUSED_TestWorkflowStepStatus && s != RUNNING_TestWorkflowStepStatus
+}

pkg/api/v1/testkube/model_test_workflow_system.go

@@ -0,0 +1,15 @@
+/*
+ * Testkube API
+ *
+ * Testkube provides a Kubernetes-native framework for test definition, execution and results
+ *
+ * API version: 1.0.0
+ * Contact: [email protected]
+ * Generated by: Swagger Codegen (https://github.com/swagger-api/swagger-codegen.git)
+ */
+package testkube
+
+type TestWorkflowSystem struct {
+	PureByDefault      *BoxedBoolean `json:"pureByDefault,omitempty"`
+	IsolatedContainers *BoxedBoolean `json:"isolatedContainers,omitempty"`
+}

pkg/api/v1/testkube/model_test_workflow_template_spec.go

@@ -11,6 +11,7 @@ package testkube
 
 type TestWorkflowTemplateSpec struct {
 	Config    map[string]TestWorkflowParameterSchema        `json:"config,omitempty"`
+	System    *TestWorkflowSystem                           `json:"system,omitempty"`
 	Content   *TestWorkflowContent                          `json:"content,omitempty"`
 	Services  map[string]TestWorkflowIndependentServiceSpec `json:"services,omitempty"`
 	Container *TestWorkflowContainerConfig                  `json:"container,omitempty"`