Merge branch 'vsukhin/feature/run-workflow-selector' into sandbox/twrc

api/v1/testkube.yaml

@@ -9042,7 +9042,7 @@ components:
           config:
             $ref: "#/components/schemas/TestWorkflowConfigValue"
           selector:
-            $ref: "https://raw.githubusercontent.com/garethr/kubernetes-json-schema/master/v1.7.8/_definitions.json#/definitions/io.k8s.apimachinery.pkg.apis.meta.v1.LabelSelector"
+            $ref: "#/components/schemas/LabelSelector"
             description: label selector for test workflow
 
     TestWorkflowStepExecuteTestRef:

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

@@ -47,7 +47,6 @@ func NewRunTestWorkflowCmd() *cobra.Command {
 		masks                    []string
 		tags                     map[string]string
 		selectors                []string
-		concurrencyLevel         int
 	)
 
 	cmd := &cobra.Command{
@@ -102,7 +101,7 @@ func NewRunTestWorkflowCmd() *cobra.Command {
 				executions = append(executions, execution)
 			case len(selectors) != 0:
 				selector := strings.Join(selectors, ",")
-				executions, err = client.ExecuteTestWorkflows(selector, concurrencyLevel, request)
+				executions, err = client.ExecuteTestWorkflows(selector, request)
 			default:
 				ui.Failf("Pass Test workflow name or labels to run by labels ")
 			}
@@ -182,7 +181,6 @@ func NewRunTestWorkflowCmd() *cobra.Command {
 	cmd.Flags().StringArrayVarP(&masks, "mask", "", []string{}, "regexp to filter downloaded files, single or comma separated, like report/.* or .*\\.json,.*\\.js$")
 	cmd.Flags().StringToStringVarP(&tags, "tag", "", map[string]string{}, "execution tags in a form of name1=val1 passed to executor")
 	cmd.Flags().StringSliceVarP(&selectors, "label", "l", nil, "label key value pair: --label key1=value1 or label expression")
-	cmd.Flags().IntVar(&concurrencyLevel, "concurrency", 10, "concurrency level for multiple test workflow executions")
 
 	return cmd
 }

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

@@ -378,6 +378,10 @@ func NewExecuteCmd() *cobra.Command {
 					ui.Fail(errors.Wrap(err, "unmarshal workflow definition"))
 				}
 
+				if w.Name == "" && w.Selector == nil {
+					ui.Fail(errors.New("either workflow name or selector should be specified"))
+				}
+
 				var testWorkflowNames []string
 				if w.Name != "" {
 					testWorkflowNames = []string{w.Name}

internal/app/api/v1/testworkflows.go

@@ -390,12 +390,7 @@ func (s *TestkubeAPI) ExecuteTestWorkflowHandler() fiber.Handler {
 
 		if len(testWorkflows) != 0 {
 			request.TestWorkflowExecutionName = strings.Clone(c.Query("testWorkflowExecutionName"))
-			concurrencyLevel, err := strconv.Atoi(c.Query("concurrency", strconv.Itoa(scheduler.DefaultConcurrencyLevel)))
-			if err != nil {
-				return s.BadRequest(c, errPrefix, "can't detect concurrency level", err)
-			}
-
-			workerpoolService := workerpool.New[testworkflowsv1.TestWorkflow, testkube.TestWorkflowExecutionRequest, testkube.TestWorkflowExecution](concurrencyLevel)
+			workerpoolService := workerpool.New[testworkflowsv1.TestWorkflow, testkube.TestWorkflowExecutionRequest, testkube.TestWorkflowExecution](scheduler.DefaultConcurrencyLevel)
 
 			go workerpoolService.SendRequests(s.prepareTestWorkflowRequests(testWorkflows, request))
 			go workerpoolService.Run(ctx)

pkg/api/v1/client/interface.go

@@ -151,7 +151,7 @@ type TestWorkflowAPI interface {
 	UpdateTestWorkflow(workflow testkube.TestWorkflow) (testkube.TestWorkflow, error)
 	DeleteTestWorkflow(name string) error
 	ExecuteTestWorkflow(name string, request testkube.TestWorkflowExecutionRequest) (testkube.TestWorkflowExecution, error)
-	ExecuteTestWorkflows(sselector string, concurrencyLevel int, request testkube.TestWorkflowExecutionRequest) ([]testkube.TestWorkflowExecution, error)
+	ExecuteTestWorkflows(selector string, request testkube.TestWorkflowExecutionRequest) ([]testkube.TestWorkflowExecution, error)
 	GetTestWorkflowExecutionNotifications(id string) (chan testkube.TestWorkflowExecutionNotification, error)
 	GetTestWorkflowExecutionLogs(id string) ([]byte, error)
 }

pkg/api/v1/client/testworkflow.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
-	"strconv"
 
 	"github.com/kubeshop/testkube/pkg/api/v1/testkube"
 )
@@ -124,7 +123,7 @@ func (c TestWorkflowClient) ExecuteTestWorkflow(name string, request testkube.Te
 
 // ExecuteTestWorkflows starts new external test workflow executions, reads data and returns IDs
 // Executions are started asynchronously client can check later for results
-func (c TestWorkflowClient) ExecuteTestWorkflows(selector string, concurrencyLevel int, request testkube.TestWorkflowExecutionRequest) (executions []testkube.TestWorkflowExecution, err error) {
+func (c TestWorkflowClient) ExecuteTestWorkflows(selector string, request testkube.TestWorkflowExecutionRequest) (executions []testkube.TestWorkflowExecution, err error) {
 	uri := c.testWorkflowExecutionTransport.GetURI("/test-workflow-executions")
 
 	body, err := json.Marshal(request)
@@ -133,8 +132,7 @@ func (c TestWorkflowClient) ExecuteTestWorkflows(selector string, concurrencyLev
 	}
 
 	params := map[string]string{
-		"selector":    selector,
-		"concurrency": strconv.Itoa(concurrencyLevel),
+		"selector": selector,
 	}
 
 	return c.testWorkflowExecutionTransport.ExecuteMultiple(http.MethodPost, uri, body, params)

pkg/api/v1/testkube/model_test_workflow_step_execute_test_workflow_ref.go

@@ -23,6 +23,6 @@ type TestWorkflowStepExecuteTestWorkflowRef struct {
 	// 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"`
-	Selector *IoK8sApimachineryPkgApisMetaV1LabelSelector `json:"selector,omitempty"`
+	Shards   map[string]interface{} `json:"shards,omitempty"`
+	Selector *LabelSelector         `json:"selector,omitempty"`
 }

pkg/mapper/testworkflows/kube_openapi.go

@@ -806,16 +806,16 @@ func MapStepExecuteTestKubeToAPI(v testworkflowsv1.StepExecuteTest) testkube.Tes
 	}
 }
 
-func MapLabelSelectorRequirementToAPI(v metav1.LabelSelectorRequirement) testkube.IoK8sApimachineryPkgApisMetaV1LabelSelectorRequirement {
-	return testkube.IoK8sApimachineryPkgApisMetaV1LabelSelectorRequirement{
+func MapLabelSelectorRequirementToAPI(v metav1.LabelSelectorRequirement) testkube.LabelSelectorRequirement {
+	return testkube.LabelSelectorRequirement{
 		Key:      v.Key,
 		Operator: string(v.Operator),
 		Values:   v.Values,
 	}
 }
 
-func MapSelectorToAPI(v metav1.LabelSelector) testkube.IoK8sApimachineryPkgApisMetaV1LabelSelector {
-	return testkube.IoK8sApimachineryPkgApisMetaV1LabelSelector{
+func MapSelectorToAPI(v metav1.LabelSelector) testkube.LabelSelector {
+	return testkube.LabelSelector{
 		MatchLabels:      v.MatchLabels,
 		MatchExpressions: common.MapSlice(v.MatchExpressions, MapLabelSelectorRequirementToAPI),
 	}

pkg/mapper/testworkflows/openapi_kube.go

@@ -846,15 +846,15 @@ func MapStepExecuteTestAPIToKube(v testkube.TestWorkflowStepExecuteTestRef) test
 	}
 }
 
-func MapLabelSelectorRequirementToCRD(v testkube.IoK8sApimachineryPkgApisMetaV1LabelSelectorRequirement) metav1.LabelSelectorRequirement {
+func MapLabelSelectorRequirementToCRD(v testkube.LabelSelectorRequirement) metav1.LabelSelectorRequirement {
 	return metav1.LabelSelectorRequirement{
 		Key:      v.Key,
 		Operator: metav1.LabelSelectorOperator(v.Operator),
 		Values:   v.Values,
 	}
 }
 
-func MapSelectorToCRD(v testkube.IoK8sApimachineryPkgApisMetaV1LabelSelector) metav1.LabelSelector {
+func MapSelectorToCRD(v testkube.LabelSelector) metav1.LabelSelector {
 	return metav1.LabelSelector{
 		MatchLabels:      v.MatchLabels,
 		MatchExpressions: common.MapSlice(v.MatchExpressions, MapLabelSelectorRequirementToCRD),