You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "gingeekrishna (via GitHub)" <gi...@apache.org> on 2026/05/16 09:01:14 UTC
[PR] Add zombie KPO pod cleanup feature for KubernetesPodOperator [airflow]
gingeekrishna opened a new pull request, #67030:
URL: https://github.com/apache/airflow/pull/67030
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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] Add zombie KPO pod cleanup feature for KubernetesPodOperator [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #67030:
URL: https://github.com/apache/airflow/pull/67030#issuecomment-4466751106
Note: Before merging main hoping that simple static checks fix themself: Please check contribution docs how to run `Prek` that will fox most of the findings of CI: https://github.com/apache/airflow/blob/main/contributing-docs/README.rst
--
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] Add zombie KPO pod cleanup feature for KubernetesPodOperator [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #67030:
URL: https://github.com/apache/airflow/pull/67030#issuecomment-4466440341
This PR lags any description. Can you please add any intention?
--
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] Add zombie KPO pod cleanup feature for KubernetesPodOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #67030:
URL: https://github.com/apache/airflow/pull/67030#issuecomment-4657822485
@gingeekrishna Converting to **draft** — this PR does not yet meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
- :x: **Merge conflicts** with `main`. See [docs](https://github.com/apache/airflow/blob/main/contributing-docs/10_working_with_git.rst).
See the linked criteria for how to fix each item, then mark the PR "Ready for review". This is **not** a rejection — just an invitation to bring the PR up to standard. 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._
--
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] feat: add zombie KPO pod cleanup to KubernetesExecutor [airflow]
Posted by "gingeekrishna (via GitHub)" <gi...@apache.org>.
gingeekrishna commented on PR #67030:
URL: https://github.com/apache/airflow/pull/67030#issuecomment-4722466677
> Hi @eladkal — fixed! The test failure had two root causes in `_cleanup_zombie_kpo_pods`:
>
> 1. **No phase filter** — the pod list included all pods (Pending, Completed, newly submitted), not just Running ones as the feature intends. Added `field_selector="status.phase=Running"` to scope it correctly.
> 2. **No age guard** — a pod with no matching TaskInstance in the DB (including system-test pods like `dag_id="dag"`) was immediately treated as a zombie and force-deleted. Added a minimum-age check: pods younger than `zombie_kpo_pod_cleanup_interval` seconds are now skipped.
>
> Also rebased onto `main` to clean up the merge commits from the diff.
Hi @eladkal
is that my PR good to merge now?
--
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] Add zombie KPO pod cleanup feature for KubernetesPodOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #67030:
URL: https://github.com/apache/airflow/pull/67030#issuecomment-4529582125
@gingeekrishna 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: **CI fails**: `CI image checks / Static checks` (and possibly other checks — see the [Checks tab](https://github.com/apache/airflow/pull/67030/checks) for the full list).
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] feat: add zombie KPO pod cleanup to KubernetesExecutor [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #67030:
URL: https://github.com/apache/airflow/pull/67030#issuecomment-4688397489
Please fix the tests
```
FAILED tests/kubernetes_tests/test_kubernetes_pod_operator.py::TestKubernetesPodOperatorSystem::test_full_pod_spec - kubernetes.client.exceptions.NotFoundException: (404)
Reason: Not Found
HTTP response headers: HTTPHeaderDict({'Audit-Id': 'de64a3d2-dbab-4e4a-befd-9ec7094844de', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Kubernetes-Pf-Flowschema-Uid': 'a5a8be79-1bdd-4dda-a9ac-f32865598441', 'X-Kubernetes-Pf-Prioritylevel-Uid': '2961fe89-97fb-4331-a496-064497325545', 'Date': 'Thu, 11 Jun 2026 13:45:07 GMT', 'Content-Length': '202'})
HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"pods \"test-pod-bbut043z\" not found","reason":"NotFound","details":{"name":"test-pod-bbut043z","kind":"pods"},"code":404}
```
--
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] feat: add zombie KPO pod cleanup to KubernetesExecutor [airflow]
Posted by "gingeekrishna (via GitHub)" <gi...@apache.org>.
gingeekrishna commented on PR #67030:
URL: https://github.com/apache/airflow/pull/67030#issuecomment-4663912136
> I am not a K8sExecutor user so my experience is a bit limited. Would ask some other maintainer for opinion.
Thank you @jscheffl!
--
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] feat: add zombie KPO pod cleanup to KubernetesExecutor [airflow]
Posted by "gingeekrishna (via GitHub)" <gi...@apache.org>.
gingeekrishna commented on PR #67030:
URL: https://github.com/apache/airflow/pull/67030#issuecomment-4694829339
Hi @eladkal — fixed! The test failure had two root causes in `_cleanup_zombie_kpo_pods`:
1. **No phase filter** — the pod list included all pods (Pending, Completed, newly submitted), not just Running ones as the feature intends. Added `field_selector="status.phase=Running"` to scope it correctly.
2. **No age guard** — a pod with no matching TaskInstance in the DB (including system-test pods like `dag_id="dag"`) was immediately treated as a zombie and force-deleted. Added a minimum-age check: pods younger than `zombie_kpo_pod_cleanup_interval` seconds are now skipped.
Also rebased onto `main` to clean up the merge commits from the diff.
--
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] Add zombie KPO pod cleanup feature for KubernetesPodOperator [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #67030:
URL: https://github.com/apache/airflow/pull/67030#discussion_r3252572837
##########
providers/cncf/kubernetes/docs/changelog.rst:
##########
Review Comment:
We are not adding text to changelog in PRs - we alsways have this made by release managers. Please add the description to other documentstion pages.
--
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] feat: add zombie KPO pod cleanup to KubernetesExecutor [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #67030:
URL: https://github.com/apache/airflow/pull/67030#discussion_r3383332322
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -824,6 +831,133 @@ def _adopt_completed_pods(self, kube_client: client.CoreV1Api) -> None:
)
)
+ # Operators whose pods this cleanup manages. EksPodOperator and
+ # GKEStartPodOperator are intentionally excluded: their pods run in
+ # external clusters where this kube_client has no authority.
+ _KPO_OPERATORS: frozenset[str] = frozenset(
+ [
+ "KubernetesPodOperator",
+ "SparkKubernetesOperator",
+ ]
+ )
+
+ @provide_session
+ def _cleanup_zombie_kpo_pods(
+ self, kube_client: client.CoreV1Api, *, session: Session = NEW_SESSION
+ ) -> None:
+ """
+ Force-delete KubernetesPodOperator pods whose TaskInstance is no longer active.
+
+ A pod is a zombie when either:
+
+ - No matching non-terminal TaskInstance exists in the DB (the TI already
+ finished or was never recorded), or
+ - The pod's ``try_number`` label is less than the active TI's current
+ ``try_number`` (the pod is a leftover from a previous retry attempt).
+
+ Force-deletion (``grace_period_seconds=0``) is used so that sidecar
+ containers that ignore SIGTERM cannot delay pod termination.
+ """
+ from sqlalchemy import or_
+
+ from airflow.models.taskinstance import TaskInstance
+ from airflow.providers.cncf.kubernetes.pod_generator import make_safe_label_value
+ from airflow.utils.state import State
+
+ pod_list = self._list_pods({"label_selector": "kubernetes_pod_operator=True"})
+ if not pod_list:
+ return
+
+ # Parse pod labels. Values were already normalized by make_safe_label_value()
+ # at pod-creation time (operators/pod.py get_labels()), so they are ready to
+ # compare directly against normalized DB values.
+ pod_identities = []
+ for pod in pod_list:
+ labels = pod.metadata.labels or {}
+ dag_id = labels.get("dag_id")
+ task_id = labels.get("task_id")
+ run_id = labels.get("run_id")
+ if not (dag_id and task_id and run_id):
+ continue
+ map_index = int(labels.get("map_index", -1))
+ try_number = int(labels.get("try_number", 0))
+ pod_identities.append((pod, dag_id, task_id, run_id, map_index, try_number))
+
+ if not pod_identities:
+ return
+
+ # Single batch query: all non-terminal TIs for KPO-managed operators.
+ # We match on notin_(finished) rather than in_(unfinished) because SQL
+ # evaluates "NULL NOT IN (...)" as NULL, not TRUE — the explicit is_(None)
+ # arm ensures we also capture TIs with no state set yet.
+ terminal_state_values = [s.value for s in State.finished]
+ active_tis = session.execute(
+ select(
+ TaskInstance.dag_id,
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.try_number,
+ ).where(
+ or_(
+ TaskInstance.state.notin_(terminal_state_values),
+ TaskInstance.state.is_(None),
+ ),
+ TaskInstance.operator.in_(self._KPO_OPERATORS),
+ )
+ ).all()
Review Comment:
This is an very expensive operation, on one side because by this select you might spool-out thousands of tasks as well as the source table might contain millions of rows. In my view a better selection strategy should be used. The approach might be feasible for small installations only.
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -824,6 +831,133 @@ def _adopt_completed_pods(self, kube_client: client.CoreV1Api) -> None:
)
)
+ # Operators whose pods this cleanup manages. EksPodOperator and
+ # GKEStartPodOperator are intentionally excluded: their pods run in
+ # external clusters where this kube_client has no authority.
+ _KPO_OPERATORS: frozenset[str] = frozenset(
+ [
+ "KubernetesPodOperator",
+ "SparkKubernetesOperator",
+ ]
+ )
+
+ @provide_session
+ def _cleanup_zombie_kpo_pods(
+ self, kube_client: client.CoreV1Api, *, session: Session = NEW_SESSION
+ ) -> None:
+ """
+ Force-delete KubernetesPodOperator pods whose TaskInstance is no longer active.
+
+ A pod is a zombie when either:
+
+ - No matching non-terminal TaskInstance exists in the DB (the TI already
+ finished or was never recorded), or
+ - The pod's ``try_number`` label is less than the active TI's current
+ ``try_number`` (the pod is a leftover from a previous retry attempt).
+
+ Force-deletion (``grace_period_seconds=0``) is used so that sidecar
+ containers that ignore SIGTERM cannot delay pod termination.
+ """
+ from sqlalchemy import or_
+
+ from airflow.models.taskinstance import TaskInstance
+ from airflow.providers.cncf.kubernetes.pod_generator import make_safe_label_value
+ from airflow.utils.state import State
+
+ pod_list = self._list_pods({"label_selector": "kubernetes_pod_operator=True"})
+ if not pod_list:
+ return
+
+ # Parse pod labels. Values were already normalized by make_safe_label_value()
+ # at pod-creation time (operators/pod.py get_labels()), so they are ready to
+ # compare directly against normalized DB values.
+ pod_identities = []
+ for pod in pod_list:
+ labels = pod.metadata.labels or {}
+ dag_id = labels.get("dag_id")
+ task_id = labels.get("task_id")
+ run_id = labels.get("run_id")
+ if not (dag_id and task_id and run_id):
+ continue
+ map_index = int(labels.get("map_index", -1))
+ try_number = int(labels.get("try_number", 0))
+ pod_identities.append((pod, dag_id, task_id, run_id, map_index, try_number))
+
+ if not pod_identities:
+ return
+
+ # Single batch query: all non-terminal TIs for KPO-managed operators.
+ # We match on notin_(finished) rather than in_(unfinished) because SQL
+ # evaluates "NULL NOT IN (...)" as NULL, not TRUE — the explicit is_(None)
+ # arm ensures we also capture TIs with no state set yet.
+ terminal_state_values = [s.value for s in State.finished]
+ active_tis = session.execute(
+ select(
+ TaskInstance.dag_id,
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.try_number,
+ ).where(
+ or_(
+ TaskInstance.state.notin_(terminal_state_values),
+ TaskInstance.state.is_(None),
+ ),
+ TaskInstance.operator.in_(self._KPO_OPERATORS),
Review Comment:
This is somehow mis-placed. if you want to clean KPO then this is not correct from side of KubernetesExecutor. If you use K8sExecutor you typically do not use KPO. And if you use another executor then the code path is not active.
--
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] feat: add zombie KPO pod cleanup to KubernetesExecutor [airflow]
Posted by "gingeekrishna (via GitHub)" <gi...@apache.org>.
gingeekrishna commented on code in PR #67030:
URL: https://github.com/apache/airflow/pull/67030#discussion_r3383764618
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -824,6 +831,133 @@ def _adopt_completed_pods(self, kube_client: client.CoreV1Api) -> None:
)
)
+ # Operators whose pods this cleanup manages. EksPodOperator and
+ # GKEStartPodOperator are intentionally excluded: their pods run in
+ # external clusters where this kube_client has no authority.
+ _KPO_OPERATORS: frozenset[str] = frozenset(
+ [
+ "KubernetesPodOperator",
+ "SparkKubernetesOperator",
+ ]
+ )
+
+ @provide_session
+ def _cleanup_zombie_kpo_pods(
+ self, kube_client: client.CoreV1Api, *, session: Session = NEW_SESSION
+ ) -> None:
+ """
+ Force-delete KubernetesPodOperator pods whose TaskInstance is no longer active.
+
+ A pod is a zombie when either:
+
+ - No matching non-terminal TaskInstance exists in the DB (the TI already
+ finished or was never recorded), or
+ - The pod's ``try_number`` label is less than the active TI's current
+ ``try_number`` (the pod is a leftover from a previous retry attempt).
+
+ Force-deletion (``grace_period_seconds=0``) is used so that sidecar
+ containers that ignore SIGTERM cannot delay pod termination.
+ """
+ from sqlalchemy import or_
+
+ from airflow.models.taskinstance import TaskInstance
+ from airflow.providers.cncf.kubernetes.pod_generator import make_safe_label_value
+ from airflow.utils.state import State
+
+ pod_list = self._list_pods({"label_selector": "kubernetes_pod_operator=True"})
+ if not pod_list:
+ return
+
+ # Parse pod labels. Values were already normalized by make_safe_label_value()
+ # at pod-creation time (operators/pod.py get_labels()), so they are ready to
+ # compare directly against normalized DB values.
+ pod_identities = []
+ for pod in pod_list:
+ labels = pod.metadata.labels or {}
+ dag_id = labels.get("dag_id")
+ task_id = labels.get("task_id")
+ run_id = labels.get("run_id")
+ if not (dag_id and task_id and run_id):
+ continue
+ map_index = int(labels.get("map_index", -1))
+ try_number = int(labels.get("try_number", 0))
+ pod_identities.append((pod, dag_id, task_id, run_id, map_index, try_number))
+
+ if not pod_identities:
+ return
+
+ # Single batch query: all non-terminal TIs for KPO-managed operators.
+ # We match on notin_(finished) rather than in_(unfinished) because SQL
+ # evaluates "NULL NOT IN (...)" as NULL, not TRUE — the explicit is_(None)
+ # arm ensures we also capture TIs with no state set yet.
+ terminal_state_values = [s.value for s in State.finished]
+ active_tis = session.execute(
+ select(
+ TaskInstance.dag_id,
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.try_number,
+ ).where(
+ or_(
+ TaskInstance.state.notin_(terminal_state_values),
+ TaskInstance.state.is_(None),
+ ),
+ TaskInstance.operator.in_(self._KPO_OPERATORS),
+ )
+ ).all()
Review Comment:
Good catch — fixed in the latest commit. The query is now scoped to `TaskInstance.dag_id.in_(pod_dag_ids)`, where `pod_dag_ids` is the set of DAG IDs extracted from the live KPO pod labels. This bounds the result set to TIs from DAGs that actually have running KPO pods, rather than scanning all non-terminal KPO TIs across the entire table.
A note on correctness: pod label values are normalized by `make_safe_label_value()` at pod-creation time. For `dag_id` specifically, the encoding is almost always a no-op (dag IDs are short alphanumeric strings), so the filter correctly scopes the query in practice. The remaining normalization comparison (for `task_id`, `run_id`, `map_index`) still happens in Python on the reduced result set.
--
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] feat: add zombie KPO pod cleanup to KubernetesExecutor [airflow]
Posted by "gingeekrishna (via GitHub)" <gi...@apache.org>.
gingeekrishna commented on PR #67030:
URL: https://github.com/apache/airflow/pull/67030#issuecomment-4658577813
> This PR lags any description. Can you please add any intention?
Apologies for that, I just updated the right description
--
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] feat: add zombie KPO pod cleanup to KubernetesExecutor [airflow]
Posted by "gingeekrishna (via GitHub)" <gi...@apache.org>.
gingeekrishna commented on code in PR #67030:
URL: https://github.com/apache/airflow/pull/67030#discussion_r3383765840
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -824,6 +831,133 @@ def _adopt_completed_pods(self, kube_client: client.CoreV1Api) -> None:
)
)
+ # Operators whose pods this cleanup manages. EksPodOperator and
+ # GKEStartPodOperator are intentionally excluded: their pods run in
+ # external clusters where this kube_client has no authority.
+ _KPO_OPERATORS: frozenset[str] = frozenset(
+ [
+ "KubernetesPodOperator",
+ "SparkKubernetesOperator",
+ ]
+ )
+
+ @provide_session
+ def _cleanup_zombie_kpo_pods(
+ self, kube_client: client.CoreV1Api, *, session: Session = NEW_SESSION
+ ) -> None:
+ """
+ Force-delete KubernetesPodOperator pods whose TaskInstance is no longer active.
+
+ A pod is a zombie when either:
+
+ - No matching non-terminal TaskInstance exists in the DB (the TI already
+ finished or was never recorded), or
+ - The pod's ``try_number`` label is less than the active TI's current
+ ``try_number`` (the pod is a leftover from a previous retry attempt).
+
+ Force-deletion (``grace_period_seconds=0``) is used so that sidecar
+ containers that ignore SIGTERM cannot delay pod termination.
+ """
+ from sqlalchemy import or_
+
+ from airflow.models.taskinstance import TaskInstance
+ from airflow.providers.cncf.kubernetes.pod_generator import make_safe_label_value
+ from airflow.utils.state import State
+
+ pod_list = self._list_pods({"label_selector": "kubernetes_pod_operator=True"})
+ if not pod_list:
+ return
+
+ # Parse pod labels. Values were already normalized by make_safe_label_value()
+ # at pod-creation time (operators/pod.py get_labels()), so they are ready to
+ # compare directly against normalized DB values.
+ pod_identities = []
+ for pod in pod_list:
+ labels = pod.metadata.labels or {}
+ dag_id = labels.get("dag_id")
+ task_id = labels.get("task_id")
+ run_id = labels.get("run_id")
+ if not (dag_id and task_id and run_id):
+ continue
+ map_index = int(labels.get("map_index", -1))
+ try_number = int(labels.get("try_number", 0))
+ pod_identities.append((pod, dag_id, task_id, run_id, map_index, try_number))
+
+ if not pod_identities:
+ return
+
+ # Single batch query: all non-terminal TIs for KPO-managed operators.
+ # We match on notin_(finished) rather than in_(unfinished) because SQL
+ # evaluates "NULL NOT IN (...)" as NULL, not TRUE — the explicit is_(None)
+ # arm ensures we also capture TIs with no state set yet.
+ terminal_state_values = [s.value for s in State.finished]
+ active_tis = session.execute(
+ select(
+ TaskInstance.dag_id,
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.try_number,
+ ).where(
+ or_(
+ TaskInstance.state.notin_(terminal_state_values),
+ TaskInstance.state.is_(None),
+ ),
+ TaskInstance.operator.in_(self._KPO_OPERATORS),
Review Comment:
This is a fair design question. The placement rationale is:
1. **KPO and KubernetesExecutor do coexist**: KubernetesExecutor runs each Airflow task in its own worker pod. One of those tasks can be a `KubernetesPodOperator` task, which then spawns a *second* pod for its actual work. When that KPO task finishes, Airflow signals the KPO-managed pod to terminate, but sidecar containers ignoring `SIGTERM` (xcom sidecar, log agents) can keep it running indefinitely. This is the scenario from issue #58968.
2. **`kube_client` is already available here**: `KubernetesExecutor`'s `sync()` loop already holds an authoritative `kube_client` for the cluster where both executor worker pods *and* KPO-managed pods run. It's the natural home for any periodic cleanup that needs Kubernetes API access.
3. **The limitation you identified is real**: Users running KPO with `CeleryExecutor` or `LocalExecutor` won't benefit from this cleanup since this code path is inactive for them. That's a known limitation. An alternative would be to put the cleanup in KPO's `execute()` itself, but a task that has already reached a terminal state from Airflow's perspective cannot reliably re-enter cleanup logic.
Happy to move or restructure it if there's a preferred pattern in the codebase — or if another maintainer with K8s executor experience has a stronger opinion.
--
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] feat: add zombie KPO pod cleanup to KubernetesExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #67030:
URL: https://github.com/apache/airflow/pull/67030#issuecomment-5021137796
@gingeekrishna This PR has been converted to **draft** because it does not yet meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
**Issues found:**
- :x: **Build docs**: Failing: `CI image checks / Build documentation (--docs-only)`, `CI image checks / Build documentation (--spellcheck-only)`. Run `breeze build-docs` locally to reproduce.
- :x: **Pre-commit / static checks**: Failing: `CI image checks / Static checks`. Run `prek run --from-ref main --stage pre-commit` locally to reproduce and fix.
- :x: **Provider tests**: Failing: `Provider distributions tests / Compat 2.11.1:P3.10:`, `Provider distributions tests / Compat 3.0.6:P3.10:`, `Provider distributions tests / Compat 3.1.8:P3.10:`, `Provider distributions tests / Compat 3.2.2:P3.10:`. Run `breeze testing providers-tests --test-type "Providers[<provider>]"` locally.
> **Note:** Your branch is **85 commits behind `main`**. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results.
**What to do next:**
- Fix each issue listed above.
- Make sure static checks pass locally (`prek run --from-ref main --stage pre-commit`).
- Mark the PR as "Ready for review" when you're done.
Converting a PR to draft is **not** a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
---
_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._
--
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] feat: add zombie KPO pod cleanup to KubernetesExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk closed pull request #67030: feat: add zombie KPO pod cleanup to KubernetesExecutor
URL: https://github.com/apache/airflow/pull/67030
--
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] Add zombie KPO pod cleanup feature for KubernetesPodOperator [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #67030:
URL: https://github.com/apache/airflow/pull/67030#issuecomment-4466404712
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