fweilun commented on issue #52943:
URL: https://github.com/apache/airflow/issues/52943#issuecomment-3050113817
@potiuk
The original author resets self.arn to None when a `TaskFailedToStart` error
occurs, so that a new ECS task can be started instead of repeatedly checking
the same failed one.
However, reset of `self.arn` isn't reset on retry. Instead, the method
returns early as successful if `self.client` or `self.arn` is None, which leads
to this issue.
I think `self.client` or `self.arn` being None should raise an error instead
of silently returning.
I added the following block to address the issue — could you check if this
solution is appropriate? I’m unsure whether calling `_start_task` here would
cause any issues with the framework.
```python
if not self.arn:
self._start_task()
self._wait_for_task_ended()
```
Revised `_check_success_task`:
```python
@AwsBaseHook.retry(should_retry_eni)
def _check_success_task(self) -> None:
if not self.arn:
self._start_task()
self._wait_for_task_ended()
if not self.client or not self.arn:
return
response = self.client.describe_tasks(cluster=self.cluster,
tasks=[self.arn])
self.log.info("ECS Task stopped, check status: %s", response)
if len(response.get("failures", [])) > 0:
raise AirflowException(response)
for task in response["tasks"]:
if task.get("stopCode", "") == "TaskFailedToStart":
# Reset task arn here otherwise the retry run will not start
# a new task but keep polling the old dead one
# I'm not resetting it for other exceptions here because
# EcsTaskFailToStart is the only exception that's being
retried at the moment
self.arn = None
raise EcsTaskFailToStart(f"The task failed to start due to:
{task.get('stoppedReason', '')}")
# This is a `stoppedReason` that indicates a task has not
# successfully finished, but there is no other indication of
failure
# in the response.
#
https://docs.aws.amazon.com/AmazonECS/latest/developerguide/stopped-task-errors.html
if re.match(r"Host EC2 \(instance .+?\) (stopped|terminated)\.",
task.get("stoppedReason", "")):
raise AirflowException(
f"The task was stopped because the host instance
terminated:"
f" {task.get('stoppedReason', '')}"
)
containers = task["containers"]
for container in containers:
if container.get("lastStatus") == "STOPPED" and
container.get("exitCode", 1) != 0:
if self.task_log_fetcher:
last_logs = "\n".join(
self.task_log_fetcher.get_last_log_messages(self.number_logs_exception)
)
raise AirflowException(
f"This task is not in success state - last
{self.number_logs_exception} "
f"logs from Cloudwatch:\n{last_logs}"
)
raise AirflowException(f"This task is not in success
state {task}")
if container.get("lastStatus") == "PENDING":
raise AirflowException(f"This task is still pending
{task}")
if "error" in container.get("reason", "").lower():
raise AirflowException(
f"This containers encounter an error during
launching: "
f"{container.get('reason', '').lower()}"
)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
