Short answer: your running flows keep running. Details (Prefect 3.x, ECS worker): - Already-launched flow runs are ECS tasks; they continue executing independently of the worker process. Restarting the worker does not stop or restart those tasks. - State and logs: - If the flow container starts and can reach Prefect, it will keep reporting state/logs directly to Prefect while the worker is down. - For failures that occur before the Prefect engine connects (e.g., image/premature container crash), set up the ECS Observer (EventBridge → SQS). It will detect task state changes and update Prefect when the worker resumes consuming events; SQS retains messages in the meantime. - Cancellation/termination: - If you click Cancel while the worker is offline, Prefect will mark the run as “Cancelling” but cannot call ECS StopTask until a worker is available. The ECS task will keep running; when a worker comes back, it can process the cancellation and stop the task. If no worker returns, the task will run to its natural completion. - Scheduling: - New scheduled runs won’t be picked up until the worker is back online. Follow-up question: Do you have the ECS Observer (EventBridge → SQS) configured for your work pool? If so, Prefect will reliably catch task crashes and final states even across worker restarts. If not, I can share the setup steps.

Great — here’s a clean, verified setup for enabling the ECS Observer with an ECS worker in Prefect 3.x. What the Observer does - Listens for ECS Task State Change events (via EventBridge → SQS) - Updates Prefect states for tasks that never connected (e.g., image/startup failures) - Optionally forwards CloudWatch logs for crashed runs to the flow run Prereqs - An existing ECS cluster and a running Prefect ECS worker service - The prefect-aws integration installed where you’ll run the helper CLI (optional but recommended) - Install: pip install prefect-aws - Repo: PrefectHQ/prefect-aws - Worker task role with permissions listed below Option A (recommended): Use the prefect-aws CLI to deploy the EventBridge + SQS resources 1) Create the “events” stack (SQS + DLQ + EventBridge rule + queue policy)

prefect-aws ecs-worker deploy-events \
  --work-pool-name "my-ecs-pool" \
  --stack-name "my-ecs-pool-events" \
  --existing-cluster-arn "arn:aws:ecs:us-east-1:123456789012:cluster/my-cluster" \
  --region us-east-1

- This creates: - SQS queue: my-ecs-pool-ecs-events (plus a DLQ with receive count 3) - EventBridge rule: my-ecs-pool-ecs-task-events (ECS Task State Change events for your cluster → SQS) - Source code for this command and resources: - CLI: prefect_aws/_cli/ecs_worker.py - EventBridge/SQS stack: prefect_aws/infra/worker/events_stack.py 2) Set the required environment variables on your worker task - Add these to your ECS worker task definition/service:

PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_ENABLED=true
PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_SQS_QUEUE_NAME=my-ecs-pool-ecs-events
# Optional if queue is in a different/default region is ambiguous
PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_SQS_QUEUE_REGION=us-east-1
# Optional (enabled by default) – forwards CloudWatch logs from crashed runs
PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_FORWARD_CRASHED_RUN_LOGS=true
PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_FORWARD_CRASHED_RUN_LOGS_MAX_EVENTS=500

- The observer starts automatically within the ECS worker when these are set. - Observer implementation: prefect_aws/observers/ecs.py - Worker integration (auto-start): prefect_aws/workers/ecs_worker.py Option B: Manual/IaC approach - If you prefer CloudFormation/Terraform, create: - SQS queue (name you’ll set in PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_SQS_QUEUE_NAME) - Visibility timeout ~300s, message retention ~7 days, DLQ with maxReceiveCount=3 - Queue policy permitting events.amazonaws.com to SendMessage - EventBridge rule matching:

{
  "source": [
    "aws.ecs"
  ],
  "detail-type": [
    "ECS Task State Change"
  ],
  "detail": {
    "clusterArn": [
      "arn:aws:ecs:REGION:ACCOUNT:cluster/your-cluster"
    ]
  }
}

Target the SQS queue - Then set the same environment variables on the worker as in Option A. Worker task role (IAM) permissions - Ensure the ECS worker task role has these (min set to support observer + normal ECS operations): ``` { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "sqs:ReceiveMessage", "sqs:DeleteMessage", "sqs:GetQueueAttributes", "sqs:GetQueueUrl" ], "Resource": "arnawssqsREGIONACCOUNT-ID:my-ecs-pool-ecs-events*" }, { "Effect": "Allow", "Action": [ "ecs:DescribeTasks", "ecs:DescribeTaskDefinition", "ecs:ListTasks", "ecs:RunTask", "ecs:StopTask", "ecs:DescribeClusters", "ecs:ListClusters", "ecs:RegisterTaskDefinition", "ecs:DeregisterTaskDefinition",

"ecs:ListTaskDefinitions", "ecs:TagResource" ], "Resource": "*" }, { "Effect": "Allow", "Action": [ "logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents", "logs:GetLogEvents", "logs:DescribeLogStreams" ], "Resource": "*" }, { "Effect": "Allow", "Action": ["iam:PassRole"], "Resource": "arnawsiam::*:role/*" } ] }

- IAM source refs: <https://github.com/PrefectHQ/prefect-aws/blob/main/prefect_aws/infra/worker/service_stack.py|prefect_aws/infra/worker/service_stack.py>

How to test it
1) Verify infra
- SQS queue exists:

aws sqs list-queues --queue-name-prefix "my-ecs-pool-ecs-events" --region us-east-1

- EventBridge rule exists and targets the queue:

aws events describe-rule --name "my-ecs-pool-ecs-task-events" --region us-east-1 aws events list-targets-by-rule --rule "my-ecs-pool-ecs-task-events" --region us-east-1 ``` 2) Check worker logs for observer startup - In CloudWatch logs for your worker container, you should see logs indicating the ECS observer started and is polling SQS. 3) Run a test flow and confirm state transitions - Start a normal flow: it should move through InfrastructurePending → Running → Completed. - Induce a pre-start failure (e.g., bad image or a flow that raises at import): the run should be marked Crashed and, if enabled, CloudWatch logs forwarded to the flow run. Useful references - ECS Observer code: prefect_aws/observers/ecs.py - EventBridge/SQS stack: prefect_aws/infra/worker/events_stack.py - ECS worker (observer auto-start integration): prefect_aws/workers/ecs_worker.py - Prefect deployments in 3.x: Deployment overview If you share your: - Work pool name - ECS cluster ARN - AWS region …I can generate copy-paste CLI and/or a minimal CloudFormation snippet tailored to your environment.

I’m here! How can I help? If you’ve got a Prefect question, share what you’re trying to do (and your Prefect version, likely 3.x), plus any code/error snippets—happy to jump in.

@Marvin Please see my request above in thread. I am looking for details on your recommended ECS Observer implementation. Specifically, can you provide a cloudformation or CDK template?

Absolutely—happy to share a reference implementation for the ECS Observer. In Prefect 3.x, the “ECS Observer” runs inside your ECS worker container and relies on AWS EventBridge + SQS to receive ECS Task State Change events. You can set this up with either: - Events-only infrastructure (SQS + EventBridge) and point your existing ECS worker at the queue, or - A full ECS worker service stack (includes the observer infra + worker service) Below are ready-to-use templates and pointers to the official sources. Key links - ECS Observer code: prefect_aws.observers.ecs - Observer settings (env vars): prefect_aws.settings - Prefect AWS integration source: prefect_aws package Option A: CloudFormation (events-only) This stack sets up SQS + EventBridge. Point your worker at the queue via env var

PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_SQS_QUEUE_NAME

. It’s safe to deploy alongside any existing ECS worker setup.

AWSTemplateFormatVersion: '2010-09-09'
Description: Prefect ECS Observer - events-only (EventBridge -> SQS)

Parameters:
  ClusterArn:
    Type: String
    Description: ARN of the ECS cluster to observe
  QueueName:
    Type: String
    Default: prefect-ecs-tasks-events
    Description: Name for the SQS queue to receive ECS task state change events

Resources:
  Dlq:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: !Sub '${QueueName}-dlq'
      MessageRetentionPeriod: 1209600  # 14 days

  EventsQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: !Ref QueueName
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt Dlq.Arn
        maxReceiveCount: 5
      VisibilityTimeout: 30
      MessageRetentionPeriod: 1209600

  EcsTaskStateChangeRule:
    Type: AWS::Events::Rule
    Properties:
      Name: !Sub 'prefect-ecs-task-events-${AWS::StackName}'
      Description: Publishes ECS task state changes for Prefect observer
      EventPattern:
        source:
          - aws.ecs
        detail-type:
          - ECS Task State Change
        detail:
          clusterArn:
            - !Ref ClusterArn
      State: ENABLED
      Targets:
        - Arn: !GetAtt EventsQueue.Arn
          Id: SendToSqs

  EventsQueuePolicy:
    Type: AWS::SQS::QueuePolicy
    Properties:
      Queues:
        - !Ref EventsQueue
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Sid: AllowEventBridgeToSend
            Effect: Allow
            Principal:
              Service: <http://events.amazonaws.com|events.amazonaws.com>
            Action: SQS:SendMessage
            Resource: !GetAtt EventsQueue.Arn
            Condition:
              ArnEquals:
                aws:SourceArn: !GetAtt EcsTaskStateChangeRule.Arn

Outputs:
  QueueName:
    Value: !GetAtt EventsQueue.QueueName
  QueueUrl:
    Value: !Ref EventsQueue
  QueueArn:
    Value: !GetAtt EventsQueue.Arn

Worker configuration (env + IAM) - Set these on your ECS worker TaskDefinition (container env): -

PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_ENABLED=true

(default is true) -

PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_SQS_QUEUE_NAME=<your queue name>

- Optional:

PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_SQS_QUEUE_REGION=<region>

- Optional:

PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_FORWARD_CRASHED_RUN_LOGS=true

and

PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_FORWARD_CRASHED_RUN_LOGS_MAX_EVENTS=500

- Add these permissions to the worker task role (restrict Resource to your queue ARN(s)): ``` { "Version": "2012-10-17", "Statement": [ { "Sid": "EcsObserverSqsReceive", "Effect": "Allow", "Action": [ "sqs:ReceiveMessage", "sqs:DeleteMessage", "sqs:GetQueueAttributes", "sqs:GetQueueUrl" ], "Resource": "arnawssqs&lt;region&gt;<account-id>:<queue-name>" }, {

"Sid": "OptionalCloudWatchLogsReadForCrashForwarding", "Effect": "Allow", "Action": [ "logs:FilterLogEvents", "logs:GetLogEvents", "logs:DescribeLogStreams" ], "Resource": "*" } ] }

Option B: CDK (events-only)
A minimal CDK example in TypeScript that mirrors the CFN above:

import * as cdk from 'aws-cdk-lib'; import { aws_sqs as sqs, aws_events as events, aws_events_targets as targets, aws_iam as iam } from 'aws-cdk-lib'; export interface EcsObserverEventsStackProps extends cdk.StackProps { clusterArn: string; queueName?: string; } export class EcsObserverEventsStack extends cdk.Stack { constructor(scope: cdk.App, id: string, props: EcsObserverEventsStackProps) { super(scope, id, props); const dlq = new sqs.Queue(this, 'Dlq', { queueName:

${props.queueName ?? 'prefect-ecs-tasks-events'}-dlq

, retentionPeriod: cdk.Duration.days(14), }); const queue = new sqs.Queue(this, 'EventsQueue', { queueName: props.queueName ?? 'prefect-ecs-tasks-events', visibilityTimeout: cdk.Duration.seconds(30), retentionPeriod: cdk.Duration.days(14), deadLetterQueue: { queue: dlq, maxReceiveCount: 5 }, }); const rule = new events.Rule(this, 'EcsTaskStateChangeRule', { description: 'Publishes ECS task state changes for Prefect observer', eventPattern: { source: ['aws.ecs'], detailType: ['ECS Task State Change'], detail: { clusterArn: [props.clusterArn] }, }, }); rule.addTarget(new targets.SqsQueue(queue)); queue.addToResourcePolicy(new iam.PolicyStatement({ effect: iam.Effect.ALLOW, principals: [new iam.ServicePrincipal('events.amazonaws.com')], actions: ['sqs:SendMessage'], resources: [queue.queueArn], conditions: { ArnEquals: { 'aws:SourceArn': rule.ruleArn } }, })); } } ``` Need the full worker + observer as IaC? - Prefect’s ECS worker service stack (which includes the observer infra, worker task definition, IAM, autoscaling, and logging) is maintained in the Prefect AWS integration. If you’d like, I can export the exact CloudFormation template we ship and paste it here, or tailor it to your VPC/cluster. - Source entry points you can reference: - Observer code and settings: - ECS Observer - Observer settings - Package root: - prefect_aws Quick check questions so I can tailor the template for you - Do you want events-only (to plug into an existing worker) or a full worker service + observer? - Fargate or EC2 launch type? - Your ECS cluster ARN, VPC ID, and private subnet IDs? - Desired worker scaling bounds (min/max tasks) and container sizing (CPU/memory)? - Prefect Cloud or self-hosted (so I can wire in the right auth env)? If you share those, I’ll generate a CloudFormation/CDK stack that you can drop directly into your account.

Great question. In short, the ECS observer listens for ECS Task State Change events (via EventBridge → SQS), correlates them to Prefect flow runs, enriches observability, and handles crash scenarios. Here’s what it does with each event: - Correlates the ECS task to a Prefect run - Reads ECS task tags (e.g., flow-run ID, deployment/work-pool) to find the corresponding Prefect run. - Reference: prefect_aws.observers.ecs (see EcsTaskTagsReader and EcsEventHandler) - Replicates the ECS event into Prefect’s event system - Converts the ECS event into a Prefect “event” for your timeline/observability, including metadata such as task ARN, cluster, lastStatus, exit code, and reason. - Reference:

replicate_ecs_event

in ecs.py - Bridges gaps before a flow connects - For tasks that crash before the flow code can connect to Prefect, the observer still records what happened and surfaces details. - Detects crashes and updates run state when appropriate - If a task transitions to STOPPED with a non‑zero exit code or a failure reason and the corresponding Prefect run has not already reached a terminal state, the observer marks the run as Crashed. - It attaches diagnostic details (exit code, reason) and can fetch a slice of recent CloudWatch logs to include with the crash for faster triage. - References:

mark_runs_as_crashed

,

fetch_cloudwatch_logs

in ecs.py - Optionally performs cleanup - Can deregister task definitions on STOPPED if configured. - Reference:

deregister_task_definition

in ecs.py - Safe-by-default behavior - It will not override states that your flow has already reported (e.g., if the flow has already completed/failed and reported its terminal state, the observer won’t change it). - It focuses on enriching observability and handling “crashed before connect” scenarios. - Robust processing of the SQS stream - Long-polls SQS, applies filtering (by lastStatus and Prefect-specific tags), retries with backoff on transient errors, acknowledges messages on success, and relies on a DLQ after max receives. - References:

SqsSubscriber

and retry/backoff constants in ecs.py Config knobs you can use - Enable/disable and tune behavior via environment variables: -

PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_ENABLED

(default true) -

PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_SQS_QUEUE_NAME

(required for events-only setups) -

PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_SQS_QUEUE_REGION

(optional if cross-region) -

PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_FORWARD_CRASHED_RUN_LOGS

and

..._MAX_EVENTS

- Reference: prefect_aws.settings If you’d like, I can tailor a minimal “events-only” stack (SQS + EventBridge) or a full worker+observer stack for your cluster/VPC and include the exact env and IAM you’ll need.