ashb closed pull request #4039: [AIRFLOW-3046] Fix ECS Operator mistakenly
reports success
URL: https://github.com/apache/incubator-airflow/pull/4039
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/airflow/contrib/operators/ecs_operator.py
b/airflow/contrib/operators/ecs_operator.py
index 8bad285ffd..2217398512 100644
--- a/airflow/contrib/operators/ecs_operator.py
+++ b/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 \
diff --git a/tests/contrib/operators/test_ecs_operator.py
b/tests/contrib/operators/test_ecs_operator.py
index 5f8c220260..51391c7b1e 100644
--- a/tests/contrib/operators/test_ecs_operator.py
+++ b/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__':
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
