This is an automated email from the ASF dual-hosted git repository.
dstandish 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 6bbe015905 Use "remote" pod when patching KPO pod as "checked" (#23676)
6bbe015905 is described below
commit 6bbe015905bd2709e621455d9f71a78b374d1337
Author: Daniel Standish <[email protected]>
AuthorDate: Thu May 26 07:40:19 2022 -0700
Use "remote" pod when patching KPO pod as "checked" (#23676)
When patching as "checked", we have to use the current version of the pod
otherwise we may get an error when trying to patch it, e.g.:
```
Operation cannot be fulfilled on pods
\"test-kubernetes-pod-db9eedb7885c40099dd40cd4edc62415\": the object has been
modified; please apply your changes to the latest version and try again"
```
This error would not cause a failure of the task, since errors in `cleanup`
are suppressed. However, it would fail to patch.
I believe one scenario when the pod may be updated is when retrieving xcom,
since the sidecar is terminated after extracting the value.
Concerning some changes in the tests re the "already_checked" label, it was
added to a few "expected pods" recently, when we changed it to patch even in
the case of a successful pod.
Since we are changing the "patch" code to patch with the latest read on the
pod that we have (i.e. using the `remote_pod` variable), and no longer the pod
object stored on `k.pod`, the label no longer shows up in those tests (that's
because in k.pod isn't actually a read of the remote pod, but just happens to
get mutated in the patch function before it is used to actually patch the pod).
Further, since the `remote_pod` is a local variable, we can't observe it in
tests. So we have to read the pod using k8s api. _But_, our "find pod"
function excludes "already checked" pods! So we have to make this configurable.
So, now we have a proper integration test for the "already_checked"
behavior (there was already a unit test).
---
.../cncf/kubernetes/operators/kubernetes_pod.py | 14 ++++--
kubernetes_tests/test_kubernetes_pod_operator.py | 55 ++++++++++++++++++++--
.../test_kubernetes_pod_operator_backcompat.py | 1 -
3 files changed, 61 insertions(+), 9 deletions(-)
diff --git a/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
b/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
index ee55d34ae1..6e46fdbb16 100644
--- a/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
+++ b/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py
@@ -330,9 +330,9 @@ class KubernetesPodOperator(BaseOperator):
kwargs.update(in_cluster=self.in_cluster)
return kube_client.get_kube_client(**kwargs)
- def find_pod(self, namespace, context) -> Optional[k8s.V1Pod]:
+ def find_pod(self, namespace, context, *, exclude_checked=True) ->
Optional[k8s.V1Pod]:
"""Returns an already-running pod for this task instance if one
exists."""
- label_selector = self._build_find_pod_label_selector(context)
+ label_selector = self._build_find_pod_label_selector(context,
exclude_checked=exclude_checked)
pod_list = self.client.list_namespaced_pod(
namespace=namespace,
label_selector=label_selector,
@@ -412,7 +412,7 @@ class KubernetesPodOperator(BaseOperator):
pod_phase = remote_pod.status.phase if hasattr(remote_pod, 'status')
else None
if not self.is_delete_operator_pod:
with _suppress(Exception):
- self.patch_already_checked(pod)
+ self.patch_already_checked(remote_pod)
if pod_phase != PodPhase.SUCCEEDED:
if self.log_events_on_failure:
with _suppress(Exception):
@@ -436,10 +436,14 @@ class KubernetesPodOperator(BaseOperator):
else:
self.log.info("skipping deleting pod: %s", pod.metadata.name)
- def _build_find_pod_label_selector(self, context: Optional[dict] = None)
-> str:
+ def _build_find_pod_label_selector(self, context: Optional[dict] = None,
*, exclude_checked=True) -> str:
labels = self._get_ti_pod_labels(context, include_try_number=False)
label_strings = [f'{label_id}={label}' for label_id, label in
sorted(labels.items())]
- return ','.join(label_strings) +
f',{self.POD_CHECKED_KEY}!=True,!airflow-worker'
+ labels_value = ','.join(label_strings)
+ if exclude_checked:
+ labels_value += f',{self.POD_CHECKED_KEY}!=True'
+ labels_value += ',!airflow-worker'
+ return labels_value
def _set_name(self, name):
if name is None:
diff --git a/kubernetes_tests/test_kubernetes_pod_operator.py
b/kubernetes_tests/test_kubernetes_pod_operator.py
index 25c5bab763..56705c7a9d 100644
--- a/kubernetes_tests/test_kubernetes_pod_operator.py
+++ b/kubernetes_tests/test_kubernetes_pod_operator.py
@@ -169,9 +169,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
context = create_context(k)
k.execute(context)
expected_pod = copy(self.expected_pod)
- expected_pod['metadata']['labels']['already_checked'] = 'True'
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
- assert expected_pod == actual_pod
+ assert actual_pod == expected_pod
def test_working_pod(self):
k = KubernetesPodOperator(
@@ -210,6 +209,57 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
assert self.expected_pod['spec'] == actual_pod['spec']
assert self.expected_pod['metadata']['labels'] ==
actual_pod['metadata']['labels']
+ def test_already_checked_on_success(self):
+ """
+ When ``is_delete_operator_pod=False``, pod should have
'already_checked'
+ label, whether pod is successful or not.
+ """
+ pod_name = "test-" + str(random.randint(0, 1000000))
+ k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["echo 10"],
+ labels={"foo": "bar"},
+ name=pod_name,
+ task_id="task" + self.get_current_task_name(),
+ in_cluster=False,
+ do_xcom_push=False,
+ is_delete_operator_pod=False,
+ )
+ context = create_context(k)
+ k.execute(context)
+ actual_pod = k.find_pod('default', context, exclude_checked=False)
+ actual_pod = self.api_client.sanitize_for_serialization(actual_pod)
+ assert actual_pod['metadata']['labels']['already_checked'] == 'True'
+
+ def test_already_checked_on_failure(self):
+ """
+ When ``is_delete_operator_pod=False``, pod should have
'already_checked'
+ label, whether pod is successful or not.
+ """
+ pod_name = "test-" + str(random.randint(0, 1000000))
+ k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["lalala"],
+ labels={"foo": "bar"},
+ name=pod_name,
+ task_id="task" + self.get_current_task_name(),
+ in_cluster=False,
+ do_xcom_push=False,
+ is_delete_operator_pod=False,
+ )
+ context = create_context(k)
+ with pytest.raises(AirflowException):
+ k.execute(context)
+ actual_pod = k.find_pod('default', context, exclude_checked=False)
+ actual_pod = self.api_client.sanitize_for_serialization(actual_pod)
+ status = next(iter(filter(lambda x: x['name'] == 'base',
actual_pod['status']['containerStatuses'])))
+ assert status['state']['terminated']['reason'] == 'Error'
+ assert actual_pod['metadata']['labels']['already_checked'] == 'True'
+
def test_pod_hostnetwork(self):
k = KubernetesPodOperator(
namespace='default',
@@ -763,7 +813,6 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
'kubernetes_pod_operator': 'True',
'task_id': mock.ANY,
'try_number': '1',
- 'already_checked': 'True',
}
assert k.pod.spec.containers[0].env == [k8s.V1EnvVar(name="env_name",
value="value")]
assert result == {"hello": "world"}
diff --git a/kubernetes_tests/test_kubernetes_pod_operator_backcompat.py
b/kubernetes_tests/test_kubernetes_pod_operator_backcompat.py
index 430387118a..21a3bda7d3 100644
--- a/kubernetes_tests/test_kubernetes_pod_operator_backcompat.py
+++ b/kubernetes_tests/test_kubernetes_pod_operator_backcompat.py
@@ -310,7 +310,6 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
expected_pod['spec']['volumes'] = [
{'name': 'test-volume', 'persistentVolumeClaim': {'claimName':
'test-volume'}}
]
- expected_pod['metadata']['labels']['already_checked'] = 'True'
assert expected_pod == actual_pod
def test_run_as_user_root(self):
