ECSOperator returns last logs when ECS task fails

[pmalafosse]

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:

• airflow/airflow/providers/amazon/aws/operators/ecs.py

  Line 354 in 2ce6e8d

  for event in self._cloudwatch_log_events():

• airflow/airflow/providers/amazon/aws/operators/ecs.py

  Line 375 in 2ce6e8d

  raise AirflowException(f'This task is not in success state {task}')

[uranusjr]

Would it be better if we include the task object in the exception? If we’re to return anything more specific, we can subclass AirflowException, which allows putting a lot more context on the exception object.

[pmalafosse]

Sorry I don’t understand what you mean. Nowtask object is in the exception but I would like instead to see, in Airflow alerts, the exception from the code that ran in ECS, not the exception defined in the operator that just says “ECS task failed” with a big JSON that I think is irrelevant noise.

[uranusjr]

Could you describe what exactly you would like to see (and probably how that can be implemented)? The value of self._last_log_message()?

[pmalafosse]

Yes I was thinking of doing something like that as I'm not sure there is a consistent way to identify the rows corresponding to the Exception message from the code running in ECS from Cloudwatch logs:

something like raise AirflowException(self._last_log_messages(5)) that would return the last 5 lines from Cloudwatch for example (_last_log_message() would be equal to _last_log_messages(1))

[pmalafosse]

I tried locally and my proposal would be like that:

    def _last_log_messages(self, number_messages):
        try:
            logs = [log["message"] for log in self._cloudwatch_log_events()]
            return "\n".join(logs[-number_messages:])
        except IndexError:
            return None

    def _last_log_message(self):
        return self._last_log_messages(1)

...

raise AirflowException(f"This task is not in success state - last logs from Cloudwatch: \n{self._last_log_messages(10)}")

[uranusjr]

We can probably use collections.deque for better performance, but aside from that, sounds good to me! Would you be interested in opening a pull request for this?

p.s. I don’t think IndexError can ever be raised?

[potiuk]

Assigned you @pmalafosse