Raise run_task errors (#6993)

python_modules/libraries/dagster-aws/dagster_aws/ecs/launcher.py

@@ -177,7 +177,19 @@ def launch_run(self, context: LaunchRunContext) -> None:
             launchType="FARGATE",
         )
 
-        arn = response["tasks"][0]["taskArn"]
+        tasks = response["tasks"]
+
+        if not tasks:
+            failures = response["failures"]
+            exceptions = []
+            for failure in failures:
+                arn = failure.get("arn")
+                reason = failure.get("reason")
+                detail = failure.get("detail")
+                exceptions.append(Exception(f"Task {arn} failed because {reason}: {detail}"))
+            raise Exception(exceptions)
+
+        arn = tasks[0]["taskArn"]
         self._set_run_tags(run.run_id, task_arn=arn)
         self._set_ecs_tags(run.run_id, task_arn=arn)
         self._instance.report_engine_event(

python_modules/libraries/dagster-aws/dagster_aws_tests/ecs_tests/launcher_tests/test_failures.py

@@ -0,0 +1,28 @@
+import pytest
+
+
+def test_run_task_failure(ecs, instance, workspace, run):
+    def run_task(self=ecs, **kwargs):
+        self.stubber.activate()
+        self.stubber.add_response(
+            method="run_task",
+            service_response={
+                "tasks": [],
+                "failures": [
+                    {"arn": "failing-arn-1", "reason": "boom", "detail": "detailed boom 1"},
+                    {"arn": "failing-arn-2", "reason": "boom", "detail": "detailed boom 2"},
+                ],
+            },
+            expected_params={**kwargs},
+        )
+        response = self.client.run_task(**kwargs)
+        self.stubber.deactivate()
+        return response
+
+    instance.run_launcher.ecs.run_task = run_task
+
+    with pytest.raises(Exception) as ex:
+        instance.launch_run(run.run_id, workspace)
+
+    assert ex.match("Task failing-arn-1 failed because boom: detailed boom 1")
+    assert ex.match("Task failing-arn-2 failed because boom: detailed boom 2")