Airflow task is marked as succeeded on the EcsTaskFailToStart exception

[vladimirbinshtok]

Apache Airflow Provider(s)

amazon

Versions of Apache Airflow Providers

When using EcsRunTaskOperator, if an ECS task fails to start (e.g., due to resource constraints or misconfiguration), the _check_success_task method raises EcsTaskFailToStart. The @AwsBaseHook.retry(should_retry_eni) decorator retries this exception. However, the arn is set to None, so the function exits as success.

@AwsBaseHook.retry(should_retry_eni)
def _check_success_task(self) -> None:
    if not self.client or not self.arn:
        return

    response = self.client.describe_tasks(cluster=self.cluster, tasks=[self.arn])
    self.log.info("ECS Task stopped, check status: %s", response)

    if len(response.get("failures", [])) > 0:
        raise AirflowException(response)

    for task in response["tasks"]:
        if task.get("stopCode", "") == "TaskFailedToStart":
            # Reset task arn here otherwise the retry run will not start
            # a new task but keep polling the old dead one
            # I'm not resetting it for other exceptions here because
            # EcsTaskFailToStart is the only exception that's being retried at the moment
            self.arn = None
            raise EcsTaskFailToStart(f"The task failed to start due to: {task.get('stoppedReason', '')}")

Apache Airflow version

2.7.2

Operating System

Ubuntu 22.04

Deployment

Other

Deployment details

No response

What happened

The Airflow task is marked as successful even though the ECS task never started successfully.

What you think should happen instead

The Airflow task should be marked as failed if the ECS task fails to start and all retries are exhausted.

How to reproduce

Create a misconfigured ECS Task definition

Anything else

No response

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.

[fweilun]

I'd like to address this issue. @vladimirbinshtok Could you give me an example on how to reproduce this?

[vladimirbinshtok]

@fweilun The way to reproduce the issue is to create an ECS Task Definition with a wrong command that the container cannot execute (run-script in my example).

In that case, the describe_tasks will return the following JSON.

{
    "tasks": [
        {
            ...
            "stopCode": "TaskFailedToStart",
            "stoppedReason": "CannotStartContainerError: ResourceInitializationError: failed to create new container runtime task: failed to create shim task: OCI runtime create failed: runc create failed: unable to start container process: error during container init: exec: run-script: executable file not found in $PATH: unknown",
            ...
        }
    ],
    "failures": [],
    "ResponseMetadata": {
        ...
    }
}

There is, for some reason, a special handling of TaskFailedToStart ResourceInitializationError, which leads to a retry of _check_success_task function (that without arn returns None):

def should_retry_eni(exception: Exception):
    """Check if exception is related to ENI (Elastic Network Interfaces)."""
    if isinstance(exception, EcsTaskFailToStart):
        return any(
            eni_reason in exception.message
            for eni_reason in ["network interface provisioning", "ResourceInitializationError"]
        )
    return False

I also reproduced the behavior when the task is TaskFailedToStart, but the stoppedReason is 'CannotPullContainerError: pull image manifest has been retried 1 time(s), then Airflow marks the DAG as failed, as expected.

I don't understand the reason for this custom handling of TaskFailedToStart errors that lead to retries in specific cases.

[fweilun]

I used this operator with Airflow 3.1.0 to reproduce the error by adding "command": ["run-script"] to the container definition, which triggered a ResourceInitializationError.

The task is correctly marked as failed, but I’m not sure if the retry logic was triggered. Is there any difference from our logs?

from airflow import DAG
from datetime import datetime
from airflow.providers.amazon.aws.operators.ecs import EcsRunTaskOperator

with DAG(dag_id="test-ecs", start_date=datetime(2024, 1, 1)) as dag:
    task = EcsRunTaskOperator(
        task_id="test-ecs",
        task_definition="testing:2",
        cluster="xxx",
        launch_type="FARGATE",
        platform_version="LATEST",
        network_configuration={
            "awsvpcConfiguration": {
                "subnets": ["subnet-id"],
                "assignPublicIp": "ENABLED"
            }
        },
        overrides={},
        aws_conn_id="aws_conn",
        region_name="us-east-2"
    )

[vladimirbinshtok]

@fweilun I'm getting the following log messages:

Starting call to '***.providers.amazon.aws.operators.ecs.EcsRunTask0perator._check_success_task', this is the lst time calling it.
ECS Task stopped, check status: ('tasks': [{'attachments': [...
Finished call to '***.providers.amazon.aws.operators.ecs.EcsRunTask0perator._check_success_task' after 0.395(s), this was the ist time cal
Starting call to '***.providers.amazon.aws.operators.ecs.EcsRunTask0perator._check_success_task', this is the 2nd time calling it.
Marking task as SUCCESS. dag_id=.... task_id=main_task, execution_date=20250706T080803, start_date=2025076

Starting/Finished call logs come from the tenacity_before_logger and tenacity_after_logger functions that are defined here

@staticmethod
def retry(should_retry: Callable[[Exception], bool]):
    """Repeat requests in response to exceeding a temporary quote limit."""

    def retry_decorator(fun: Callable):
        @wraps(fun)
        def decorator_f(self, *args, **kwargs):
            retry_args = getattr(self, "retry_args", None)
            if retry_args is None:
                return fun(self, *args, **kwargs)
            multiplier = retry_args.get("multiplier", 1)
            min_limit = retry_args.get("min", 1)
            max_limit = retry_args.get("max", 1)
            stop_after_delay = retry_args.get("stop_after_delay", 10)
            tenacity_before_logger = tenacity.before_log(self.log, logging.INFO) if self.log else None
            tenacity_after_logger = tenacity.after_log(self.log, logging.INFO) if self.log else None
            default_kwargs = {
                "wait": tenacity.wait_exponential(multiplier=multiplier, max=max_limit, min=min_limit),
                "retry": tenacity.retry_if_exception(should_retry),
                "stop": tenacity.stop_after_delay(stop_after_delay),
                "before": tenacity_before_logger,
                "after": tenacity_after_logger,
            }
            return tenacity.retry(**default_kwargs)(fun)(self, *args, **kwargs)

        return decorator_f

    return retry_decorator

Also, I missed in the description that I'm using ECS operator in deferable mode.

[fweilun]

The original author resets self.arn to None when a TaskFailedToStart error occurs, so that a new ECS task can be started instead of repeatedly checking the same failed one.

However, reset of self.arn isn't reset on retry. Instead, the method returns early as successful if self.client or self.arn is None, which leads to this issue.

I think self.client or self.arn being None should raise an error instead of silently returning.

I added the following block to address the issue — could you check if this solution is appropriate? I’m unsure whether calling _start_task here would cause any issues with the framework.

if not self.arn:
    self._start_task()
    self._wait_for_task_ended()

Revised _check_success_task:

@AwsBaseHook.retry(should_retry_eni)
    def _check_success_task(self) -> None:
        if not self.arn:
            self._start_task()
            self._wait_for_task_ended()

        if not self.client or not self.arn:
            return

        response = self.client.describe_tasks(cluster=self.cluster, tasks=[self.arn])
        self.log.info("ECS Task stopped, check status: %s", response)

        if len(response.get("failures", [])) > 0:
            raise AirflowException(response)

        for task in response["tasks"]:
            if task.get("stopCode", "") == "TaskFailedToStart":
                # Reset task arn here otherwise the retry run will not start
                # a new task but keep polling the old dead one
                # I'm not resetting it for other exceptions here because
                # EcsTaskFailToStart is the only exception that's being retried at the moment
                self.arn = None
                raise EcsTaskFailToStart(f"The task failed to start due to: {task.get('stoppedReason', '')}")

            # This is a `stoppedReason` that indicates a task has not
            # successfully finished, but there is no other indication of failure
            # in the response.
            # https://docs.aws.amazon.com/AmazonECS/latest/developerguide/stopped-task-errors.html
            if re.match(r"Host EC2 \(instance .+?\) (stopped|terminated)\.", task.get("stoppedReason", "")):
                raise AirflowException(
                    f"The task was stopped because the host instance terminated:"
                    f" {task.get('stoppedReason', '')}"
                )
            containers = task["containers"]
            for container in containers:
                if container.get("lastStatus") == "STOPPED" and container.get("exitCode", 1) != 0:
                    if self.task_log_fetcher:
                        last_logs = "\n".join(
                            self.task_log_fetcher.get_last_log_messages(self.number_logs_exception)
                        )
                        raise AirflowException(
                            f"This task is not in success state - last {self.number_logs_exception} "
                            f"logs from Cloudwatch:\n{last_logs}"
                        )
                    raise AirflowException(f"This task is not in success state {task}")
                if container.get("lastStatus") == "PENDING":
                    raise AirflowException(f"This task is still pending {task}")
                if "error" in container.get("reason", "").lower():
                    raise AirflowException(
                        f"This containers encounter an error during launching: "
                        f"{container.get('reason', '').lower()}"
                    )

[vladimirbinshtok]

@fweilun
According to AWS documentation (https://docs.aws.amazon.com/AmazonECS/latest/developerguide/resource-initialization-error.html), most ResourceInitializationErrors are caused by misconfiguration in the Task Definition or Permissions; so, retry would not help in that case. Probably, there are some edge cases where retry should be implemented. Are there any contact points in AWS that can help define these cases?

I'm using the ESC operator with the following flags: deferrable=True and wait_for_completion=False. Therefore, the function that should be used in the retry is self._execute, not self._start_task. The AwsBaseHook.retry mechanism currently doesn't support this mode, as the current retry attempt is not passed to the next task execution (which would lead to an infinite retry) and can only be performed in-process.

Only ECS Operator use this retry logic; all other operators will fail in any case of AWS internal error. I suggest completely turning off the retry mechanism.