Repository navigation
Fix KubernetesPodOperator 404 on pod preemption (#59626) - #60626
cruseakshay wants to merge 5 commits into
Conversation
Add state-aware retry for 404 errors when pods are preempted by daemonsets during node bootstrap. Only retry if pod never reached Running state to prevent duplicate execution of non-idempotent tasks.
boring-cyborg
Bot
added
area:providers
provider:cncf-kubernetes
jscheffl
left a comment
jscheffl
left a comment
There was a problem hiding this comment.
There are some questions and logic problems in this PR.
Not sure if I have a mis-understanding or what the aim of the fix is.
| if pod.status.phase == PodPhase.RUNNING: | ||
| self.ever_reached_running = True |
There was a problem hiding this comment.
How about the race condition if the Pod was so fast that it is already failed or completed when checking? I think it might be a bit dangerous checking just for running.
There was a problem hiding this comment.
for pods running very quickly, (Pending → Running → Succeeded/Failed),
if pod.status.phase in (PodPhase.RUNNING, PodPhase.SUCCEEDED, PodPhase.FAILED):
self.ever_started_execution = True
with this we can safely consider, any phase beyond Pending means pod was scheduled and not preempted.
| except HTTPError as e: | ||
| raise KubernetesApiException(f"There was an error reading the kubernetes API: {e}") | ||
|
|
||
| def _handle_pod_not_found(self, pod: V1Pod, phase_tracker: PodPhaseTracker | None) -> None: |
There was a problem hiding this comment.
The code here is redundant (and some more) here in the utility and the hook. please consolidate to one place and copy only.
There was a problem hiding this comment.
Sure, will consolidate under PodPhaseTracker.
| raise PodNotFoundException( | ||
| f"Pod '{name}' in namespace '{namespace}' not found. " | ||
| f"The pod was previously observed in Running state (last phase: " | ||
| f"{phase_tracker.last_observed_phase}), so this is treated as a " | ||
| f"terminal failure to prevent duplicate execution of non-idempotent workloads." | ||
| ) |
There was a problem hiding this comment.
That is not clear to me. There might be cases where a logic is only allowed to start only once and then a broken state must not retried - and there are cases where a failed execution still should retry.
I understand that you want to handle it differently but the exceptions are just different exceptions, but it is not changing anything in retry handling. Also the text is mis-leading because besides the exception types the "normal" retry would happen if configured?
For me this PR only makes sense if a configurable setting based on Exception type skips or makes the retry based on preemption.
There was a problem hiding this comment.
@jscheffl, thanks for the valuable feedback.
As task-level retries will, in turn, try pod re-creation, I am wondering if it is worth adding more complexity for internal retries. I will rely on your discretion here.
If I were to continue implementing - configurable part:
- modify operator to support config: user can opt-in for this change in behaviour
KubernetesPodOperator(
task_id="my_task",
retry_on_preemption=True, # Default False
...
)- modify triggerer: PodPreemptedException is caught, yield status="preempted"
- modify operator to distinguish preemption from other failures and decide whether to retry (based on config)
| def _should_retry_api(exc: BaseException) -> bool: | ||
| """Retry on selected ApiException status codes, plus plain HTTP/timeout errors.""" | ||
| """Retry on transient API errors, HTTP errors, and PodPreemptedException.""" | ||
| if isinstance(exc, PodPreemptedException): |
There was a problem hiding this comment.
This is wrong. The function is used to retry calling API if e.g. a HTTP 500 or 403 is returned - so on call level. I assume you wanted to make the Pod execution retryable - but this is not defined with this method.
Problem
When using
KubernetesPodOperatorwith cluster autoscaling, task pods can be preempted by higher-priority daemonsets during node bootstrap. This results in a 404 error when Airflow tries to read the pod status, causing immediate task failure instead of allowing Kubernetes to reschedule the pod.Fixes #59626
Solution
Introduce state-aware retry logic that tracks whether a pod ever reached the
Runningstate:PodPreemptedException(retriable)PodNotFoundException(terminal failure)This prevents duplicate execution of non-idempotent tasks while allowing safe retries for pods preempted before they started.
Changes
exceptions.pyPodNotFoundExceptionandPodPreemptedExceptionpod_manager.pyPodPhaseTrackerdataclass and 404 handling logichooks/kubernetes.pyget_pod()triggers/pod.pyKubernetesPodTriggerkubernetes_helper_functions.pyPodPreemptedExceptionto retry logictest_pod_manager.pyTesting
PodPhaseTrackerstate transitionsWas generative AI tooling used to co-author this PR?
Generated-by: Claude (Cursor) following the guidelines