Airflow tasks marked as failed without retry with KubernetesExecutor

[nicor88]

Apache Airflow version

2.2.3 (latest released)

What happened

We use Kubernetes Executor in combinations with AWS spot instances. Time to time spot instances are evicted, therefore the tasks/pods running on it fail. All our tasks are having a retry >=1, which generally works fine, but in this case, the tasks running in the evicted K8s node, are marked immediately as failed.

What you expected to happen

I expected that the scheduler considers the retry, and trigger another instance of the task without marking it immediately as failed.

How to reproduce

Kill a running pod when is running, to obtain the same error described in this issue.

Operating System

Official Docker Airflow image 2.2.3

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==2.4.0
apache-airflow-providers-postgres==2.3.0
apache-airflow-providers-slack==4.1.0

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[ephraimbuddy]

Can you upload the scheduler logs when this happens

[potiuk]

Converted it to discussion until more information is provided that might enable to asses if this is an Airflow issue or not.

[nicor88]

@ephraimbuddy Somehow I missed this.

Somehow this issue happen today, again. The tasks are scheduled in pods that belong to a Ec2 spot instance node.
The spot node goes down, and the tasks pods are marked as failed, without retrying. If would have retry, then the new task will be run a pod in a node that is up, and this issue won't happen.
Here the logs.

Date,Message
"2022-03-16T00:03:26.443Z","[2022-03-16 00:03:26,146] {scheduler_job.py:570} INFO - TaskInstance Finished: dag_id=continuous_load, task_id=load_table, run_id=scheduled__2022-03-15T23:45:00+00:00, run_start_date=2022-03-16 00:01:26.128956+00:00, run_end_date=None, run_duration=None, state=running, executor_state=failed, try_number=1, max_tries=1, job_id=403468, pool=default_pool, queue=default, priority_weight=1, operator=ExtractLoadOperator"
"2022-03-16T00:03:26.442Z","[2022-03-16 00:03:26,069] {kubernetes_executor.py:575} INFO - Changing state of (TaskInstanceKey(dag_id='continuous_load', task_id='load_table', run_id='scheduled__2022-03-15T23:45:00+00:00', try_number=1), , 'load_table.b45adf3ca7ef4f5482639bfedbc4c340', 'airflow-data-eng', '109859634') to failed"
"2022-03-16T00:03:26.442Z","[2022-03-16 00:03:26,064] {kubernetes_executor.py:374} INFO - Attempting to finish pod; pod_id: load_table.b45adf3ca7ef4f5482639bfedbc4c340; state: failed; annotations: {'dag_id': 'continuous_load', 'task_id': 'load_table', 'execution_date': None, 'run_id': 'scheduled__2022-03-15T23:45:00+00:00', 'try_number': '1'}"
"2022-03-16T00:00:21.204Z","[2022-03-16 00:00:20,563] {kubernetes_executor.py:297} INFO - Kubernetes job is (TaskInstanceKey(dag_id='continuous_load', task_id='load_table', run_id='scheduled__2022-03-15T23:45:00+00:00', try_number=1), ['airflow', 'tasks', 'run', 'continuous_load', 'load_table', 'scheduled__2022-03-15T23:45:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/continuous_load.py'], {'api_version': 'v1', 'kind': 'Pod', 'metadata': {'annotations': None, 'cluster_name': None, 'creation_timestamp': None, 'deletion_grace_period_seconds': None, 'deletion_timestamp': None, 'finalizers': None, 'generate_name': None, 'generation': None, 'initializers': None, 'labels': None, 'managed_fields': None, 'name': None, 'namespace': None, 'owner_references': None, 'resource_version': None, 'self_link': None, 'uid': None}, 'spec': {'active_deadline_seconds': None, 'affinity': None, 'automount_service_account_token': None, 'containers': [{'args': [], 'command': [], 'env': [], 'env_from': [], 'image': None, 'image_pull_policy': None, 'lifecycle': None, 'liveness_probe': None, 'name': 'base', 'ports': [], 'readiness_probe': None, 'resources': {'limits': {'cpu': '3000m', 'memory': '6G'}, 'requests': {'cpu': '1000m', 'memory': '2G'}}, 'security_context': None, 'stdin': None, 'stdin_once': None, 'termination_message_path': None, 'termination_message_policy': None, 'tty': None, 'volume_devices': None, 'volume_mounts': [], 'working_dir': None}], 'dns_config': None, 'dns_policy': None, 'enable_service_links': None, 'host_aliases': None, 'host_ipc': None, 'host_network': False, 'host_pid': None, 'hostname': None, 'image_pull_secrets': [], 'init_containers': None, 'node_name': None, 'node_selector': None, 'preemption_policy': None, 'priority': None, 'priority_class_name': None, 'readiness_gates': None, 'restart_policy': None, 'runtime_class_name': None, 'scheduler_name': None, 'security_context': None, 'service_account': None, 'service_account_name': None, 'share_process_namespace': None, 'subdomain': None, 'termination_grace_period_seconds': None, 'tolerations': None, 'volumes': []}, 'status': None}, None)"
"2022-03-16T00:00:01.934Z","[2022-03-16 00:00:00,968] {kubernetes_executor.py:530} INFO - Add task TaskInstanceKey(dag_id='continuous_load', task_id='load_table', run_id='scheduled__2022-03-15T23:45:00+00:00', try_number=1) with command ['airflow', 'tasks', 'run', 'continuous_load', 'load_table', 'scheduled__2022-03-15T23:45:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/continuous_load.py'] with executor_config {'KubernetesExecutor': {'request_memory': '2G', 'request_cpu': '1000m', 'limit_memory': '6G', 'limit_cpu': '3000m'}}"
"2022-03-16T00:00:01.933Z","[2022-03-16 00:00:00,953] {base_executor.py:82} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'continuous_load', 'load_table', 'scheduled__2022-03-15T23:45:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/continuous_load.py']"
"2022-03-16T00:00:01.933Z","[2022-03-16 00:00:00,952] {scheduler_job.py:473} INFO - Sending TaskInstanceKey(dag_id='continuous_load', task_id='load_table', run_id='scheduled__2022-03-15T23:45:00+00:00', try_number=1) to executor with priority 1 and queue default"

[ephraimbuddy]

@ephraimbuddy Somehow I missed this.

Somehow this issue happen today, again. The tasks are scheduled in pods that belong to a Ec2 spot instance node. The spot node goes down, and the tasks pods are marked as failed, without retrying. If would have retry, then the new task will be run a pod in a node that is up, and this issue won't happen. Here the logs.

Date,Message
"2022-03-16T00:03:26.443Z","[2022-03-16 00:03:26,146] {scheduler_job.py:570} INFO - TaskInstance Finished: dag_id=continuous_load, task_id=load_table, run_id=scheduled__2022-03-15T23:45:00+00:00, run_start_date=2022-03-16 00:01:26.128956+00:00, run_end_date=None, run_duration=None, state=running, executor_state=failed, try_number=1, max_tries=1, job_id=403468, pool=default_pool, queue=default, priority_weight=1, operator=ExtractLoadOperator"
"2022-03-16T00:03:26.442Z","[2022-03-16 00:03:26,069] {kubernetes_executor.py:575} INFO - Changing state of (TaskInstanceKey(dag_id='continuous_load', task_id='load_table', run_id='scheduled__2022-03-15T23:45:00+00:00', try_number=1), , 'load_table.b45adf3ca7ef4f5482639bfedbc4c340', 'airflow-data-eng', '109859634') to failed"
"2022-03-16T00:03:26.442Z","[2022-03-16 00:03:26,064] {kubernetes_executor.py:374} INFO - Attempting to finish pod; pod_id: load_table.b45adf3ca7ef4f5482639bfedbc4c340; state: failed; annotations: {'dag_id': 'continuous_load', 'task_id': 'load_table', 'execution_date': None, 'run_id': 'scheduled__2022-03-15T23:45:00+00:00', 'try_number': '1'}"
"2022-03-16T00:00:21.204Z","[2022-03-16 00:00:20,563] {kubernetes_executor.py:297} INFO - Kubernetes job is (TaskInstanceKey(dag_id='continuous_load', task_id='load_table', run_id='scheduled__2022-03-15T23:45:00+00:00', try_number=1), ['airflow', 'tasks', 'run', 'continuous_load', 'load_table', 'scheduled__2022-03-15T23:45:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/continuous_load.py'], {'api_version': 'v1', 'kind': 'Pod', 'metadata': {'annotations': None, 'cluster_name': None, 'creation_timestamp': None, 'deletion_grace_period_seconds': None, 'deletion_timestamp': None, 'finalizers': None, 'generate_name': None, 'generation': None, 'initializers': None, 'labels': None, 'managed_fields': None, 'name': None, 'namespace': None, 'owner_references': None, 'resource_version': None, 'self_link': None, 'uid': None}, 'spec': {'active_deadline_seconds': None, 'affinity': None, 'automount_service_account_token': None, 'containers': [{'args': [], 'command': [], 'env': [], 'env_from': [], 'image': None, 'image_pull_policy': None, 'lifecycle': None, 'liveness_probe': None, 'name': 'base', 'ports': [], 'readiness_probe': None, 'resources': {'limits': {'cpu': '3000m', 'memory': '6G'}, 'requests': {'cpu': '1000m', 'memory': '2G'}}, 'security_context': None, 'stdin': None, 'stdin_once': None, 'termination_message_path': None, 'termination_message_policy': None, 'tty': None, 'volume_devices': None, 'volume_mounts': [], 'working_dir': None}], 'dns_config': None, 'dns_policy': None, 'enable_service_links': None, 'host_aliases': None, 'host_ipc': None, 'host_network': False, 'host_pid': None, 'hostname': None, 'image_pull_secrets': [], 'init_containers': None, 'node_name': None, 'node_selector': None, 'preemption_policy': None, 'priority': None, 'priority_class_name': None, 'readiness_gates': None, 'restart_policy': None, 'runtime_class_name': None, 'scheduler_name': None, 'security_context': None, 'service_account': None, 'service_account_name': None, 'share_process_namespace': None, 'subdomain': None, 'termination_grace_period_seconds': None, 'tolerations': None, 'volumes': []}, 'status': None}, None)"
"2022-03-16T00:00:01.934Z","[2022-03-16 00:00:00,968] {kubernetes_executor.py:530} INFO - Add task TaskInstanceKey(dag_id='continuous_load', task_id='load_table', run_id='scheduled__2022-03-15T23:45:00+00:00', try_number=1) with command ['airflow', 'tasks', 'run', 'continuous_load', 'load_table', 'scheduled__2022-03-15T23:45:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/continuous_load.py'] with executor_config {'KubernetesExecutor': {'request_memory': '2G', 'request_cpu': '1000m', 'limit_memory': '6G', 'limit_cpu': '3000m'}}"
"2022-03-16T00:00:01.933Z","[2022-03-16 00:00:00,953] {base_executor.py:82} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'continuous_load', 'load_table', 'scheduled__2022-03-15T23:45:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/continuous_load.py']"
"2022-03-16T00:00:01.933Z","[2022-03-16 00:00:00,952] {scheduler_job.py:473} INFO - Sending TaskInstanceKey(dag_id='continuous_load', task_id='load_table', run_id='scheduled__2022-03-15T23:45:00+00:00', try_number=1) to executor with priority 1 and queue default"

Can you also get the task log for this failed task? This looks like an issue we solved long time ago

[nicor88]

The tasks logs are not saying anything relevant. e.g.

[2022-03-16 00:01:38,767] {dagbag.py:500} INFO - Filling up the DagBag from /opt/airflow/dags/continuous_load.py
[2022-03-16 00:01:38,792] {base.py:317} INFO - Statsd buffering is disabled
[2022-03-16 00:01:38,793] {base.py:335} INFO - Statsd periodic buffer flush is disabled

we are running the latest 2.2.4 airflow btw.

I just would like to see airflow retrying after the pod for some reason cannot run.

If for example I do this:

• run the dag
• manaully kill the pod
• airflow retry again the execution

I don't know what is going on in here, but e.g. if the pod is terminated due to some operational reason: e.g. OOM or Ec2 instance down, then retries are not happening.

[TRBaldim]

We are facing the issue with our configuration as well, seems like a bug when we lost a machine by spot instances our task fails without any retry. I will try to gather the logs to share here but I can say that seems the case in our side. We were expecting to the task retry the other two times, but that is not being the case.

[TRBaldim]

Thanks @potiuk, we are running the same version as @nicor88. We will try upgrade the version and will let you all know.

[TRBaldim]

Hello @potiuk as I can found you were right that was a bug associated with our Airflow Version 2.1.x

We have migrated the airflow to 2.3.3 and we were able to find the bug stop happening.

In our peak hours the K8s cluster where scaling up from 3 node to 4-5 nodes. At the end of the process the k8s start to release nodes, but that was at random. The nodes we had lost were marking as failed in the first try. Now is retrying as expected.

[potiuk]

Cool! Thanks for confirming it !

[bharos]

I'm seeing this similar issue, we're on Airflow 2.3.4.
We are using k8s executor with spot ocean instances. Was there a bug that was fixed?
The pods reach TERMINATING state, and the corresponding Airflow tasks are directly marked as FAILED without any retry.

[potiuk]

I suggest you open a new issue detailing your case with logs and everything. Sometims similar issues do not mean the same ones. General advice - we just release 2.7.3 today with a lot of bug fixes (on top of likely 1000s of bugfixes since 2.3.4. So my strong advice is: upgrade, check if the issue still exists, report it then.

[gabrielwry]

Any update on this, we are still facing this issue with airflow 2.8.1

[TRBaldim]

@gabrielwry this issue is not happening anymore on more recent versions.

We are on the 2.7.X