itz-puneet commented on code in PR #72606:
URL: https://github.com/apache/airflow/pull/72606#discussion_r3945645816
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -886,6 +886,17 @@ def execute_sync(self, context: Context):
pod=pod_to_clean, remote_pod=self.remote_pod, context=context,
result=result
)
+ if self._killed:
+ # on_kill() ran while the block above was waiting on the pod, and
that wait
+ # returned normally because the pod it was watching simply went
away. The
+ # workload never finished, so falling through here would finalise
the task
+ # instance as success. This check sits after the finally block on
purpose: if
+ # the body raised, that exception propagates untouched and already
fails the
+ # task with its own reason.
+ raise AirflowException(
+ f"Pod {self.pod and self.pod.metadata.name} was interrupted
before it completed."
+ )
Review Comment:
Good catch — the hook is real and both files were over their cap (pod.py
allows 3 and had 4; the test module allows none). Added a dedicated
`PodInterrupted(AirflowException)` alongside the other pod exceptions in
aa59bcf, and the message now falls back to `self.name` so it cannot read "Pod
None".
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##########
@@ -823,6 +823,89 @@ def test_process_pod_deletion(self, delete_pod_mock,
on_finish_action, pod_phase
assert result == should_delete
+ def _interrupted_pod_operator(self, **kwargs):
+ return KubernetesPodOperator(
+ namespace="default",
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["sleep 120"],
+ name="sleep-worker",
+ task_id="task",
+ do_xcom_push=False,
+ get_logs=True,
+ **kwargs,
+ )
+
+ @staticmethod
+ def _running_pod():
+ pod = MagicMock()
+ pod.metadata.name = "sleep-worker"
+ pod.metadata.namespace = "default"
+ pod.status.phase = PodPhase.RUNNING
+ return pod
+
+ @patch(HOOK_CLASS, new=MagicMock)
+ @patch(KUB_OP_PATH.format("get_or_create_pod"))
+ @patch(KUB_OP_PATH.format("find_pod"))
+ @patch(KUB_OP_PATH.format("await_pod_completion"))
+ def test_execute_sync_fails_when_on_kill_ran_during_the_wait(
+ self, await_pod_completion_mock, find_pod_mock, get_or_create_pod_mock
+ ):
+ """A pod interrupted by on_kill must not finalise the task instance as
success.
+
+ Under KubernetesExecutor the task pod and the KPO child pod can be
interrupted
+ within about a second of each other. SIGTERM reaches the task process,
the runner
+ calls on_kill(), which deletes the child, and the wait in execute_sync
then returns
+ normally because the pod it was watching has gone away. cleanup()
skips its usual
+ failure signalling once _killed is set, so execute_sync used to fall
through and
+ return, and the task was recorded as success (apache/airflow#71202).
Review Comment:
Removed the issue reference from the docstring in aa59bcf — it lives in the
commit message and PR description instead.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##########
@@ -823,6 +823,89 @@ def test_process_pod_deletion(self, delete_pod_mock,
on_finish_action, pod_phase
assert result == should_delete
+ def _interrupted_pod_operator(self, **kwargs):
+ return KubernetesPodOperator(
+ namespace="default",
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["sleep 120"],
+ name="sleep-worker",
+ task_id="task",
+ do_xcom_push=False,
+ get_logs=True,
+ **kwargs,
+ )
+
+ @staticmethod
+ def _running_pod():
+ pod = MagicMock()
+ pod.metadata.name = "sleep-worker"
+ pod.metadata.namespace = "default"
+ pod.status.phase = PodPhase.RUNNING
+ return pod
+
+ @patch(HOOK_CLASS, new=MagicMock)
+ @patch(KUB_OP_PATH.format("get_or_create_pod"))
+ @patch(KUB_OP_PATH.format("find_pod"))
+ @patch(KUB_OP_PATH.format("await_pod_completion"))
+ def test_execute_sync_fails_when_on_kill_ran_during_the_wait(
+ self, await_pod_completion_mock, find_pod_mock, get_or_create_pod_mock
+ ):
+ """A pod interrupted by on_kill must not finalise the task instance as
success.
+
+ Under KubernetesExecutor the task pod and the KPO child pod can be
interrupted
+ within about a second of each other. SIGTERM reaches the task process,
the runner
+ calls on_kill(), which deletes the child, and the wait in execute_sync
then returns
+ normally because the pod it was watching has gone away. cleanup()
skips its usual
+ failure signalling once _killed is set, so execute_sync used to fall
through and
+ return, and the task was recorded as success (apache/airflow#71202).
+ """
+ k = self._interrupted_pod_operator()
+ running_pod = self._running_pod()
+ get_or_create_pod_mock.return_value = running_pod
+ find_pod_mock.return_value = running_pod
+ self.await_pod_mock.return_value = running_pod
+
+ # The wait returns rather than raising: the log stream simply ended
when the
+ # child pod was deleted out from under it.
+ await_pod_completion_mock.side_effect = lambda pod: k.on_kill()
+
+ context = create_context(k)
+ context["ti"].xcom_push = MagicMock()
+
+ with pytest.raises(AirflowException, match="was interrupted before it
completed"):
Review Comment:
Updated in aa59bcf: the interrupted test now expects `PodInterrupted` by
type, and the "body already failed" test raises `RuntimeError`, which also
keeps that file clear of the AirflowException cap and makes the stronger point
that any in-flight exception survives.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##########
@@ -823,6 +823,89 @@ def test_process_pod_deletion(self, delete_pod_mock,
on_finish_action, pod_phase
assert result == should_delete
+ def _interrupted_pod_operator(self, **kwargs):
+ return KubernetesPodOperator(
+ namespace="default",
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["sleep 120"],
+ name="sleep-worker",
+ task_id="task",
+ do_xcom_push=False,
+ get_logs=True,
+ **kwargs,
+ )
+
+ @staticmethod
+ def _running_pod():
+ pod = MagicMock()
+ pod.metadata.name = "sleep-worker"
+ pod.metadata.namespace = "default"
+ pod.status.phase = PodPhase.RUNNING
+ return pod
Review Comment:
Leaving this one as is: the surrounding module builds pod mocks with bare
`MagicMock()` throughout, and `spec_set` on `V1Pod` plus its nested
`status`/`metadata` objects would make these two tests inconsistent with the
rest of the file. Happy to switch if a maintainer would rather standardise on
it.
--
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]
