ECS Worker Guide

Why use ECS for flow run execution?

ECS (Elastic Container Service) tasks are a good option for executing Prefect 2 flow runs for several reasons:

Scalability: ECS scales your infrastructure in response to demand, effectively managing Prefect flow runs. ECS automatically administers container distribution across multiple instances based on demand.
Flexibility: ECS lets you choose between AWS Fargate and Amazon EC2 for container operation. Fargate abstracts the underlying infrastructure, while EC2 has faster job start times and offers additional control over instance management and configuration.
AWS Integration: Easily connect with other AWS services, such as AWS IAM and CloudWatch.
Containerization: ECS supports Docker containers and offers managed execution. Containerization encourages reproducible deployments.

ECS Flow Run Execution

Prefect enables remote flow execution via workers and work pools. To learn more about these concepts please see our deployment tutorial.

For details on how workers and work pools are implemented for ECS, see the diagram below:

Architecture Diagram

graph TB

  subgraph ecs_cluster[ECS cluster]
    subgraph ecs_service[ECS service]
      td_worker[Worker task definition] --> |defines| prefect_worker((Prefect worker))
    end
    prefect_worker -->|kicks off| ecs_task
    fr_task_definition[Flow run task definition]


    subgraph ecs_task["ECS task execution <br>  (Flow run infrastructure)"]
    style ecs_task text-align:center

      flow_run((Flow run))

    end
    fr_task_definition -->|defines| ecs_task
  end

  subgraph prefect_cloud[Prefect Cloud]
    subgraph prefect_workpool[ECS work pool]
      workqueue[Work queue]
    end
  end

  subgraph github["GitHub"]
    flow_code{{"Flow code"}}
  end
  flow_code --> |pulls| ecs_task
  prefect_worker -->|polls| workqueue
  prefect_workpool -->|configures| fr_task_definition

ECS in Prefect Terms

ECS tasks != Prefect tasks

An ECS task is not the same thing as a Prefect task.

ECS tasks are groupings of containers that run within an ECS Cluster. An ECS task's behavior is determined by its task definition.

An ECS task definition is the blueprint for the ECS task. It describes which Docker containers to run and what you want to have happen inside these containers.

ECS tasks are instances of a task definition. A Task Execution launches container(s) as defined in the task definition until they are stopped or exit on their own. This setup is ideal for ephemeral processes such as a Prefect flow run.

The ECS task running the Prefect worker should be an ECS Service, given its long-running nature and need for auto-recovery in case of failure. An ECS service automatically replaces any task that fails, which is ideal for managing a long-running process such as a Prefect worker.

When a Prefect flow is scheduled to run it goes into the work pool specified in the flow's deployment. Work pools are typed according to the infrastructure the flow will run on. Flow runs scheduled in an ecs typed work pool are executed as ECS tasks. Only Prefect ECS workers can poll an ecs typed work pool.

When the ECS worker receives a scheduled flow run from the ECS work pool it is polling, it spins up the specified infrastructure on AWS ECS. The worker knows to build an ECS task definition for each flow run based on the configuration specified in the work pool.

Once the flow run completes, the ECS containers of the cluster are spun down to a single container that continues to run the Prefect worker. This worker continues polling for work from the Prefect work pool.

If you specify a task definition ARN (Amazon Resource Name) in the work pool, the worker will use that ARN when spinning up the ECS Task, rather than creating a task definition from the fields supplied in the work pool configuration.

You can use either EC2 or Fargate as the capacity provider. Fargate simplifies initiation, but lengthens infrastructure setup time for each flow run. Using EC2 for the ECS cluster can reduce setup time. In this example, we will show how to use Fargate.

AWS CLI Guide

Tip

If you prefer infrastructure as code check out this Terraform module to provision an ECS cluster with a worker.

Prerequisites

Before you begin, make sure you have:

An AWS account with permissions to create ECS services and IAM roles.
The AWS CLI installed on your local machine. You can download it from the AWS website.
An ECS Cluster to host both the worker and the flow runs it submits. Follow this guide to create an ECS cluster or simply use the default cluster.
A VPC configured for your ECS tasks. A VPC is a good idea if using EC2 and required if using Fargate.

Step 1: Set Up an ECS work pool

Before setting up the worker, create a simple work pool of type ECS for the worker to pull work from.

Create a work pool from the Prefect UI or CLI:

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

Configure the VPC and ECS cluster for your work pool via the UI: VPC

Configuring custom fields is easiest from the UI.

ECSCluster

Warning

You need to have a VPC specified for your work pool if you are using AWS Fargate.

Launch

Next, set up a Prefect ECS worker that will discover and pull work from this work pool.

Step 2: Start a Prefect worker in your ECS cluster.

To create an IAM role for the ECS task using the AWS CLI, follow these steps:

Create a trust policy

The trust policy will specify that ECS can assume the role.

Save this policy to a file, such as ecs-trust-policy.json:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "ecs-tasks.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

Create the IAM role

Use the aws iam create-role command to create the role:

aws iam create-role \ 
--role-name ecsTaskExecutionRole \
--assume-role-policy-document file://ecs-trust-policy.json

Attach the policy to the role

Amazon has a managed policy named AmazonECSTaskExecutionRolePolicy that grants the permissions necessary for ECS tasks. Attach this policy to your role:
```
aws iam attach-role-policy \
--role-name ecsTaskExecutionRole \ 
--policy-arn arn:aws:iam::aws:policy/service-role/AmazonECSTaskExecutionRolePolicy
```
Remember to replace the --role-name and --policy-arn with the actual role name and policy Amazon Resource Name (ARN) you want to use.

Now, you have a role named ecsTaskExecutionRole that you can assign to your ECS tasks. This role has the necessary permissions to pull container images and publish logs to CloudWatch.

Launch an ECS Service to host the worker

Next, create an ECS task definition that specifies the Docker image for the Prefect worker, the resources it requires, and the command it should run. In this example, the command to start the worker is prefect worker start --pool my-ecs-pool.

Create a JSON file with the following contents:

{
    "family": "prefect-worker-task",
    "networkMode": "awsvpc",
    "requiresCompatibilities": [
        "FARGATE"
    ],
    "cpu": "512",
    "memory": "1024",
    "executionRoleArn": "<your-ecs-task-role-arn>",
    "taskRoleArn": "<your-ecs-task-role-arn>",
    "containerDefinitions": [
        {
            "name": "prefect-worker",
            "image": "prefecthq/prefect:2-latest",
            "cpu": 512,
            "memory": 1024,
            "essential": true,
            "command": [
                "/bin/sh",
                "-c",
                "pip install prefect-aws && prefect worker start --pool my-ecs-pool --type ecs"
            ],
            "environment": [
                {
                    "name": "PREFECT_API_URL",
                    "value": "https://api.prefect.cloud/api/accounts/<your-account-id>/workspaces/<your-workspace-id>"
                },
                {
                    "name": "PREFECT_API_KEY",
                    "value": "<your-prefect-api-key>"
                }
            ]
        }
    ]
}

Use prefect config view to view the PREFECT_API_URL for your current Prefect profile. Use this to replace both <your-account-id> and <your-workspace-id>.
For the PREFECT_API_KEY, individuals on the organization tier can create a service account for the worker. If on a personal tier, you can pass a user’s API key.
Replace both instances of <your-ecs-task-role-arn> with the ARN of the IAM role you created in Step 2.
Notice that the CPU and Memory allocations are relatively small. The worker's main responsibility is to submit work through API calls to AWS, not to execute your Prefect flow code.

Tip

To avoid hardcoding your API key into the task definition JSON see how to add environment variables to the container definition. The API key must be stored as plain text, not the key-value pair dictionary that it is formatted in by default.

Register the task definition:

Before creating a service, you first need to register a task definition. You can do that using the register-task-definition command in the AWS CLI. Here is an example:
```
aws ecs register-task-definition --cli-input-json file://task-definition.json
```
Replace task-definition.json with the name of your JSON file.
Create an ECS service to host your worker:

Finally, create a service that will manage your Prefect worker:

Open a terminal window and run the following command to create an ECS Fargate service:
```
aws ecs create-service \
    --service-name prefect-worker-service \
    --cluster <your-ecs-cluster> \
    --task-definition <task-definition-arn> \
    --launch-type FARGATE \
    --desired-count 1 \
    --network-configuration "awsvpcConfiguration={subnets=[<your-subnet-ids>],securityGroups=[<your-security-group-ids>]}"
```
- Replace <your-ecs-cluster> with the name of your ECS cluster.
- Replace <path-to-task-definition-file> with the path to the JSON file you created in Step 2, <your-subnet-ids> with a comma-separated list of your VPC subnet IDs. Ensure that these subnets are aligned with the vpc specified on the work pool in step 1.
- Replace <your-security-group-ids> with a comma-separated list of your VPC security group IDs.
- Replace <task-definition-arn> with the ARN of the task definition you just registered.
Sanity check

The work pool page in the Prefect UI allows you to check the health of your workers - make sure your new worker is live!

Step 4: Pick up a flow run with your new worker!

Write a simple test flow in a repo of your choice:

my_flow.py

from prefect import flow, get_run_logger

@flow
def my_flow():
    logger = get_run_logger()
    logger.info("Hello from ECS!!")

if __name__ == "__main__":
    my_flow()

Deploy the flow to the server, specifying the ECS work pool when prompted.
```
prefect deploy my_flow.py:my_flow
```
Find the deployment in the UI and click the Quick Run button!

Optional Next Steps

Now that you are confident your ECS worker is healthy, you can experiment with different work pool configurations.
- Do your flow runs require higher CPU?
- Would an EC2 Launch Type speed up your flow run execution?
These infrastructure configuration values can be set on your ECS work pool or they can be overridden on the deployment level through job_variables if desired.
Consider adding a build action to your Prefect Project prefect.yaml if you want to automatically build a Docker image and push it to an image registry prefect deploy is run.

Here is an example build action for ECR:

build:
- prefect.deployments.steps.run_shell_script:
    id: get-commit-hash
    script: git rev-parse --short HEAD
    stream_output: false
- prefect.deployments.steps.run_shell_script:
    id: ecr-auth-step
    script: aws ecr get-login-password --region <region> | docker login --username
        AWS --password-stdin <>.dkr.ecr.<region>.amazonaws.com
    stream_output: false
- prefect_docker.deployments.steps.build_docker_image:
    requires: prefect-docker>=0.3.0
    image_name: <your-AWS-account-number>.dkr.ecr.us-east-2.amazonaws.com/<registry>
    tag: '{{ get-commit-hash.stdout }}'
    dockerfile: auto
    push: true