(airflow) branch main updated: K8s Executor: failing the task in case the watcher receives an event with the reason ProviderFailed (#41186)

[je...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jedcunningham pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 09ce3a5100 K8s Executor: failing the task in case the watcher receives an event with the reason ProviderFailed (#41186)
09ce3a5100 is described below

commit 09ce3a5100c266369350c85e9f9a0f72ecca9e98
Author: Gopal Dirisala <39...@users.noreply.github.com>
AuthorDate: Fri Aug 23 11:44:02 2024 +0530

    K8s Executor: failing the task in case the watcher receives an event with the reason ProviderFailed (#41186)
---
 .../cncf/kubernetes/executors/kubernetes_executor_utils.py     | 10 ++++++++++
 .../cncf/kubernetes/executors/test_kubernetes_executor.py      |  7 +++++++
 2 files changed, 17 insertions(+)

diff --git a/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py b/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py
index fce631b9b7..a6e49adb93 100644
--- a/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py
+++ b/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py
@@ -220,6 +220,16 @@ class KubernetesJobWatcher(multiprocessing.Process, LoggingMixin):
             # However, need to free the executor slot from the current executor.
             self.log.info("Event: pod %s adopted, annotations: %s", pod_name, annotations_string)
             self.watcher_queue.put((pod_name, namespace, ADOPTED, annotations, resource_version))
+        elif hasattr(pod.status, "reason") and pod.status.reason == "ProviderFailed":
+            # Most likely this happens due to Kubernetes setup (virtual kubelet, virtual nodes, etc.)
+            self.log.error(
+                "Event: %s failed to start with reason ProviderFailed, annotations: %s",
+                pod_name,
+                annotations_string,
+            )
+            self.watcher_queue.put(
+                (pod_name, namespace, TaskInstanceState.FAILED, annotations, resource_version)
+            )
         elif status == "Pending":
             # deletion_timestamp is set by kube server when a graceful deletion is requested.
             # since kube server have received request to delete pod set TI state failed
diff --git a/tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py b/tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py
index e062d77ef6..769eb9c980 100644
--- a/tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py
+++ b/tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py
@@ -1804,6 +1804,13 @@ class TestKubernetesJobWatcher:
         self._run()
         self.assert_watcher_queue_called_once_with_state(State.FAILED)
 
+    def test_process_status_provider_failed(self):
+        self.pod.status.reason = "ProviderFailed"
+        self.events.append({"type": "MODIFIED", "object": self.pod})
+
+        self._run()
+        self.assert_watcher_queue_called_once_with_state(State.FAILED)
+
     def test_process_status_succeeded(self):
         self.pod.status.phase = "Succeeded"
         self.events.append({"type": "MODIFIED", "object": self.pod})