This is an automated email from the ASF dual-hosted git repository.
eladkal pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new eca48163048 Add `--min-completed-minutes` to `cleanup-pods` to prevent
KPO race condition (#70595)
eca48163048 is described below
commit eca48163048a46bba4a242e66096df3e34637f9e
Author: noamst-monday <[email protected]>
AuthorDate: Thu Jul 30 19:57:09 2026 +0300
Add `--min-completed-minutes` to `cleanup-pods` to prevent KPO race
condition (#70595)
* Add --min-completed-minutes to cleanup-pods to prevent KPO race condition
The ``airflow kubernetes cleanup-pods`` command currently deletes
Succeeded/Failed/Evicted pods immediately, with no minimum-age guard
for terminal states. ``KubernetesPodOperator`` in synchronous mode
polls pod status every ~2 seconds via ``await_pod_completion``. If
the cleanup job fires in the window between the pod reaching
``Succeeded`` and KPO's next poll, KPO receives a 404 and fails the
task -- even though the pod completed successfully (exit code 0).
A ``--min-pending-minutes`` guard already exists for Pending pods
(default 30 m, floor 5 m). No equivalent exists for terminal states.
This commit adds ``--min-completed-minutes`` (default ``0``, which
preserves the existing behaviour). When set to any positive value,
Succeeded/Failed/Evicted pods are skipped unless their completion time
is older than the threshold. Completion time is derived from the
latest ``containerStatuses[*].state.terminated.finishedAt`` timestamp
(falls back to ``metadata.creationTimestamp`` for pods that were
evicted before any container started).
Root-cause investigation
------------------------
This was confirmed via Kubernetes API server audit logs on a production
EKS cluster. Timeline for an affected KPO task:
13:15:12Z KPO polls pod → phase Running
13:15:14Z Container exits with code 0 (pod transitions to Succeeded)
13:15:17Z cleanup-pods CronJob deletes the pod (3 s after completion)
13:15:18Z KPO polls pod → 404 Not Found → task marked FAILED
The pod had succeeded; the task failure was a false positive caused
entirely by the race. Multiple production DAGs exhibited the same
pattern with the cleanup CronJob set to run every 5 minutes.
Reducing the CronJob frequency is a partial mitigation (lowers the
probability) but does not eliminate the race. Setting
``--min-completed-minutes=5`` gives KPO a 5-minute window to observe
the terminal phase -- 150x wider than the 2 s poll interval -- closing
the race completely in practice.
Changes
-------
* ``definition.py`` – add ``ARG_MIN_COMPLETED_MINUTES``; wire into
cleanup-pods args tuple
* ``kubernetes_command.py`` – add ``_get_pod_completion_time()`` helper;
gate terminal-pod deletion by age when
``min_completed_minutes > 0``
* ``test_kubernetes_command.py`` – 4 new unit tests covering:
- Succeeded pod too young → not deleted
- Succeeded pod old enough → deleted
- Default (0) preserves immediate-deletion behaviour
- Failed/Never pod too young → not deleted
* ``changelog.rst`` – entry under 10.21.0
CLI docs (``cli-ref.rst``) are auto-generated via ``.. argparse::``
and will pick up the new flag automatically.
* Update changelog with PR number #70595
* Fix ruff D213 docstring format in _get_pod_completion_time
* Use exact poll interval (2 s) in --min-completed-minutes help text
* Improve --min-completed-minutes help text clarity
* Revert changelog.rst — auto-generated by release manager
* Address jedcunningham review on cleanup-pods min-completed-minutes
- Set default to 1 minute (no reason to keep the race by default)
- Shorten --min-completed-minutes help text
- Fix _get_pod_completion_time: scan init_container_statuses too;
fall back to max(conditions.last_transition_time) instead of
creation_timestamp (which predates actual completion)
- Add TestGetPodCompletionTime with real k8s model objects covering
main-only, init-only, both, conditions fallback, and creation_timestamp
last-resort cases
- Fix test_cleanup_min_completed_zero_deletes_immediately to pass
--min-completed-minutes=0 explicitly now that default is 1
* Test cleanup-pods age guard for evicted and init-container pods
The guard must not fall through to creation_timestamp for pods whose
containers never reached a terminated state — that path reports an
inflated age and deletes immediately, the unsafe direction. Freezing
time keeps every case at a realistic completion offset instead of a
timestamp in the future.
* Cover a terminated container status with no finish time
_get_pod_completion_time guards three levels of the container state, but
the fallback cases only reached the first two, so a terminated status
carrying no finishedAt went unexercised. Modelling the non-terminated
case as a waiting container also matches what k8s reports for a pod
evicted before its containers started.
* Use a tuple for parametrize names in cleanup-pods tests
* Name the pending-pod age condition in cleanup-pods
---
.../providers/cncf/kubernetes/cli/definition.py | 12 +-
.../cncf/kubernetes/cli/kubernetes_command.py | 45 +++++-
.../cncf/kubernetes/cli/test_kubernetes_command.py | 180 +++++++++++++++++++++
3 files changed, 229 insertions(+), 8 deletions(-)
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
index 90929ffaf5f..cbcbfac9e45 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/definition.py
@@ -68,6 +68,16 @@ ARG_MIN_PENDING_MINUTES = Arg(
),
)
+ARG_MIN_COMPLETED_MINUTES = Arg(
+ ("--min-completed-minutes",),
+ default=1,
+ type=positive_int(allow_zero=True),
+ help=(
+ "Minimum age in minutes of a completed (Succeeded/Failed/Evicted) pod
before it is deleted. "
+ "Default is 1. Set to 0 to delete immediately."
+ ),
+)
+
ARG_TEAM = Arg(
("--team",),
default=None,
@@ -84,7 +94,7 @@ KUBERNETES_COMMANDS = (
"in evicted/failed/succeeded/pending states"
),
func=lazy_load_command("airflow.providers.cncf.kubernetes.cli.kubernetes_command.cleanup_pods"),
- args=(ARG_NAMESPACE, ARG_MIN_PENDING_MINUTES, ARG_VERBOSE),
+ args=(ARG_NAMESPACE, ARG_MIN_PENDING_MINUTES,
ARG_MIN_COMPLETED_MINUTES, ARG_VERBOSE),
),
ActionCommand(
name="generate-dag-yaml",
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
index ea59ae7755b..2d2490134dd 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py
@@ -119,6 +119,31 @@ def generate_pod_yaml(args):
print(f"YAML output can be found at {yaml_output_path}")
+def _get_pod_completion_time(pod):
+ """
+ Return the time the pod's last container finished.
+
+ Scans both ``container_statuses`` and ``init_container_statuses`` and
returns the
+ latest ``finished_at`` timestamp. Falls back to the latest condition
+ ``last_transition_time`` (which is updated at the terminal transition),
and finally
+ to ``creation_timestamp`` as a last resort.
+ """
+ statuses = [*(pod.status.container_statuses or []),
*(pod.status.init_container_statuses or [])]
+ times = [
+ s.state.terminated.finished_at
+ for s in statuses
+ if s.state and s.state.terminated and s.state.terminated.finished_at
+ ]
+ if times:
+ return max(times)
+ condition_times = [
+ c.last_transition_time for c in (pod.status.conditions or []) if
c.last_transition_time
+ ]
+ if condition_times:
+ return max(condition_times)
+ return pod.metadata.creation_timestamp
+
+
@cli_utils.action_cli(check_db=False)
@providers_configuration_loaded
def cleanup_pods(args):
@@ -130,6 +155,8 @@ def cleanup_pods(args):
if min_pending_minutes < 5:
min_pending_minutes = 5
+ min_completed_minutes = args.min_completed_minutes
+
# https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/
# All Containers in the Pod have terminated in success, and will not be
restarted.
pod_succeeded = "succeeded"
@@ -173,16 +200,20 @@ def cleanup_pods(args):
pod_restart_policy = pod.spec.restart_policy.lower()
current_time = datetime.now(pod.metadata.creation_timestamp.tzinfo)
- if (
+ is_terminal = (
pod_phase == pod_succeeded
or (pod_phase == pod_failed and pod_restart_policy ==
pod_restart_policy_never)
or (pod_reason == pod_reason_evicted)
- or (
- pod_phase == pod_pending
- and current_time - pod.metadata.creation_timestamp
- > timedelta(minutes=min_pending_minutes)
- )
- ):
+ )
+ is_terminal_old_enough = is_terminal and (
+ min_completed_minutes == 0
+ or current_time - _get_pod_completion_time(pod) >
timedelta(minutes=min_completed_minutes)
+ )
+ is_pending_too_long = (
+ pod_phase == pod_pending
+ and current_time - pod.metadata.creation_timestamp >
timedelta(minutes=min_pending_minutes)
+ )
+ if is_terminal_old_enough or is_pending_too_long:
print(
f'Deleting pod "{pod_name}" phase "{pod_phase}" and reason
"{pod_reason}", '
f'restart policy "{pod_restart_policy}"'
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
index 6ae8b81c50f..29eadcc55a2 100644
---
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
+++
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/cli/test_kubernetes_command.py
@@ -18,12 +18,15 @@ from __future__ import annotations
import importlib
import os
+from datetime import timedelta
from unittest import mock
from unittest.mock import MagicMock, call
import kubernetes
import pytest
+import time_machine
from dateutil.parser import parse
+from kubernetes.client import models as k8s
from airflow.cli import cli_parser
from airflow.executors import executor_loader
@@ -34,6 +37,47 @@ from tests_common.test_utils.version_compat import
AIRFLOW_V_3_0_PLUS
pytestmark = pytest.mark.db_test
+NOW = parse("2024-01-01T13:15:17Z")
+
+
+def make_container_status(state):
+ return k8s.V1ContainerStatus(
+ name="base", ready=False, restart_count=0, image="img", image_id="id",
state=state
+ )
+
+
+def make_terminated_status(finished_at):
+ return make_container_status(
+
k8s.V1ContainerState(terminated=k8s.V1ContainerStateTerminated(exit_code=0,
finished_at=finished_at))
+ )
+
+
+def make_terminal_pod(
+ name,
+ phase,
+ reason=None,
+ restart_policy="Never",
+ finished_at=None,
+ init_finished_at=None,
+ condition_time=None,
+):
+ conditions = (
+ [k8s.V1PodCondition(type="Ready", status="False",
last_transition_time=condition_time)]
+ if condition_time
+ else []
+ )
+ return k8s.V1Pod(
+ metadata=k8s.V1ObjectMeta(name=name, creation_timestamp=NOW -
timedelta(hours=1)),
+ spec=k8s.V1PodSpec(containers=[], restart_policy=restart_policy),
+ status=k8s.V1PodStatus(
+ phase=phase,
+ reason=reason,
+ container_statuses=[make_terminated_status(finished_at)] if
finished_at else [],
+ init_container_statuses=[make_terminated_status(init_finished_at)]
if init_finished_at else [],
+ conditions=conditions,
+ ),
+ )
+
class TestGenerateDagYamlCommand:
@classmethod
@@ -283,3 +327,139 @@ class TestCleanUpPodsCommand:
list_namespaced_pod.assert_has_calls(calls)
delete_pod.assert_called_with("dummy", "awesome-namespace")
load_incluster_config.assert_called_once()
+
+ @pytest.mark.parametrize(
+ ("pod_kwargs", "min_completed_minutes", "expect_deleted"),
+ [
+ pytest.param(
+ {"phase": "Succeeded", "finished_at": NOW -
timedelta(seconds=3)},
+ 1,
+ False,
+ id="succeeded-just-finished-kept",
+ ),
+ pytest.param(
+ {"phase": "Succeeded", "finished_at": NOW -
timedelta(minutes=5)},
+ 1,
+ True,
+ id="succeeded-old-enough-deleted",
+ ),
+ pytest.param(
+ {"phase": "Succeeded", "finished_at": NOW -
timedelta(seconds=3)},
+ 0,
+ True,
+ id="zero-disables-guard",
+ ),
+ pytest.param(
+ {"phase": "Failed", "finished_at": NOW - timedelta(seconds=3)},
+ 1,
+ False,
+ id="failed-just-finished-kept",
+ ),
+ pytest.param(
+ {"phase": "Failed", "init_finished_at": NOW -
timedelta(seconds=3)},
+ 1,
+ False,
+ id="init-container-failed-just-finished-kept",
+ ),
+ pytest.param(
+ {"phase": "Failed", "reason": "Evicted", "condition_time": NOW
- timedelta(seconds=3)},
+ 1,
+ False,
+ id="evicted-before-containers-started-kept",
+ ),
+ pytest.param(
+ {"phase": "Failed", "reason": "Evicted", "condition_time": NOW
- timedelta(minutes=5)},
+ 1,
+ True,
+ id="evicted-old-enough-deleted",
+ ),
+ ],
+ )
+ @time_machine.travel(NOW, tick=False)
+
@mock.patch("airflow.providers.cncf.kubernetes.cli.kubernetes_command._delete_pod")
+ @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_pod")
+ @mock.patch("kubernetes.config.load_incluster_config")
+ def test_cleanup_pods_min_completed_minutes(
+ self,
+ load_incluster_config,
+ list_namespaced_pod,
+ delete_pod,
+ pod_kwargs,
+ min_completed_minutes,
+ expect_deleted,
+ ):
+ pods = MagicMock()
+ pods.metadata._continue = None
+ pods.items = [make_terminal_pod("run-o1sxc2on", **pod_kwargs)]
+ list_namespaced_pod.return_value = pods
+ kubernetes_command.cleanup_pods(
+ self.parser.parse_args(
+ [
+ "kubernetes",
+ "cleanup-pods",
+ "--namespace",
+ "awesome-namespace",
+ "--min-completed-minutes",
+ str(min_completed_minutes),
+ ]
+ )
+ )
+ if expect_deleted:
+ delete_pod.assert_called_once_with("run-o1sxc2on",
"awesome-namespace")
+ else:
+ delete_pod.assert_not_called()
+
+
+class TestGetPodCompletionTime:
+ T1 = parse("2024-01-01T10:00:00Z")
+ T2 = parse("2024-01-01T10:05:00Z")
+ CREATED_AT = parse("2024-01-01T09:00:00Z") # earlier than T1/T2
+
+ def _pod(self, container_statuses=None, init_container_statuses=None,
conditions=None):
+ return k8s.V1Pod(
+ metadata=k8s.V1ObjectMeta(name="run-o1sxc2on",
creation_timestamp=self.CREATED_AT),
+ status=k8s.V1PodStatus(
+ container_statuses=container_statuses,
+ init_container_statuses=init_container_statuses,
+ conditions=conditions,
+ ),
+ )
+
+ def test_single_main_container(self):
+ pod = self._pod(container_statuses=[make_terminated_status(self.T1)])
+ assert kubernetes_command._get_pod_completion_time(pod) == self.T1
+
+ def test_single_init_container_no_main(self):
+ pod = self._pod(container_statuses=[],
init_container_statuses=[make_terminated_status(self.T1)])
+ assert kubernetes_command._get_pod_completion_time(pod) == self.T1
+
+ def test_main_and_init_returns_max(self):
+ pod = self._pod(
+ container_statuses=[make_terminated_status(self.T1)],
+ init_container_statuses=[make_terminated_status(self.T2)],
+ )
+ assert kubernetes_command._get_pod_completion_time(pod) == self.T2
+
+ @pytest.mark.parametrize(
+ "container_status",
+ [
+ pytest.param(make_container_status(state=None), id="no-state"),
+ pytest.param(
+ make_container_status(
+
k8s.V1ContainerState(waiting=k8s.V1ContainerStateWaiting(reason="ContainerCreating"))
+ ),
+ id="never-terminated",
+ ),
+ pytest.param(make_terminated_status(finished_at=None),
id="terminated-without-finished-at"),
+ ],
+ )
+ def test_falls_back_to_conditions(self, container_status):
+ pod = self._pod(
+ container_statuses=[container_status],
+ conditions=[k8s.V1PodCondition(type="Ready", status="False",
last_transition_time=self.T1)],
+ )
+ assert kubernetes_command._get_pod_completion_time(pod) == self.T1
+
+ def
test_no_containers_no_conditions_falls_back_to_creation_timestamp(self):
+ pod = self._pod(container_statuses=[], init_container_statuses=[],
conditions=[])
+ assert kubernetes_command._get_pod_completion_time(pod) ==
self.CREATED_AT
