Specifying Kubernetes resources when running a Deployment

[ohadch]

Dear Prefect Team,

First, I would like express our satisfaction with Prefect, as our data infrastructure relies heavily on its capabilities. We have successfully implemented a system with multiple work pools, each executed by a Kubernetes worker residing on different Kubernetes clusters.

However, we have encountered a challenge related to resource control. Currently, we have exposed the Job's resources, including both requests and limits, in the work pool's base job parameters. While this setup works adequately, we are eager to enhance our resource management capabilities.

Specifically, we are looking to introduce dynamic control over Kubernetes resource requests per flow run. This would allow us to adapt to various scenarios requiring vertical scaling of pods. Our goal is to pass the requests and limits, which we added to the work pool's base job templates, as parameters to flow runs triggered by deployments, both through the UI and Python API.

Upon investigating the source code, we identified that the run_deployment() function does not currently accept parameters to override the base job template. Unfortunately, this limitation seems to be a roadblock for our desired functionality:

@sync_compatible
@inject_client
async def run_deployment(
    name: Union[str, UUID],
    client: Optional[PrefectClient] = None,
    parameters: Optional[dict] = None,
    scheduled_time: Optional[datetime] = None,
    flow_run_name: Optional[str] = None,
    timeout: Optional[float] = None,
    poll_interval: Optional[float] = 5,
    tags: Optional[Iterable[str]] = None,
    idempotency_key: Optional[str] = None,
    work_queue_name: Optional[str] = None,
    as_subflow: Optional[bool] = True,
) -> FlowRun:
    ...

We thought about adding something like base_job_template_parameters_overrides parameters to the run_deployment() function which would let us to pass our required resources for the flow run:

@sync_compatible
@inject_client
async def run_deployment(
    name: Union[str, UUID],
    client: Optional[PrefectClient] = None,
    parameters: Optional[dict] = None,
    scheduled_time: Optional[datetime] = None,
    flow_run_name: Optional[str] = None,
    timeout: Optional[float] = None,
    poll_interval: Optional[float] = 5,
    tags: Optional[Iterable[str]] = None,
    idempotency_key: Optional[str] = None,
    work_queue_name: Optional[str] = None,
    as_subflow: Optional[bool] = True,
    base_job_template_parameters_overrides: Optional[dict] = None,
) -> FlowRun:
    ...

Then, when the worker dispatches the flow run, it would use the custom resources that we requested, in case provided.

This feature would be a significant enabler for us, and we are eager to contribute to its development if necessary. We appreciate your time and consideration in addressing this matter. Your guidance or insights into potential solutions would be immensely valuable to our team.

Looking forward to your response.

Best Regards,

Ohad Chaet