You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by di...@apache.org on 2021/01/21 20:57:46 UTC
[airflow] branch master updated: Fix error with quick-failing tasks
in KubernetesPodOperator (#13621)
This is an automated email from the ASF dual-hosted git repository.
dimberman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/master by this push:
new 94d3ed6 Fix error with quick-failing tasks in KubernetesPodOperator (#13621)
94d3ed6 is described below
commit 94d3ed61d60b134d649a4e9785b2d9c2a88cff05
Author: Daniel Imberman <da...@gmail.com>
AuthorDate: Thu Jan 21 12:57:35 2021 -0800
Fix error with quick-failing tasks in KubernetesPodOperator (#13621)
* Fix error with quick-failing tasks in KubernetesPodOperator
Addresses an issue with the KubernetesPodOperator where tasks that die
quickly are not patched with "already_checked" because they never make
it to the monitoring logic.
* static fix
---
.../cncf/kubernetes/operators/kubernetes_pod.py | 1 +
kubernetes_tests/test_kubernetes_pod_operator.py | 25 ++++++++++++++++++++++
2 files changed, 26 insertions(+)
diff --git a/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py b/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
index 3f42ab1..996b7da 100644
--- a/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
+++ b/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
@@ -343,6 +343,7 @@ class KubernetesPodOperator(BaseOperator): # pylint: disable=too-many-instance-
raise AirflowException(f'Pod {self.pod.metadata.name} returned a failure: {status}')
return result
except AirflowException as ex:
+ self.patch_already_checked(self.pod)
raise AirflowException(f'Pod Launching failed: {ex}')
def handle_pod_overlap(
diff --git a/kubernetes_tests/test_kubernetes_pod_operator.py b/kubernetes_tests/test_kubernetes_pod_operator.py
index 574f7ff..e860303 100644
--- a/kubernetes_tests/test_kubernetes_pod_operator.py
+++ b/kubernetes_tests/test_kubernetes_pod_operator.py
@@ -1000,5 +1000,30 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
k.execute(context)
create_mock.assert_called_once()
+ def test_reatttach_quick_failure(self):
+ client = kube_client.get_kube_client(in_cluster=False)
+ namespace = "default"
+
+ name = "test"
+ k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["exit 1"],
+ labels={"foo": "bar"},
+ name="test",
+ task_id=name,
+ in_cluster=False,
+ do_xcom_push=False,
+ is_delete_operator_pod=False,
+ termination_grace_period=0,
+ )
+
+ context = create_context(k)
+ with self.assertRaises(AirflowException):
+ k.execute(context)
+ pod = client.read_namespaced_pod(name=k.pod.metadata.name, namespace=namespace)
+ self.assertEqual(pod.metadata.labels["already_checked"], "True")
+
# pylint: enable=unused-argument