**Description**

Currently when the ECSOperator fails because the ECS task is not in 'success' state it returns a generic message like that in Airflow alerts that doesn't have much value when we want to debug things quickly.

`This task is not in success state {<huge JSON from AWS containing all the ECS task details>}`


**Use case / motivation**

This is to make it faster for people to fix an issue when a task running ECSOperator fails.

**Proposal**

The idea would be to return instead the last lines of logs from Cloudwatch (that are printed above in Airflow logs) so when we receive the alert we know what failed in the ECS task instead of having to go to Airflow logs to find it. This feature would involve changes there I think:
- https://github.com/apache/airflow/blob/2ce6e8de53adc98dd3ae80fa7c74b38eb352bc3a/airflow/providers/amazon/aws/operators/ecs.py#L354
- https://github.com/apache/airflow/blob/2ce6e8de53adc98dd3ae80fa7c74b38eb352bc3a/airflow/providers/amazon/aws/operators/ecs.py#L375


airflow/providers/amazon/aws/operators/ecs.py
#
#
#
import re
import sys
from collections import deque
from datetime import datetime
from typing import Dict, Generator, Optional
from botocore.waiter import Waiter
from airflow.exceptions import AirflowException
from airflow.models import BaseOperator, XCom
from airflow.providers.amazon.aws.exceptions import ECSOperatorError
from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
from airflow.providers.amazon.aws.hooks.logs import AwsLogsHook
from airflow.typing_compat import Protocol, runtime_checkable
from airflow.utils.session import provide_session
def should_retry(exception: Exception):
    """Check if exception is related to ECS resource quota (CPU, MEM)."""
    if isinstance(exception, ECSOperatorError):
        return any(
            quota_reason in failure['reason']
            for quota_reason in ['RESOURCE:MEMORY', 'RESOURCE:CPU']
            for failure in exception.failures
        )
    return False
@runtime_checkable
class ECSProtocol(Protocol):