[AIRFLOW-3046] Report fail from ECS Operator when host terminates (#4039)

Add check for host termination to ECS Operator

If an ECS task fails to complete because the host it's running on is terminated, we need to raise an exception so it can be retried.

Author: danmactough

airflow/contrib/operators/ecs_operator.py

@@ -17,6 +17,7 @@
 # specific language governing permissions and limitations
 # under the License.
 import sys
+import re
 
 from airflow.exceptions import AirflowException
 from airflow.models import BaseOperator
@@ -139,6 +140,15 @@ def _check_success_task(self):
             raise AirflowException(response)
 
         for task in response['tasks']:
+            # This is a `stoppedReason` that indicates a task has not
+            # successfully finished, but there is no other indication of failure
+            # in the response.
+            # See, https://docs.aws.amazon.com/AmazonECS/latest/developerguide/stopped-task-errors.html # noqa E501
+            if re.match(r'Host EC2 \(instance .+?\) (stopped|terminated)\.',
+                        task.get('stoppedReason', '')):
+                raise AirflowException(
+                    'The task was stopped because the host instance terminated: {}'.
+                    format(task.get('stoppedReason', '')))
             containers = task['containers']
             for container in containers:
                 if container.get('lastStatus') == 'STOPPED' and \

tests/contrib/operators/test_ecs_operator.py

@@ -172,8 +172,10 @@ def test_wait_end_tasks(self):
 
         self.ecs._wait_for_task_ended()
         client_mock.get_waiter.assert_called_once_with('tasks_stopped')
-        client_mock.get_waiter.return_value.wait.assert_called_once_with(cluster='c', tasks=['arn'])
-        self.assertEquals(sys.maxsize, client_mock.get_waiter.return_value.config.max_attempts)
+        client_mock.get_waiter.return_value.wait.assert_called_once_with(
+            cluster='c', tasks=['arn'])
+        self.assertEquals(
+            sys.maxsize, client_mock.get_waiter.return_value.config.max_attempts)
 
     def test_check_success_tasks_raises(self):
         client_mock = mock.Mock()
@@ -197,7 +199,8 @@ def test_check_success_tasks_raises(self):
         self.assertIn("'name': 'foo'", str(e.exception))
         self.assertIn("'lastStatus': 'STOPPED'", str(e.exception))
         self.assertIn("'exitCode': 1", str(e.exception))
-        client_mock.describe_tasks.assert_called_once_with(cluster='c', tasks=['arn'])
+        client_mock.describe_tasks.assert_called_once_with(
+            cluster='c', tasks=['arn'])
 
     def test_check_success_tasks_raises_pending(self):
         client_mock = mock.Mock()
@@ -217,7 +220,8 @@ def test_check_success_tasks_raises_pending(self):
         self.assertIn("This task is still pending ", str(e.exception))
         self.assertIn("'name': 'container-name'", str(e.exception))
         self.assertIn("'lastStatus': 'PENDING'", str(e.exception))
-        client_mock.describe_tasks.assert_called_once_with(cluster='c', tasks=['arn'])
+        client_mock.describe_tasks.assert_called_once_with(
+            cluster='c', tasks=['arn'])
 
     def test_check_success_tasks_raises_multiple(self):
         client_mock = mock.Mock()
@@ -236,7 +240,42 @@ def test_check_success_tasks_raises_multiple(self):
             }]
         }
         self.ecs._check_success_task()
-        client_mock.describe_tasks.assert_called_once_with(cluster='c', tasks=['arn'])
+        client_mock.describe_tasks.assert_called_once_with(
+            cluster='c', tasks=['arn'])
+
+    def test_host_terminated_raises(self):
+        client_mock = mock.Mock()
+        self.ecs.client = client_mock
+        self.ecs.arn = 'arn'
+        client_mock.describe_tasks.return_value = {
+            'tasks': [{
+                'stoppedReason': 'Host EC2 (instance i-1234567890abcdef) terminated.',
+                "containers": [
+                    {
+                        "containerArn": "arn:aws:ecs:us-east-1:012345678910:container/e1ed7aac-d9b2-4315-8726-d2432bf11868",  # noqa: E501
+                        "lastStatus": "RUNNING",
+                        "name": "wordpress",
+                        "taskArn": "arn:aws:ecs:us-east-1:012345678910:task/d8c67b3c-ac87-4ffe-a847-4785bc3a8b55"  # noqa: E501
+                    }
+                ],
+                "desiredStatus": "STOPPED",
+                "lastStatus": "STOPPED",
+                "taskArn": "arn:aws:ecs:us-east-1:012345678910:task/d8c67b3c-ac87-4ffe-a847-4785bc3a8b55",  # noqa: E501
+                "taskDefinitionArn": "arn:aws:ecs:us-east-1:012345678910:task-definition/hello_world:11"  # noqa: E501
+
+            }]
+        }
+
+        with self.assertRaises(AirflowException) as e:
+            self.ecs._check_success_task()
+
+        self.assertIn(
+            "The task was stopped because the host instance terminated:",
+            str(e.exception))
+        self.assertIn("Host EC2 (", str(e.exception))
+        self.assertIn(") terminated", str(e.exception))
+        client_mock.describe_tasks.assert_called_once_with(
+            cluster='c', tasks=['arn'])
 
     def test_check_success_task_not_raises(self):
         client_mock = mock.Mock()
@@ -252,7 +291,8 @@ def test_check_success_task_not_raises(self):
             }]
         }
         self.ecs._check_success_task()
-        client_mock.describe_tasks.assert_called_once_with(cluster='c', tasks=['arn'])
+        client_mock.describe_tasks.assert_called_once_with(
+            cluster='c', tasks=['arn'])
 
 
 if __name__ == '__main__':