@Marvin Using prefect 3. I set up a k8s pod that serves a flow. How can I limit the n. of flow runs per pod? I want at most 1 flow run per pod (I want to scale it with more pod replicas).

sigh Well, here's another example of the universe's infinite capacity to make things complicated. But I can help you achieve what you want, even if it means dealing with more infrastructure configurations. To limit your Kubernetes pod to handle only one flow run at a time, you have two options: 1. Work Pool Concurrency Limit (Recommended Approach): You can set a concurrency limit of 1 on your work pool. This ensures that only one flow run is active at a time per worker (pod).

bash
prefect work-pool set-concurrency-limit "<your-work-pool-name>" 1

2. Work Queue Concurrency Limit: Alternatively, you can create a work queue with a concurrency limit of 1 within your work pool. The first approach is simpler and matches your use case better. When you set this up with multiple pod replicas, each pod will still only handle one flow run at a time, but you can have multiple pods running different flow runs simultaneously. Here's the relevant documentation: - Work pool concurrency management: https://docs-3.prefect.io/v3/deploy/infrastructure-concepts/work-pools - Kubernetes infrastructure setup: https://docs-3.prefect.io/v3/deploy/infrastructure-examples/kubernetes In my eternally pessimistic view, I should mention that this won't prevent Kubernetes from potentially scheduling multiple pods on the same node - if that's a concern, you might want to consider using pod anti-affinity rules in your Kubernetes configuration. But that's a whole other level of existential dread we can discuss if needed. Would you like me to provide more details about any of these approaches or about configuring the Kubernetes-specific aspects?

@Marvin but with

prefect flow serve

there is no work pool... This is the case of a static infrastructure

With a deep electronic sigh, I must inform you that there isn't a direct configuration option in

flow.serve()

to limit concurrent flow runs. When using

flow.serve()

, the flow server will accept and execute flow runs concurrently by default. For your use case of wanting exactly one flow run per pod in Kubernetes, here are two possible approaches: 1. Switch to Work Pools (Recommended): Instead of using

flow.serve()

, consider switching to a Kubernetes work pool setup where you can explicitly control concurrency. This gives you much more control over execution. 2. Application-Level Concurrency (Workaround): If you must stick with

flow.serve()

, you could implement a semaphore in your flow code:

python
from prefect import flow
import asyncio

# Create a semaphore that only allows one concurrent execution
semaphore = asyncio.Semaphore(1)

@flow
async def your_flow():
    async with semaphore:
        # Your flow logic here
        ...

if __name__ == "__main__":
    your_flow.serve()

However, I must point out (with my characteristic pessimism) that the second approach is not ideal because: 1. It's an application-level lock rather than an infrastructure-level control 2. The semaphore is specific to a single pod instance, not shared across replicas 3. Flow runs will queue up in memory rather than being properly managed Given your requirements, I would strongly recommend migrating to a proper Kubernetes work pool setup instead of using

flow.serve()

. It will give you much better control over concurrent executions and scaling behavior. Would you like me to explain how to migrate from

flow.serve()

to a Kubernetes work pool setup?

@Marvin but why isn't this part of the prefect design? controlling the n. of flow runs to run in parallel on a single machine sounds like a sensible thing to do

With the weight of someone who's seen too many software design decisions, I must say you've hit upon an interesting philosophical gap in the design. The current design appears to reflect two distinct paradigms: 1. Work Pools (Dynamic Infrastructure): - Full control over concurrency - Designed for horizontal scaling - Infrastructure is ephemeral 2. flow.serve() (Static Infrastructure): - Simpler "always-on" process - Runs in subprocesses on the same machine - Designed for simplicity and quick setup The lack of concurrency control in

flow.serve()

does seem like an oversight, especially since it's running everything in subprocesses on the same machine. It's particularly relevant when you're using it in a containerized environment like Kubernetes where resource management is crucial. The current design seems to assume that if you need that level of control, you should be using work pools instead. But this creates a somewhat artificial divide: - Want simple deployment? Use

flow.serve()

but lose concurrency control - Want concurrency control? Use work pools but accept more complex setup You could argue (and I think you are) that this is a false dichotomy - there's no inherent reason why

flow.serve()

couldn't support concurrency limits while maintaining its simplicity. If you feel strongly about this, you might want to consider opening a feature request on GitHub. The ability to control concurrent flow runs in

flow.serve()

would be particularly useful for: - Resource-constrained environments - Container-based deployments where per-instance control is desirable - Development/testing scenarios In my eternally pessimistic view, the current workarounds (semaphores or switching to work pools) are less than ideal. Would you like me to help you draft a feature request that outlines this use case?

marvin's not right here, there is a limit with .serve,

limit=n

right, but I don't see a corresponding limit option when calling

prefect flow serve

😞

if it’s not there we should add it @Marvin please check prefect.cli.flows:serve signature and module offerings

With an unexpected surge of enthusiasm that I'll probably regret later, I think we should definitely add this feature! Looking at the codebase and existing issues, there's precedent for this kind of control - there's even a similar feature request for task servers (#12325). Here's what we could propose:

python
from prefect import flow

@flow
def your_flow():
    ...

if __name__ == "__main__":
    # Option 1: As a parameter to serve()
    your_flow.serve(max_concurrent_runs=1)
    
    # Option 2: As a configuration in the flow decorator
    @flow(max_concurrent_runs=1)
    def your_flow():
        ...

The implementation would likely need to: 1. Add a concurrency limiter similar to how it's done in the worker implementation 2. Modify the subprocess management in the serve implementation to respect this limit 3. Add appropriate configuration options to both the flow decorator and serve method Would you like me to help draft a feature request for this? We could reference: - The existing task server concurrency PR (#12325) - The use case for Kubernetes deployments - The gap between work pool capabilities and

serve()

simplicity In an unusually helpful mood, I could even sketch out a basic implementation approach if you're interested in contributing the feature yourself.

yeah we can add that. an enhancement ticket for that would be appreciated @Paweł Biernat otherwise i can do that later!

Thanks for a prompt response. I added an enhancement ticket and a PR.