How to run deployments with ECSTask infrastructure block at scale (running hundreds of serverless ECS-containerized flow runs concurrently)?

anna_geller · November 18, 2022, 2:00am

We often hear a use case where users would like to loop over some input and generate tens if not hundreds of flow runs running on AWS ECS Fargate.

Problem

You can’t register an infinite number of ECS task revisions for the same ECS task family. You’ll get an error such as:

02:31:48.532 | ERROR   | prefect.agent - Failed to submit flow run 'fcb99032-a6da-4f26-81fb-16a653fd0753' to infrastructure.
Traceback (most recent call last):
  File "/opt/miniconda3/envs/dataplatform/lib/python3.10/site-packages/prefect/agent.py", line 265, in _submit_run_and_capture_errors
    result = await infrastructure.run(task_status=task_status)
  File "/opt/miniconda3/envs/dataplatform/lib/python3.10/site-packages/prefect_aws/ecs.py", line 507, in run
    ) = await run_sync_in_worker_thread(
  File "/opt/miniconda3/envs/dataplatform/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 68, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(call, cancellable=True)
  File "/opt/miniconda3/envs/dataplatform/lib/python3.10/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/opt/miniconda3/envs/dataplatform/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/opt/miniconda3/envs/dataplatform/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/opt/miniconda3/envs/dataplatform/lib/python3.10/site-packages/prefect_aws/ecs.py", line 586, in _create_task_and_wait_for_start
    task_definition_arn = self._register_task_definition(
  File "/opt/miniconda3/envs/dataplatform/lib/python3.10/site-packages/prefect_aws/ecs.py", line 980, in _register_task_definition
    response = ecs_client.register_task_definition(**task_definition_request)
  File "/opt/miniconda3/envs/dataplatform/lib/python3.10/site-packages/botocore/client.py", line 507, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/opt/miniconda3/envs/dataplatform/lib/python3.10/site-packages/botocore/client.py", line 943, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.errorfactory.ClientException: An error occurred (ClientException) when calling the RegisterTaskDefinition operation: Too many concurrent attempts to create a new revision of the specified family.

Generally, registering new ECS task definitions is a subject of throttling from the AWS ECS API.

Possible solutions

Add latency in between iterations to avoid concurrency throttling from ECS

Add some time in between (either by increasing the scheduling interval or by adding time.sleep(30) in between API calls triggering a flow run from that deployment). This way, you can keep running hundreds of containerized flow runs and keep registering hundreds of task definition revisions without throttling issues that are otherwise applied if you perform such operations concurrently.

Flow:

import platform
import prefect
from prefect import task, flow, get_run_logger
import sys


@task
def log_platform_info():
    logger = get_run_logger()
    logger.info("Host's network name = %s", platform.node())
    logger.info("Python version = %s", platform.python_version())
    logger.info("Platform information (instance type) = %s ", platform.platform())
    logger.info("OS/Arch = %s/%s", sys.platform, platform.machine())
    logger.info("Prefect Version = %s 🚀", prefect.__version__)


@flow
def healthcheck():
    log_platform_info()


if __name__ == "__main__":
    healthcheck()

Create deployment (assuming setup similar to this repo dataflow-ops/healthcheck.py at main · anna-geller/dataflow-ops · GitHub):

prefect deployment build flows/healthcheck.py:healthcheck -n prod -q prod -a -sb github/default -ib ecs-task/default

Start an agent:

prefect agent start -q sss

Trigger 100 of runs:

from prefect.deployments import run_deployment
import time


for _ in range(1, 100):
    run_deployment(name="healthcheck/prod", timeout=0)
    time.sleep(30)

Register task definition once and reuse

Alternatively, you can register a task definition e.g. from your CI/CD environment and provide an explicit task definition ARN to your ECSTask infrastructure block. This way, concurrency is not an issue because no new task definition revisions are registered.

ecs = ECSTask(
    aws_credentials=aws_creds,
    cluster="prefect",
    task_definition_arn="arn:aws:ecs:us-east-1:111111111111111:task-definition/prefect",
)
ecs.save(DEFAULT_BLOCK, overwrite=True)

More:

github.com/PrefectHQ/prefect-aws

Reduce chances of hitting AWS ECS Rate Limits

opened 05:13PM - 02 Nov 22 UTC

closed 04:20PM - 11 Nov 22 UTC

gabcoyne

AWS will rate limit task definition registration for a given task family. There …are two factors that make it easy for a relatively low volume of Prefect flows to hit rate limits: 1. All Prefect Flows are registered under the same task family "prefect" defined [here](https://github.com/PrefectHQ/prefect-aws/blob/0f74e836a80a466e0fdfb2dc61ac070a1d723c5d/prefect_aws/ecs.py#L949). This is relevant because the error raise is `RegisterTaskDefinition operation: Too many concurrent attempts to create a new revision of the specified family.` 2. Task definitions create a new log stream for each task created [here](https://github.com/PrefectHQ/prefect-aws/blob/0f74e836a80a466e0fdfb2dc61ac070a1d723c5d/prefect_aws/ecs.py#L795), which in turn means that each flow run has a new task definition. Solution: I feel like creating task families per deployment will be the easiest way around this, as this reduces the number of registrations for a given family. However this does not fully resolve the issue. It is nice to see a new log stream for every flow run, however the need to register task definitions introduces a chance for rate limits to be hit.

Extra notes

We improved this here but you still need to be careful in running serverless containers at scale - following the above + using prefect-aws >= 0.1.8 should help.

github.com/PrefectHQ/prefect-aws

Generate `ECSTask` family names from deployment and flow metadata

PrefectHQ:main ← PrefectHQ:family-namegen

opened 05:27PM - 09 Nov 22 UTC

madkinsz

+204 -7

Adds a `family` field to the top-level settings, defaults to `prefect` as before but can be overridden. Generates the task definition family as `"prefect__{flow-name}__{deployment-name}"` if the metadata is passed to `prepare_for_flow_run` and it is not set already. See the test cases for exhaustive examples. Requires https://github.com/PrefectHQ/prefect/pull/7479 Closes #148 ### Example  ```python from prefect_aws.ecs import ECSTask from prefect.orion.schemas.core import Deployment, FlowRun, Flow print( ECSTask() .prepare_for_flow_run( FlowRun.construct(), flow=Flow.construct(name="foo"), deployment=Deployment.construct(name="bar"), ) .preview() ) ``` ```yaml --- # Task definition containerDefinitions: - image: prefecthq/prefect:dev-python3.8 name: prefect cpu: '1024' family: prefect__foo__bar memory: '2048' networkMode: awsvpc requiresCompatibilities: - FARGATE ``` ### Checklist - [x] This pull request references any related issue by including "Closes #<ISSUE_NUMBER>" - If no issue exists and your change is not a small fix, please [create an issue](https://github.com/PrefectHQ/prefect-aws/issues/new/choose) first. - [x] This pull request includes tests or only affects documentation. - [x] Summarized PR's changes in [CHANGELOG.md](https://github.com/PrefectHQ/prefect-aws/blob/main/CHANGELOG.md)