Hi! Quick question: is there currently any way to set flow-level concurrency limits? (rather than just task level concurrency with tags)

If you use DaskExecutor, the number of workers becomes the max number of concurrent tasks:

with Flow(
    "flow name",
    environment=prefect.environments.RemoteEnvironment(
        executor="prefect.engine.executors.DaskExecutor",
        executor_kwargs={'n_workers': 4},
    ),
) as flow:

Thanks Luke, we're currently using the plain Kubernetes executor and are looking to limit the number of concurrent flows the Kubernetes agent triggers. Each flow job has pretty significant resource requests, so rather than having a bunch of jobs fighting for resources would be nice to have the Kubernetes agent to hold off on creating new jobs for flows based on a flow level concurrency constraint

Hey @Cab Maddux - we don’t currently have a flow concurrency limit feature but this is definitely in-scope. Would you mind opening an issue for it? We’d need to make some decisions about whether it’s a truly global limit, a limit based on flow labels, or a per-agent limit so any suggestions from your use case would be awesome as well

Thanks @Chris White! For sure, I'll open an issue and reference our case