prefect_aws.workers.ecs_worker

Prefect worker for executing flow runs as ECS tasks.

Get started by creating a work pool:

$ prefect work-pool create --type ecs my-ecs-pool

Then, you can start a worker for the pool:

$ prefect worker start --pool my-ecs-pool

It's common to deploy the worker as an ECS task as well. However, you can run the worker locally to get started.

The worker may work without any additional configuration, but it is dependent on your specific AWS setup and we'd recommend opening the work pool editor in the UI to see the available options.

By default, the worker will register a task definition for each flow run and run a task in your default ECS cluster using AWS Fargate. Fargate requires tasks to configure subnets, which we will infer from your default VPC. If you do not have a default VPC, you must provide a VPC ID or manually setup the network configuration for your tasks.

Note, the worker caches task definitions for each deployment to avoid excessive registration. The worker will check that the cached task definition is compatible with your configuration before using it.

The launch type option can be used to run your tasks in different modes. For example, FARGATE_SPOT can be used to use spot instances for your Fargate tasks or EC2 can be used to run your tasks on a cluster backed by EC2 instances.

Generally, it is very useful to enable CloudWatch logging for your ECS tasks; this can help you debug task failures. To enable CloudWatch logging, you must provide an execution role ARN with permissions to create and write to log streams. See the configure_cloudwatch_logs field documentation for details.

The worker can be configured to use an existing task definition by setting the task definition arn variable or by providing a "taskDefinition" in the task run request. When a task definition is provided, the worker will never create a new task definition which may result in variables that are templated into the task definition payload being ignored.

Functions

parse_identifier ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L233" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

parse_identifier(identifier: str) -> ECSIdentifier

Splits identifier into its cluster and task components, e.g. input "cluster_name::task_arn" outputs ("cluster_name", "task_arn").

mask_sensitive_env_values ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L242" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

mask_sensitive_env_values(task_run_request: dict, values: List[str], keep_length = 3, replace_with = '***')

mask_api_key ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L261" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

mask_api_key(task_run_request)

Classes

ECSIdentifier ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L154" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

The identifier for a running ECS task.

CapacityProvider ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L269" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

The capacity provider strategy to use when running the task.

ECSJobConfiguration ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L279" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Job configuration for an ECS worker.

Methods:

at_least_one_container_is_essential ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L411" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

at_least_one_container_is_essential(self) -> Self

Ensures that at least one container will be marked as essential in the task definition.

cloudwatch_logs_options_requires_configure_cloudwatch_logs ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L450" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

cloudwatch_logs_options_requires_configure_cloudwatch_logs(self) -> Self

Enforces that an execution role arn is provided (or could be provided by a runtime task definition) when configuring logging.

configure_cloudwatch_logs_requires_execution_role_arn ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L427" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

configure_cloudwatch_logs_requires_execution_role_arn(self) -> Self

Enforces that an execution role arn is provided (or could be provided by a runtime task definition) when configuring logging.

container_name_default_from_task_definition ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L395" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

container_name_default_from_task_definition(self) -> Self

Infers the container name from the task definition if not provided.

json_template ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L332" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

json_template(cls) -> dict[str, Any]

Returns a dict with job configuration as keys and the corresponding templates as values

Defaults to using the job configuration parameter name as the template variable name.

e.g.

{
    key1: '{{ key1 }}',     # default variable template
    key2: '{{ template2 }}', # `template2` specifically provide as template
}

network_configuration_requires_vpc_id ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L465" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

network_configuration_requires_vpc_id(self) -> Self

Enforces a vpc_id is provided when custom network configuration mode is enabled for network settings.

prepare_for_flow_run ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L358" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

prepare_for_flow_run(self, flow_run: 'FlowRun', deployment: 'DeploymentResponse | None' = None, flow: 'APIFlow | None' = None, work_pool: 'WorkPool | None' = None, worker_name: str | None = None, worker_id: 'UUID | None' = None) -> None

task_run_request_requires_arn_if_no_task_definition_given ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L379" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

task_run_request_requires_arn_if_no_task_definition_given(self) -> Self

If no task definition is provided, a task definition ARN must be present on the task run request.

ECSVariables ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L477" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Variables for templating an ECS job.

ECSWorkerResult ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L683" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

The result of an ECS job.

ECSWorker ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L689" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

A Prefect worker to run flow runs as ECS tasks.

Methods:

kill_infrastructure ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L1683" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

kill_infrastructure(self, infrastructure_pid: str, configuration: ECSJobConfiguration, grace_seconds: int = 30) -> None

Stop an ECS task.

Args:

• infrastructure_pid: The infrastructure identifier in format "cluster::task_arn".
• configuration: The job configuration used to connect to AWS.
• grace_seconds: Not used for ECS (ECS handles graceful shutdown internally).

Raises:

• InfrastructureNotFound: If the task doesn't exist.

run ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L727" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

run(self, flow_run: 'FlowRun', configuration: ECSJobConfiguration, task_status: Optional[anyio.abc.TaskStatus] = None) -> ECSWorkerResult

Runs a given flow run on the current worker.