#5125 [Core feature] Workflow level throttling Issue created by a4501150 ### Motivation: Why do you think this is important? Currently flyte does not support workflow level throttling that limits the

same

concurrent workflows that can be running under a flyte project / domain. (same workflow means the same binary identified by workflow name). In our use case, we use flyte both for workflow scheduling and orchestration. One of our workflow will hit our backend API heavily and we want to limit the resource pressure on our backend API. Although we can implement throttling on backend API but backend API will also use in synchronized manner and we don't want to affect the current synchronized API performance and behavior, thus implement throttling on workflow side is the best choice. Compare to other workflow engines, this is the major feature that lacks in flyte. ### Goal: What should the final outcome look like, ideally? For instance, when we trigger a workflow by creating a

CreateExecution

with a launch plan (preferred) OR when we register a workflow, we should be able to configure an extra parameter, eg.

max_concurrent_execution

as type long which can limit the concurrent workflows that running at the same time. For example, let's say we have a

workflow A

, with a

Task A

. We created a launch plan for this workflow with

max_concurrent_execution

as 5. • We trigger the concurrent workflows by creating 100

CreateExecution

requests with same launch plan to flyte grpc API • The first 5 workflow executions should be running • The rest of 95 executions should be put in a queue at flyte backend, in status of

QUEUED

• The ultimate goal is there should be only

max_concurrent_execution

workflows in status of

RUNNING

at the same time • The execution id should be always returned ### Describe alternatives you've considered We are now using a hacky way by creating a

resource lock

task to achieve the functionalities: • A new task with

for loop

and

Thread.sleep()

inside the for loop along with querying the flyte endpoint to get the top

max_concurrent_execution

running executions sorted by execution created time. • If the returned top X running executions count >=

max_concurrent_execution

, and the execution id obtained by the the task via ENV

FLYTE_INTERNAL_EXECUTION_ID

is not one of the returned top X executions, then the task will just wait there in the for loop with

thread.sleep()

This effectively limit the concurrent executions of a given launchPlan (workflow) by have the throttling task as the first task in the given workflow DAG. However, this will create many semi-dangling

resource lock

tasks if the amount of concurrent execution requests is large and

max_concurrent_execution

is relatively small. ### Another alternatives solution Task level throttling can further enhance the fine control level, however this maybe harder to implement than workflow level throttling. ### Propose: Link/Inline OR Additional context No response ### Are you sure this issue hasn't been raised already? • Yes ### Have you read the Code of Conduct? • Yes flyteorg/flyte