EcsRunTaskOperator fails when no containers are provided in the response

[andreas-thomik]

Apache Airflow Provider(s)

amazon

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==9.7.0

Apache Airflow version

3.0.1

Operating System

Debian GNU/Linux 12 (bookworm)

Deployment

Docker-Compose

Deployment details

Pretty much default docker-compose deployment, but using a persistent DB instead of the postgres container.

What happened

The EcsRunTaskOperator fails when the 'containers' list is empty. This is because it tries to infer the container_name variable from the response, without checking whether it exists or not.

We've recently updated Airflow, which highlighted the problem. Our response (see below) has been the same for a long time. As you'll notice, the 'containers' list is empty. This might be because the status is still provisioning, though the AWS documentation is quite sparse on the matter.

{
    "tasks": [
        {
            "attachments": [],
            "capacityProviderName": "Infra-ECS-Cluster-***-EC2CapacityProvider-***",
            "clusterArn": "arn:aws:ecs:***:***:cluster/***",
            "containers": [],
            "cpu": "1024",
            "createdAt": datetime.datetime(
                2025, 6, 5, 8, 55, 51, 381000, tzinfo=tzlocal()
            ),
            "desiredStatus": "RUNNING",
            "enableExecuteCommand": False,
            "group": "family:***",
            "lastStatus": "PROVISIONING",
            "launchType": "EC2",
            "memory": "4096",
            "overrides": {
                "containerOverrides": [{ ... }],
                "inferenceAcceleratorOverrides": [],
            },
            "startedBy": "airflow",
            "tags": [],
            "taskArn": "arn:aws:ecs:***:***:task/***/***",
            "taskDefinitionArn": "arn:aws:ecs:***:***:task-definition/***",
            "version": 1,
        }
    ],
    "failures": [],
    "ResponseMetadata": {
        "RequestId": "***",
        "HTTPStatusCode": 200,
        "HTTPHeaders": {
            "x-amzn-requestid": "***",
            "content-type": "application/x-amz-json-1.1",
            "content-length": "1360",
            "date": "Thu, 05 Jun 2025 08:55:51 GMT",
        },
        "RetryAttempts": 0,
    },
}

The code currently picks the first container without checking for its existence:

        self.arn = response["tasks"][0]["taskArn"]
        if not self.container_name:
            self.container_name = response["tasks"][0]["containers"][0]["name"]

What you think should happen instead

Either handle the case of missing container names gracefully, or poll the task until a container name is available.

How to reproduce

Run a task on ECS which returns a response with an empty array in the 'containers' field.

Anything else

Not all tasks seem to have an empty containers field, but those which do, do so consistently.
The AWS boto3 documentation does not list the field as required, it's therefore not clear whether or not this is an upstream issue or expected behaviour, but either way it would be useful to handle it more gracefully here.

The container_name field seems to be a relatively recent addition (426dba0), I'm not quite clear on its usage and whether we can simply not set it without causing downstream problems. Any insights would be appreciated!

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[dominikhei]

The issue seems to be here:

if not self.container_name:
    self.container_name = response["tasks"][0]["containers"][0]["name"]

If I understand it correctly ECS.Client.run_task(**kwargs) does not guarantee that all information is returned as the task might not be in the ACTIVE state yet. Maybe we should add a little delay and poll the name using describe_tasks if it's not set.
We could also make it a required parameter, although it needs to be unique per ECS task definition.

@eladkal @o-nikolas What do you think?

[AryanK1511]

The issue seems to be here:

if not self.container_name:
self.container_name = response["tasks"][0]["containers"][0]["name"]
If I understand it correctly ECS.Client.run_task(**kwargs) does not guarantee that all information is returned as the task might not be in the ACTIVE state yet. Maybe we should add a little delay and poll the name using describe_tasks if it's not set. We could also make it a required parameter, although it needs to be unique per ECS task definition.

@eladkal @o-nikolas What do you think?

This discussion seems super exciting. I am more than willing to pick issue up once you guys decide on the solution that you want to opt.

[andreas-thomik]

The only place where 'container_name' really gets used is here:

    def _get_logs_stream_name(self) -> str:
        if (
            self.awslogs_stream_prefix
            and self.container_name
            and not self.awslogs_stream_prefix.endswith(f"/{self.container_name}")
        ):
            return f"{self.awslogs_stream_prefix}/{self.container_name}/{self._get_ecs_task_id(self.arn)}"
        return f"{self.awslogs_stream_prefix}/{self._get_ecs_task_id(self.arn)}"

Setting it to None in the event that the reponse doesn't return a valid list of containers might lead to missing logs, but wouldn't fail the task as it does at the moment. Maybe we could simply issue a warning if the awslogs_stream_prefix does not contain at least a / and the list of containers is empty? I'm concerned that polling for ACTIVE might take a long time, especially when new EC2 instances have to be created.

[dominikhei]

If the LogStream Airflow is trying to fetch does not exist , the get_log_events method in the logs hook should throw a CloudWatchLogs.Client.exceptions.ResourceNotFoundExceptio, that is not handled in the get_last_log_messages in the task_logs_fetcher, which again is used and not handled in the _check_success_task methods in the ECS operator.

            response = self.conn.get_log_events(
                **prune_dict(
                    {
                        "logGroupName": log_group,
                        "logStreamName": log_stream_name,
                        "startTime": start_time,
                        "endTime": end_time,
                        "startFromHead": True,
                        **token_arg,
                    }
                )
            )

            events = response["events"]

Maybe that's where the task failure stems from?

[andreas-thomik]

The failure comes from the code you identified earlier, where we try to access the first element of the containers list without checking if it exists.

if not self.container_name:
    self.container_name = response["tasks"][0]["containers"][0]["name"]

My suggestions would be to:

1. Add a check in the code above and set self.container_name = None if it fails
2. Issue a warning to the user if the container name cannot be inferred from the response and no / exists in the awslogs_stream_prefix provided by the user.
   • Note also that getting the container name is entirely un-necessary if the user has not provided awslogs_stream_prefix since in that case we don't care about logging anyway.
3. (Optional) Graciously handle exceptions in get_last_log_messages to make it consistent with the behaviour in _get_log_events

Happy to open a PR if you agree @dominikhei

[dominikhei]

I would skip the container_name logic entirely, if the user does not want to log.

If you set self.container_name = None if it fails, we run in another problem if awslogs_group and awslogs_stream_prefix is set as boto3 will not find the respective log stream. That should be addressed.

I am still in favor of adding a 40s delay or exponential backoff here if the name can't be polled immediately, that should be sufficient for most non bloated images.