Log failure reason for containers if a task fails for ECS Executor

airflow/providers/amazon/aws/executors/ecs/boto_schema.py

@@ -36,6 +36,8 @@ class BotoContainerSchema(Schema):
     exit_code = fields.Integer(data_key="exitCode")
     last_status = fields.String(data_key="lastStatus")
     name = fields.String(required=True)
+    reason = fields.String()
+    container_arn = fields.String(data_key="containerArn")
 
     class Meta:
         """Options object for a Schema. See Schema.Meta for more details and valid values."""

airflow/providers/amazon/aws/executors/ecs/ecs_executor.py

@@ -191,13 +191,14 @@ def __update_running_task(self, task):
         # Mark finished tasks as either a success/failure.
         if task_state == State.FAILED:
             self.fail(task_key)
+            self.__log_container_failures(task_arn=task.task_arn)
         elif task_state == State.SUCCESS:
             self.success(task_key)
         elif task_state == State.REMOVED:
             self.__handle_failed_task(task.task_arn, task.stopped_reason)
         if task_state in (State.FAILED, State.SUCCESS):
             self.log.debug(
-                "Airflow task %s marked as %s after running on %s",
+                "Airflow task %s marked as %s after running on ECS Task (arn) %s",
                 task_key,
                 task_state,
                 task.task_arn,
@@ -217,6 +218,21 @@ def __describe_tasks(self, task_arns):
             all_task_descriptions["failures"].extend(describe_tasks_response["failures"])
         return all_task_descriptions
 
+    def __log_container_failures(self, task_arn: str):
+        """Check if the task failed due to issues with the containers."""
+        message = "The ECS task failed due to the following containers failing: \n"
+        containers = self.active_workers.task_by_arn(task_arn).containers
+        has_exit_codes = all(["exit_code" in x for x in containers])
+        if not has_exit_codes:
+            return ""
+        reasons = [
+            f'{container["container_arn"]} - {container["reason"]}'
+            for container in containers
+            if "reason" in container
+        ]
+
+        self.log.warning(message + "\n".join(reasons)) if reasons else ""
+
     def __handle_failed_task(self, task_arn: str, reason: str):
         """If an API failure occurs, the task is rescheduled."""
         task_key = self.active_workers.arn_to_key[task_arn]

tests/providers/amazon/aws/executors/ecs/test_ecs_executor.py

@@ -731,6 +731,92 @@ def _sync_mock_with_call_counts(self, sync_func: Callable):
         sync_func()
         self.sync_call_count += 1
 
+    @pytest.mark.parametrize(
+        "desired_status, last_status, exit_code, expected_status",
+        [
+            ("RUNNING", "QUEUED", 0, State.QUEUED),
+            ("STOPPED", "RUNNING", 0, State.RUNNING),
+            ("STOPPED", "QUEUED", 0, State.REMOVED),
+        ],
+    )
+    def test_update_running_tasks(
+        self, mock_executor, desired_status, last_status, exit_code, expected_status
+    ):
+        self._add_mock_task(mock_executor, ARN1)
+        test_response_task_json = {
+            "taskArn": ARN1,
+            "desiredStatus": desired_status,
+            "lastStatus": last_status,
+            "containers": [
+                {
+                    "name": "test_container",
+                    "lastStatus": "QUEUED",
+                    "exitCode": exit_code,
+                }
+            ],
+        }
+        mock_executor.ecs.describe_tasks.return_value = {"tasks": [test_response_task_json], "failures": []}
+        mock_executor.sync_running_tasks()
+        assert mock_executor.active_workers.tasks["arn1"].get_task_state() == expected_status
+        # The task is not removed from active_workers in these states
+        assert len(mock_executor.active_workers) == 1
+
+    def test_update_running_tasks_success(self, mock_executor):
+        self._add_mock_task(mock_executor, ARN1)
+        test_response_task_json = {
+            "taskArn": ARN1,
+            "desiredStatus": "STOPPED",
+            "lastStatus": "STOPPED",
+            "startedAt": dt.datetime.now(),
+            "containers": [
+                {
+                    "name": "test_container",
+                    "lastStatus": "STOPPED",
+                    "exitCode": 0,
+                }
+            ],
+        }
+        patcher = mock.patch(
+            "airflow.providers.amazon.aws.executors.ecs.ecs_executor.AwsEcsExecutor.success", auth_spec=True
+        )
+        mock_success_function = patcher.start()
+        mock_executor.ecs.describe_tasks.return_value = {"tasks": [test_response_task_json], "failures": []}
+        mock_executor.sync_running_tasks()
+        assert len(mock_executor.active_workers) == 0
+        mock_success_function.assert_called_once()
+
+    def test_update_running_tasks_failed(self, mock_executor, caplog):
+        caplog.set_level(logging.WARNING)
+        self._add_mock_task(mock_executor, ARN1)
+        test_response_task_json = {
+            "taskArn": ARN1,
+            "desiredStatus": "STOPPED",
+            "lastStatus": "STOPPED",
+            "startedAt": dt.datetime.now(),
+            "containers": [
+                {
+                    "containerArn": "test-container-arn1",
+                    "name": "test_container",
+                    "lastStatus": "STOPPED",
+                    "exitCode": 30,
+                    "reason": "test failure",
+                }
+            ],
+        }
+
+        patcher = mock.patch(
+            "airflow.providers.amazon.aws.executors.ecs.ecs_executor.AwsEcsExecutor.fail", auth_spec=True
+        )
+        mock_failed_function = patcher.start()
+        mock_executor.ecs.describe_tasks.return_value = {"tasks": [test_response_task_json], "failures": []}
+        mock_executor.sync_running_tasks()
+        assert len(mock_executor.active_workers) == 0
+        mock_failed_function.assert_called_once()
+        assert (
+            "The ECS task failed due to the following containers failing: \ntest-container-arn1 - "
+            "test failure" in caplog.messages[0]
+        )
+
 
 class TestEcsExecutorConfig:
     @pytest.fixture()