You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "jykae (via GitHub)" <gi...@apache.org> on 2026/05/22 13:06:47 UTC
[PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
jykae opened a new pull request, #67333:
URL: https://github.com/apache/airflow/pull/67333
Fix monitoring-pod leak in ``KubernetesJobOperator``.
``KubernetesJobOperator`` inherits from ``KubernetesPodOperator`` but
overrode ``execute()`` without ever invoking the parent's pod-cleanup path,
so the "monitoring" pods discovered via ``get_pods()`` (used to stream
logs and XCom while the Job runs) were never deleted. These pods are
created by Airflow, not by the ``V1Job`` controller, so they have no
``ownerReferences`` — neither ``ttl_seconds_after_finished`` nor the
foreground cascade on ``on_kill()`` reaped them. Every task run leaked
one pod per Job.
This PR makes pod cleanup symmetric with ``KubernetesPodOperator`` and
honours ``on_finish_action`` / ``on_kill_action`` for the discovered pods.
### Changes
**``operators/job.py``**
* ``execute()`` and ``execute_complete()`` now wrap their work in
``try/finally`` and call ``post_complete_action()`` for every pod
returned by ``get_pods()``. The inherited ``on_finish_action``
(``delete_pod`` / ``delete_succeeded_pod`` / ``delete_active_pod`` /
``keep_pod``) is now respected, matching ``KubernetesPodOperator``
semantics.
* ``on_kill()`` additionally calls ``pod_manager.delete_pod()`` for each
monitoring pod, gated by ``on_kill_action``. The Job's foreground
cascade does not reach these pods because they have no
``ownerReferences``. Unexpected ``ApiException``s are logged instead
of silently suppressed.
* ``execute_complete()`` resolves monitoring pods once and shares the
lookup between the log-retrieval and cleanup paths. Resolution is
best-effort — failures in the deferrable resume path no longer break
cleanup.
* Per-pod cleanup errors are logged but never mask the in-flight
exception, so Job-level failures continue to propagate unchanged.
**``triggers/job.py``**
* The trigger event now always includes ``pod_names`` /
``pod_namespace``, regardless of ``get_logs``. This guarantees
``execute_complete()`` can reliably clean up monitoring pods even
when log streaming is disabled.
**``docs/operators.rst``**
New section documenting the cleanup contract: which pods are affected,
the meaning of each ``on_finish_action`` value for monitoring pods, and
the ``on_kill_action`` behaviour.
**Tests**
* Coverage for each ``on_finish_action`` value (``delete_pod``,
``delete_succeeded_pod``, ``delete_active_pod``, ``keep_pod``) on
both success and failure paths.
* Coverage for ``on_kill_action`` (``delete_pod`` / ``keep_pod``).
* Regression test for the deferrable ``get_logs=False`` path.
* New mocks use ``spec`` / ``autospec`` to catch attribute typos
against the real ``kubernetes`` client surface.
### Backwards compatibility
Default ``on_finish_action`` is unchanged (``delete_pod``), so existing
deployments will start reclaiming the leaked monitoring pods
automatically. Users who relied on monitoring pods surviving the task
(e.g. for offline log inspection) can opt in explicitly by passing
``on_finish_action="keep_pod"``.
### How to verify
1. Run a DAG using ``KubernetesJobOperator`` with default settings.
2. After the task finishes, both the ``V1Job``'s child pod and the
monitoring pod (label ``airflow_kpo_in_cluster=True``, no
``ownerReferences``) should be gone.
3. Repeat with ``on_finish_action="delete_succeeded_pod"`` and a
failing command — the monitoring pod should remain for forensics.
4. Repeat with ``on_finish_action="keep_pod"`` — both pods should
remain.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — GitHub Copilot (Claude Opus 4.7)
Generated-by: GitHub Copilot (Claude Opus 4.7) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on PR #67333:
URL: https://github.com/apache/airflow/pull/67333#issuecomment-4561646605
@jscheffl please ping me if I still need to do something related to check failures
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3308941325
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_job.py:
##########
@@ -120,6 +120,39 @@ async def test_run_success(self, mock_hook, trigger):
}
)
+ @pytest.mark.asyncio
+ @mock.patch(f"{TRIGGER_CLASS}.hook")
+ async def test_run_success_emits_pod_info_when_get_logs_false(self, mock_hook):
+ """pod_names/pod_namespace must be in the event even with get_logs=False.
+
+ The operator's execute_complete needs them to clean up monitoring pods.
+ """
+ trigger_no_logs = KubernetesJobTrigger(
+ job_name=JOB_NAME,
+ job_namespace=NAMESPACE,
+ pod_names=[POD_NAME],
+ pod_namespace=NAMESPACE,
+ base_container_name=CONTAINER_NAME,
+ kubernetes_conn_id=CONN_ID,
+ poll_interval=POLL_INTERVAL,
+ cluster_context=CLUSTER_CONTEXT,
+ config_file=CONFIG_FILE,
+ in_cluster=IN_CLUSTER,
+ get_logs=False,
+ do_xcom_push=XCOM_PUSH,
+ )
+ mock_job = mock.MagicMock()
+ mock_job.metadata.name = JOB_NAME
+ mock_job.metadata.namespace = NAMESPACE
Review Comment:
fixed https://github.com/apache/airflow/pull/67333/commits/d42fd91c5b16ac1d40d530b893576e5b7a11b40a
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on PR #67333:
URL: https://github.com/apache/airflow/pull/67333#issuecomment-4543184877
FYI we have been running this successfully from last weekend, no succeeded pods were left on our test cluster namespace where Airflow creates pods using KubernetesJobOperator. We have approx. 243 active DAGs running on our Airflow instances. On production we currently need manual cleanup, and on this scale this has been leaving hundreds of pods on completed stage.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #67333:
URL: https://github.com/apache/airflow/pull/67333#issuecomment-4529375473
@jykae A few things need addressing before review — see our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
- :x: **Newsfragment check fails**: `check-newsfragment-pr-number` — see [`05_pull_requests.rst#change-summary`](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst). For a provider fix, update the provider's changelog directly: `providers/cncf/kubernetes/docs/changelog.rst` (no newsfragment needed for providers).
- :x: **Static checks fail**: `CI image checks / Static checks` — lint/type issue. See the [Checks tab](https://github.com/apache/airflow/pull/67333/checks).
No rush.
---
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
---
Drafted-by: Claude Code (Opus 4.7); reviewed by @potiuk before posting
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3320890126
##########
chart/values.yaml:
##########
Review Comment:
CHart is maintained in a separate maintenance branch for this to back-port please separate-out the changes in provider and in chart. They are also released with different release cadence. And for the CHart we need to discuss whether this actually should be accepted as we have in general the aim to slim-down complexity. It might be that we feedback to put this into a Kustomize overlay.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3308940152
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_job.py:
##########
@@ -105,6 +106,25 @@ def create_context(task, persist_to_db=False, map_index=None):
}
+def _recording_pod_manager():
+ """A ``pod_manager`` mock that records ``delete_pod`` invocations by pod name.
+
+ Tests should assert against ``pm.deleted`` to observe which monitoring pods
+ were actually deleted — the only thing that matters to the cluster.
+ """
+ pm = mock.MagicMock()
+ pm.deleted = []
+ pm.delete_pod.side_effect = lambda pod: pm.deleted.append(pod.metadata.name)
Review Comment:
fixed https://github.com/apache/airflow/pull/67333/commits/d42fd91c5b16ac1d40d530b893576e5b7a11b40a
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3322221360
##########
chart/values.yaml:
##########
Review Comment:
@jscheffl ok, I take chart changes away and we maintain rbac ourselves.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3308935375
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_job.py:
##########
@@ -1193,6 +1278,279 @@ def test_create_zero_parallelism_fails_validation(
mock_hook.return_value.create_job.assert_not_called()
mock_get_pods.assert_not_called()
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleans_up_all_pods_on_success(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """Every monitoring pod is deleted after a successful job, including under ``parallelism>1``.
+
+ Real ``post_complete_action`` runs; only the K8s SDK boundary is faked.
+ """
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.return_value.is_job_failed.return_value = False
+ pod_1 = _pod("pod-1")
+ pod_2 = _pod("pod-2")
+ mock_get_pods.return_value = [pod_1, pod_2]
+ mock_find_pod.side_effect = [pod_1, pod_2]
+ mock_hook.return_value.get_pod.side_effect = lambda name, namespace: (
+ pod_1 if name == "pod-1" else pod_2
+ )
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True, parallelism=2)
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ assert sorted(pm.deleted) == ["pod-1", "pod-2"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleans_up_pod_on_failure(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """When the job fails, the monitoring pod is still deleted and the error propagates."""
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.return_value.is_job_failed.return_value = "Error"
+ pod = _pod("pod-1", phase="Failed")
+ mock_get_pods.return_value = [pod]
+ mock_find_pod.return_value = pod
+ mock_hook.return_value.get_pod.return_value = pod
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True)
+ with pytest.raises(AirflowException, match="is failed with error"):
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ assert pm.deleted == ["pod-1"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_keep_pod_skips_cleanup(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_pod_manager_prop,
+ ):
+ """``on_finish_action=keep_pod`` leaves the monitoring pod untouched."""
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.return_value.is_job_failed.return_value = False
+ pod = _pod("pod-1")
+ mock_get_pods.return_value = [pod]
+ mock_hook.return_value.get_pod.return_value = pod
+
+ op = KubernetesJobOperator(
+ task_id="test_task_id", wait_until_job_complete=True, on_finish_action="keep_pod"
+ )
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ assert pm.deleted == []
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleanup_error_on_success_does_not_raise(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """A delete failure on one pod must not stop cleanup of the others nor fail the task.
+
+ Exercises the real ``post_complete_action`` against a recording pod_manager
+ whose first ``delete_pod`` raises at the K8s SDK boundary.
+ """
+ mock_hook.return_value.is_job_failed.return_value = False
+ pod_1 = _pod("pod-1")
+ pod_2 = _pod("pod-2")
+ mock_get_pods.return_value = [pod_1, pod_2]
+ mock_find_pod.side_effect = [pod_1, pod_2]
+ mock_hook.return_value.get_pod.side_effect = lambda name, namespace: (
+ pod_1 if name == "pod-1" else pod_2
+ )
+ pm = _recording_pod_manager()
+ original_side_effect = pm.delete_pod.side_effect
+
+ def delete_side_effect(pod):
+ if pod.metadata.name == "pod-1":
+ raise ApiException(status=500, reason="boom")
+ original_side_effect(pod)
+
+ pm.delete_pod.side_effect = delete_side_effect
+ mock_pod_manager_prop.return_value = pm
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True, parallelism=2)
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ # pod-1's delete blew up; pod-2 still got deleted.
+ assert pm.deleted == ["pod-2"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleanup_error_does_not_mask_job_failure(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """When the job already failed, a cleanup error must not replace the job-level error.
+
+ The recording side_effect lets us observe that cleanup was *attempted*
+ even when it raises — the contract is "attempt cleanup, swallow any
+ error, preserve the original job failure".
+ """
+ mock_hook.return_value.is_job_failed.return_value = "Error"
+ pod = _pod("pod-1", phase="Failed")
+ mock_get_pods.return_value = [pod]
+ mock_find_pod.return_value = pod
+ mock_hook.return_value.get_pod.return_value = pod
+ attempted: list[str] = []
+
+ def boom(pod):
+ attempted.append(pod.metadata.name)
+ raise ApiException(status=500, reason="boom")
+
+ pm = mock.MagicMock()
+ pm.delete_pod.side_effect = boom
+ mock_pod_manager_prop.return_value = pm
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True)
+ # The job-level error wins; the cleanup ApiException is swallowed.
+ with pytest.raises(AirflowException, match="is failed with error"):
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ # Cleanup was still attempted even though it raised.
+ assert attempted == ["pod-1"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.execute_deferrable"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_does_not_cleanup_when_deferring(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_execute_deferrable,
+ mock_pod_manager_prop,
+ ):
+ """A ``TaskDeferred`` on the way out of ``execute()`` must leave pods alive.
+
+ The trigger still needs to watch them; cleanup happens on resume in
+ ``execute_complete``.
+ """
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ pod = _pod("pod-1")
+ mock_get_pods.return_value = [pod]
+ mock_execute_deferrable.side_effect = TaskDeferred(
+ trigger=mock.MagicMock(), method_name="execute_complete"
+ )
Review Comment:
fixed
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3292849924
##########
providers/cncf/kubernetes/newsfragments/kubernetes-job-operator-cleanup-monitoring-pods.bugfix.rst:
##########
Review Comment:
@copilot please remove newsfragments from this PR
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3292849924
##########
providers/cncf/kubernetes/newsfragments/kubernetes-job-operator-cleanup-monitoring-pods.bugfix.rst:
##########
Review Comment:
@copilot please remove newsfragments from this PR
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3298719600
##########
providers/cncf/kubernetes/docs/changelog.rst:
##########
Review Comment:
Please do no add items in a PR to changelog, this is done by release manager once release is cut.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3292590738
##########
providers/cncf/kubernetes/newsfragments/kubernetes-job-operator-cleanup-monitoring-pods.bugfix.rst:
##########
Review Comment:
We do not use newsfragments on providers. Please remove.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #67333:
URL: https://github.com/apache/airflow/pull/67333#issuecomment-4574530603
Awesome work, congrats on your first merged pull request! You are invited to check our [Issue Tracker](https://github.com/apache/airflow/issues) for additional contributions.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on PR #67333:
URL: https://github.com/apache/airflow/pull/67333#issuecomment-4574599942
@jscheffl thanks for guidance in my first ever PR for this project 👏 🙇♂️
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3301453302
##########
providers/cncf/kubernetes/docs/changelog.rst:
##########
Review Comment:
@jscheffl ok, good to know, removed changelog modifications https://github.com/apache/airflow/pull/67333/commits/254b234c56ebf6fe9be93fa0d90ca7fe9841f820
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3306561108
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_job.py:
##########
@@ -105,6 +106,25 @@ def create_context(task, persist_to_db=False, map_index=None):
}
+def _recording_pod_manager():
+ """A ``pod_manager`` mock that records ``delete_pod`` invocations by pod name.
+
+ Tests should assert against ``pm.deleted`` to observe which monitoring pods
+ were actually deleted — the only thing that matters to the cluster.
+ """
+ pm = mock.MagicMock()
+ pm.deleted = []
+ pm.delete_pod.side_effect = lambda pod: pm.deleted.append(pod.metadata.name)
Review Comment:
This new test helper uses an unspecced `MagicMock`, which can accept any attribute and hide typos in the mocked `pod_manager` API. Please give it a spec/autospec (for example against `PodManager`, or a small explicit fake object) before attaching the recording side effect.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_job.py:
##########
@@ -120,6 +120,39 @@ async def test_run_success(self, mock_hook, trigger):
}
)
+ @pytest.mark.asyncio
+ @mock.patch(f"{TRIGGER_CLASS}.hook")
+ async def test_run_success_emits_pod_info_when_get_logs_false(self, mock_hook):
+ """pod_names/pod_namespace must be in the event even with get_logs=False.
+
+ The operator's execute_complete needs them to clean up monitoring pods.
+ """
+ trigger_no_logs = KubernetesJobTrigger(
+ job_name=JOB_NAME,
+ job_namespace=NAMESPACE,
+ pod_names=[POD_NAME],
+ pod_namespace=NAMESPACE,
+ base_container_name=CONTAINER_NAME,
+ kubernetes_conn_id=CONN_ID,
+ poll_interval=POLL_INTERVAL,
+ cluster_context=CLUSTER_CONTEXT,
+ config_file=CONFIG_FILE,
+ in_cluster=IN_CLUSTER,
+ get_logs=False,
+ do_xcom_push=XCOM_PUSH,
+ )
+ mock_job = mock.MagicMock()
+ mock_job.metadata.name = JOB_NAME
+ mock_job.metadata.namespace = NAMESPACE
Review Comment:
This test introduces an unspecced `MagicMock` for a Kubernetes job object. Using an autospecced `V1Job` (or a real minimal `V1Job`) would catch attribute mistakes in the trigger payload setup.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_job.py:
##########
@@ -883,29 +903,73 @@ def test_execute_complete_pod_not_found_skips_logs(self, mock_hook, mocked_write
mocked_write_logs.assert_not_called()
@pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
@patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator._write_logs"))
@patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.hook"))
- def test_execute_complete_pod_api_error_reraises(self, mock_hook, mocked_write_logs):
- """Non-404 ApiExceptions should still be raised."""
- from kubernetes.client.rest import ApiException
+ def test_execute_complete_pod_api_error_does_not_raise(
+ self, mock_hook, mocked_write_logs, mock_pod_manager_prop
+ ):
+ """Non-404 errors fetching the monitoring pod must not fail ``execute_complete``.
- mock_ti = mock.MagicMock()
- context = {"ti": mock_ti}
- mock_job = mock.MagicMock()
+ Cleanup is best-effort: log writing is skipped, but the pod is still
+ targeted for deletion by name so it doesn't outlive the job.
+ """
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.get_pod.side_effect = ApiException(status=403, reason="Forbidden")
event = {
- "job": mock_job,
+ "job": mock.MagicMock(),
"status": "success",
Review Comment:
This event fixture uses an unspecced `MagicMock`, so it will silently accept misspelled attribute access if the production code changes to inspect the job payload. Please use a minimal dict matching the trigger event shape or a specced object instead.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_job.py:
##########
@@ -1193,6 +1278,279 @@ def test_create_zero_parallelism_fails_validation(
mock_hook.return_value.create_job.assert_not_called()
mock_get_pods.assert_not_called()
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleans_up_all_pods_on_success(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """Every monitoring pod is deleted after a successful job, including under ``parallelism>1``.
+
+ Real ``post_complete_action`` runs; only the K8s SDK boundary is faked.
+ """
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.return_value.is_job_failed.return_value = False
+ pod_1 = _pod("pod-1")
+ pod_2 = _pod("pod-2")
+ mock_get_pods.return_value = [pod_1, pod_2]
+ mock_find_pod.side_effect = [pod_1, pod_2]
+ mock_hook.return_value.get_pod.side_effect = lambda name, namespace: (
+ pod_1 if name == "pod-1" else pod_2
+ )
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True, parallelism=2)
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ assert sorted(pm.deleted) == ["pod-1", "pod-2"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleans_up_pod_on_failure(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """When the job fails, the monitoring pod is still deleted and the error propagates."""
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.return_value.is_job_failed.return_value = "Error"
+ pod = _pod("pod-1", phase="Failed")
+ mock_get_pods.return_value = [pod]
+ mock_find_pod.return_value = pod
+ mock_hook.return_value.get_pod.return_value = pod
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True)
+ with pytest.raises(AirflowException, match="is failed with error"):
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ assert pm.deleted == ["pod-1"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_keep_pod_skips_cleanup(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_pod_manager_prop,
+ ):
+ """``on_finish_action=keep_pod`` leaves the monitoring pod untouched."""
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.return_value.is_job_failed.return_value = False
+ pod = _pod("pod-1")
+ mock_get_pods.return_value = [pod]
+ mock_hook.return_value.get_pod.return_value = pod
+
+ op = KubernetesJobOperator(
+ task_id="test_task_id", wait_until_job_complete=True, on_finish_action="keep_pod"
+ )
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ assert pm.deleted == []
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleanup_error_on_success_does_not_raise(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """A delete failure on one pod must not stop cleanup of the others nor fail the task.
+
+ Exercises the real ``post_complete_action`` against a recording pod_manager
+ whose first ``delete_pod`` raises at the K8s SDK boundary.
+ """
+ mock_hook.return_value.is_job_failed.return_value = False
+ pod_1 = _pod("pod-1")
+ pod_2 = _pod("pod-2")
+ mock_get_pods.return_value = [pod_1, pod_2]
+ mock_find_pod.side_effect = [pod_1, pod_2]
+ mock_hook.return_value.get_pod.side_effect = lambda name, namespace: (
+ pod_1 if name == "pod-1" else pod_2
+ )
+ pm = _recording_pod_manager()
+ original_side_effect = pm.delete_pod.side_effect
+
+ def delete_side_effect(pod):
+ if pod.metadata.name == "pod-1":
+ raise ApiException(status=500, reason="boom")
+ original_side_effect(pod)
+
+ pm.delete_pod.side_effect = delete_side_effect
+ mock_pod_manager_prop.return_value = pm
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True, parallelism=2)
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ # pod-1's delete blew up; pod-2 still got deleted.
+ assert pm.deleted == ["pod-2"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleanup_error_does_not_mask_job_failure(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """When the job already failed, a cleanup error must not replace the job-level error.
+
+ The recording side_effect lets us observe that cleanup was *attempted*
+ even when it raises — the contract is "attempt cleanup, swallow any
+ error, preserve the original job failure".
+ """
+ mock_hook.return_value.is_job_failed.return_value = "Error"
+ pod = _pod("pod-1", phase="Failed")
+ mock_get_pods.return_value = [pod]
+ mock_find_pod.return_value = pod
+ mock_hook.return_value.get_pod.return_value = pod
+ attempted: list[str] = []
+
+ def boom(pod):
+ attempted.append(pod.metadata.name)
+ raise ApiException(status=500, reason="boom")
+
+ pm = mock.MagicMock()
+ pm.delete_pod.side_effect = boom
+ mock_pod_manager_prop.return_value = pm
+
Review Comment:
This new `MagicMock` has no spec, which makes the test less effective at catching typos in the mocked `pod_manager` surface. Please use an autospec or a small explicit fake with `delete_pod` instead.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_job.py:
##########
@@ -1193,6 +1278,279 @@ def test_create_zero_parallelism_fails_validation(
mock_hook.return_value.create_job.assert_not_called()
mock_get_pods.assert_not_called()
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleans_up_all_pods_on_success(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """Every monitoring pod is deleted after a successful job, including under ``parallelism>1``.
+
+ Real ``post_complete_action`` runs; only the K8s SDK boundary is faked.
+ """
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.return_value.is_job_failed.return_value = False
+ pod_1 = _pod("pod-1")
+ pod_2 = _pod("pod-2")
+ mock_get_pods.return_value = [pod_1, pod_2]
+ mock_find_pod.side_effect = [pod_1, pod_2]
+ mock_hook.return_value.get_pod.side_effect = lambda name, namespace: (
+ pod_1 if name == "pod-1" else pod_2
+ )
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True, parallelism=2)
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ assert sorted(pm.deleted) == ["pod-1", "pod-2"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleans_up_pod_on_failure(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """When the job fails, the monitoring pod is still deleted and the error propagates."""
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.return_value.is_job_failed.return_value = "Error"
+ pod = _pod("pod-1", phase="Failed")
+ mock_get_pods.return_value = [pod]
+ mock_find_pod.return_value = pod
+ mock_hook.return_value.get_pod.return_value = pod
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True)
+ with pytest.raises(AirflowException, match="is failed with error"):
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ assert pm.deleted == ["pod-1"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_keep_pod_skips_cleanup(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_pod_manager_prop,
+ ):
+ """``on_finish_action=keep_pod`` leaves the monitoring pod untouched."""
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.return_value.is_job_failed.return_value = False
+ pod = _pod("pod-1")
+ mock_get_pods.return_value = [pod]
+ mock_hook.return_value.get_pod.return_value = pod
+
+ op = KubernetesJobOperator(
+ task_id="test_task_id", wait_until_job_complete=True, on_finish_action="keep_pod"
+ )
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ assert pm.deleted == []
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleanup_error_on_success_does_not_raise(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """A delete failure on one pod must not stop cleanup of the others nor fail the task.
+
+ Exercises the real ``post_complete_action`` against a recording pod_manager
+ whose first ``delete_pod`` raises at the K8s SDK boundary.
+ """
+ mock_hook.return_value.is_job_failed.return_value = False
+ pod_1 = _pod("pod-1")
+ pod_2 = _pod("pod-2")
+ mock_get_pods.return_value = [pod_1, pod_2]
+ mock_find_pod.side_effect = [pod_1, pod_2]
+ mock_hook.return_value.get_pod.side_effect = lambda name, namespace: (
+ pod_1 if name == "pod-1" else pod_2
+ )
+ pm = _recording_pod_manager()
+ original_side_effect = pm.delete_pod.side_effect
+
+ def delete_side_effect(pod):
+ if pod.metadata.name == "pod-1":
+ raise ApiException(status=500, reason="boom")
+ original_side_effect(pod)
+
+ pm.delete_pod.side_effect = delete_side_effect
+ mock_pod_manager_prop.return_value = pm
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True, parallelism=2)
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ # pod-1's delete blew up; pod-2 still got deleted.
+ assert pm.deleted == ["pod-2"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleanup_error_does_not_mask_job_failure(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """When the job already failed, a cleanup error must not replace the job-level error.
+
+ The recording side_effect lets us observe that cleanup was *attempted*
+ even when it raises — the contract is "attempt cleanup, swallow any
+ error, preserve the original job failure".
+ """
+ mock_hook.return_value.is_job_failed.return_value = "Error"
+ pod = _pod("pod-1", phase="Failed")
+ mock_get_pods.return_value = [pod]
+ mock_find_pod.return_value = pod
+ mock_hook.return_value.get_pod.return_value = pod
+ attempted: list[str] = []
+
+ def boom(pod):
+ attempted.append(pod.metadata.name)
+ raise ApiException(status=500, reason="boom")
+
+ pm = mock.MagicMock()
+ pm.delete_pod.side_effect = boom
+ mock_pod_manager_prop.return_value = pm
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True)
+ # The job-level error wins; the cleanup ApiException is swallowed.
+ with pytest.raises(AirflowException, match="is failed with error"):
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ # Cleanup was still attempted even though it raised.
+ assert attempted == ["pod-1"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.execute_deferrable"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_does_not_cleanup_when_deferring(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_execute_deferrable,
+ mock_pod_manager_prop,
+ ):
+ """A ``TaskDeferred`` on the way out of ``execute()`` must leave pods alive.
+
+ The trigger still needs to watch them; cleanup happens on resume in
+ ``execute_complete``.
+ """
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ pod = _pod("pod-1")
+ mock_get_pods.return_value = [pod]
+ mock_execute_deferrable.side_effect = TaskDeferred(
+ trigger=mock.MagicMock(), method_name="execute_complete"
+ )
Review Comment:
This new mock is unspecced, so the test would not catch accidental calls to attributes that do not exist on the trigger object. Please use a specced/autospecced trigger mock here.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl merged PR #67333:
URL: https://github.com/apache/airflow/pull/67333
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3296560742
##########
providers/cncf/kubernetes/newsfragments/kubernetes-job-operator-cleanup-monitoring-pods.bugfix.rst:
##########
Review Comment:
@jscheffl @potiuk newsfragments removed and lint issue fixed
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3308936954
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_job.py:
##########
@@ -883,29 +903,73 @@ def test_execute_complete_pod_not_found_skips_logs(self, mock_hook, mocked_write
mocked_write_logs.assert_not_called()
@pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
@patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator._write_logs"))
@patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.hook"))
- def test_execute_complete_pod_api_error_reraises(self, mock_hook, mocked_write_logs):
- """Non-404 ApiExceptions should still be raised."""
- from kubernetes.client.rest import ApiException
+ def test_execute_complete_pod_api_error_does_not_raise(
+ self, mock_hook, mocked_write_logs, mock_pod_manager_prop
+ ):
+ """Non-404 errors fetching the monitoring pod must not fail ``execute_complete``.
- mock_ti = mock.MagicMock()
- context = {"ti": mock_ti}
- mock_job = mock.MagicMock()
+ Cleanup is best-effort: log writing is skipped, but the pod is still
+ targeted for deletion by name so it doesn't outlive the job.
+ """
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.get_pod.side_effect = ApiException(status=403, reason="Forbidden")
event = {
- "job": mock_job,
+ "job": mock.MagicMock(),
"status": "success",
Review Comment:
fixed
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #67333:
URL: https://github.com/apache/airflow/pull/67333#issuecomment-4518916539
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our [Contributors' Guide](https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [prek-hooks]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-prek-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/airflow-core/docs/howto/custom-operator.rst) Consider adding an example Dag that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on code in PR #67333:
URL: https://github.com/apache/airflow/pull/67333#discussion_r3308942796
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_job.py:
##########
@@ -1193,6 +1278,279 @@ def test_create_zero_parallelism_fails_validation(
mock_hook.return_value.create_job.assert_not_called()
mock_get_pods.assert_not_called()
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleans_up_all_pods_on_success(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """Every monitoring pod is deleted after a successful job, including under ``parallelism>1``.
+
+ Real ``post_complete_action`` runs; only the K8s SDK boundary is faked.
+ """
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.return_value.is_job_failed.return_value = False
+ pod_1 = _pod("pod-1")
+ pod_2 = _pod("pod-2")
+ mock_get_pods.return_value = [pod_1, pod_2]
+ mock_find_pod.side_effect = [pod_1, pod_2]
+ mock_hook.return_value.get_pod.side_effect = lambda name, namespace: (
+ pod_1 if name == "pod-1" else pod_2
+ )
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True, parallelism=2)
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ assert sorted(pm.deleted) == ["pod-1", "pod-2"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleans_up_pod_on_failure(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """When the job fails, the monitoring pod is still deleted and the error propagates."""
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.return_value.is_job_failed.return_value = "Error"
+ pod = _pod("pod-1", phase="Failed")
+ mock_get_pods.return_value = [pod]
+ mock_find_pod.return_value = pod
+ mock_hook.return_value.get_pod.return_value = pod
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True)
+ with pytest.raises(AirflowException, match="is failed with error"):
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ assert pm.deleted == ["pod-1"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_keep_pod_skips_cleanup(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_pod_manager_prop,
+ ):
+ """``on_finish_action=keep_pod`` leaves the monitoring pod untouched."""
+ pm = _recording_pod_manager()
+ mock_pod_manager_prop.return_value = pm
+ mock_hook.return_value.is_job_failed.return_value = False
+ pod = _pod("pod-1")
+ mock_get_pods.return_value = [pod]
+ mock_hook.return_value.get_pod.return_value = pod
+
+ op = KubernetesJobOperator(
+ task_id="test_task_id", wait_until_job_complete=True, on_finish_action="keep_pod"
+ )
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ assert pm.deleted == []
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleanup_error_on_success_does_not_raise(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """A delete failure on one pod must not stop cleanup of the others nor fail the task.
+
+ Exercises the real ``post_complete_action`` against a recording pod_manager
+ whose first ``delete_pod`` raises at the K8s SDK boundary.
+ """
+ mock_hook.return_value.is_job_failed.return_value = False
+ pod_1 = _pod("pod-1")
+ pod_2 = _pod("pod-2")
+ mock_get_pods.return_value = [pod_1, pod_2]
+ mock_find_pod.side_effect = [pod_1, pod_2]
+ mock_hook.return_value.get_pod.side_effect = lambda name, namespace: (
+ pod_1 if name == "pod-1" else pod_2
+ )
+ pm = _recording_pod_manager()
+ original_side_effect = pm.delete_pod.side_effect
+
+ def delete_side_effect(pod):
+ if pod.metadata.name == "pod-1":
+ raise ApiException(status=500, reason="boom")
+ original_side_effect(pod)
+
+ pm.delete_pod.side_effect = delete_side_effect
+ mock_pod_manager_prop.return_value = pm
+
+ op = KubernetesJobOperator(task_id="test_task_id", wait_until_job_complete=True, parallelism=2)
+ op.execute(context={"ti": mock.create_autospec(TaskInstance, instance=True)})
+
+ # pod-1's delete blew up; pod-2 still got deleted.
+ assert pm.deleted == ["pod-2"]
+
+ @pytest.mark.non_db_test_override
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.pod_manager"), new_callable=mock.PropertyMock)
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.find_pod"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.get_pods"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.build_job_request_obj"))
+ @patch(JOB_OPERATORS_PATH.format("KubernetesJobOperator.create_job"))
+ @patch(HOOK_CLASS)
+ def test_execute_cleanup_error_does_not_mask_job_failure(
+ self,
+ mock_hook,
+ mock_create_job,
+ mock_build_job_request_obj,
+ mock_get_pods,
+ mock_find_pod,
+ mock_pod_manager_prop,
+ ):
+ """When the job already failed, a cleanup error must not replace the job-level error.
+
+ The recording side_effect lets us observe that cleanup was *attempted*
+ even when it raises — the contract is "attempt cleanup, swallow any
+ error, preserve the original job failure".
+ """
+ mock_hook.return_value.is_job_failed.return_value = "Error"
+ pod = _pod("pod-1", phase="Failed")
+ mock_get_pods.return_value = [pod]
+ mock_find_pod.return_value = pod
+ mock_hook.return_value.get_pod.return_value = pod
+ attempted: list[str] = []
+
+ def boom(pod):
+ attempted.append(pod.metadata.name)
+ raise ApiException(status=500, reason="boom")
+
+ pm = mock.MagicMock()
+ pm.delete_pod.side_effect = boom
+ mock_pod_manager_prop.return_value = pm
+
Review Comment:
fixed https://github.com/apache/airflow/pull/67333/commits/d42fd91c5b16ac1d40d530b893576e5b7a11b40a
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix monitoring-pod leak in KubernetesJobOperator [airflow]
Posted by "jykae (via GitHub)" <gi...@apache.org>.
jykae commented on PR #67333:
URL: https://github.com/apache/airflow/pull/67333#issuecomment-4562500636
NOTE: added related Helm chart feature workerNamespaces that allows airflow to manage pods in specific namespaces rather than requiring access to whole cluster
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org