fix(api): template bulk runner count for parallel apply (#6510)

Signed-off-by: richardlt <richard.le.terrier@gmail.com>
ovh · Mar 20, 2023 · c6e3eae · c6e3eae
1 parent c6aca14
commit c6e3eae
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 9 deletions.
diff --git a/engine/api/api.go b/engine/api/api.go
@@ -842,12 +842,15 @@ func (a *API) Serve(ctx context.Context) error {
 			a.cleanWorkflowRunSecrets(ctx)
 		})
 	}
-	chanWorkflowTemplateBulkOperation := make(chan WorkflowTemplateBulkOperation)
+	if a.Config.Workflow.TemplateBulkRunnerCount == 0 {
+		a.Config.Workflow.TemplateBulkRunnerCount = 10
+	}
+	chanWorkflowTemplateBulkOperation := make(chan WorkflowTemplateBulkOperation, a.Config.Workflow.TemplateBulkRunnerCount*10)
 	defer close(chanWorkflowTemplateBulkOperation)
 	a.GoRoutines.RunWithRestart(ctx, "api.WorkflowTemplateBulk", func(ctx context.Context) {
 		a.WorkflowTemplateBulk(ctx, 100*time.Millisecond, chanWorkflowTemplateBulkOperation)
 	})
-	a.WorkflowTemplateBulkOperation(ctx, a.Config.Workflow.MaxRuns, chanWorkflowTemplateBulkOperation)
+	a.WorkflowTemplateBulkOperation(ctx, chanWorkflowTemplateBulkOperation)
 
 	log.Info(ctx, "Bootstrapping database...")
 	defaultValues := sdk.DefaultValues{

diff --git a/engine/api/workflow_template_bulk.go b/engine/api/workflow_template_bulk.go
@@ -116,7 +116,7 @@ func (api *API) workflowTemplateBulk(ctx context.Context, bulkID int64, chanOper
 	}
 
 	if b.Parallel {
-		for len(operationTodo) > 0 {
+		for i := 0; i < int(api.Config.Workflow.TemplateBulkRunnerCount) && len(operationTodo) > 0; i++ {
 			chanOperation <- operationTodo.Pop()
 		}
 	} else {
@@ -137,7 +137,13 @@ func (api *API) workflowTemplateBulk(ctx context.Context, bulkID int64, chanOper
 				return nil
 			}
 			if (res.Status == sdk.OperationStatusDone || res.Status == sdk.OperationStatusError) && len(operationTodo) > 0 {
-				chanOperation <- operationTodo.Pop()
+				if b.Parallel {
+					for i := 0; i < int(api.Config.Workflow.TemplateBulkRunnerCount) && len(operationTodo) > 0; i++ {
+						chanOperation <- operationTodo.Pop()
+					}
+				} else {
+					chanOperation <- operationTodo.Pop()
+				}
 			}
 		case <-ctx.Done():
 			b.Status = sdk.OperationStatusPending
@@ -146,11 +152,8 @@ func (api *API) workflowTemplateBulk(ctx context.Context, bulkID int64, chanOper
 	}
 }
 
-func (api *API) WorkflowTemplateBulkOperation(ctx context.Context, routineCount int64, chanOperation chan WorkflowTemplateBulkOperation) {
-	if routineCount == 0 {
-		routineCount = 10
-	}
-	for i := int64(0); i < routineCount; i++ {
+func (api *API) WorkflowTemplateBulkOperation(ctx context.Context, chanOperation chan WorkflowTemplateBulkOperation) {
+	for i := int64(0); i < api.Config.Workflow.TemplateBulkRunnerCount; i++ {
 		api.GoRoutines.RunWithRestart(ctx, fmt.Sprintf("api.WorkflowTemplateBulkOperation-%d", i), func(ctx context.Context) {
 			for {
 				select {