This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-2-test by this push:
new 53717ecc0cf KPO: treat registry 5xx errors as transient during pod
startup (#65490) (#66170)
53717ecc0cf is described below
commit 53717ecc0cfeab7f9c0e49340008112b5e8ffdd3
Author: Jarek Potiuk <[email protected]>
AuthorDate: Thu Apr 30 20:32:07 2026 +0200
KPO: treat registry 5xx errors as transient during pod startup (#65490)
(#66170)
When a pod is starting, kubelet automatically retries failed image pulls
with exponential backoff. `KubernetesPodOperator` monitors the pod via
`detect_pod_terminate_early_issues()` and aborts early on what it deems a
fatal pull error — but the transient-error pattern list did not cover
registry 5xx / gateway outages. A short Docker Hub outage (502/503/504
from `auth.docker.io` or the registry) was being classified as fatal and
would abort the pod before kubelet's next retry, even though the pull
would have succeeded once upstream recovered.
Add `bad gateway`, `service unavailable`, `gateway timeout` to
`TRANSIENT_ERROR_PATTERNS` so these conditions let kubelet keep retrying
within the caller's `startup_timeout` budget.
On kubelet >= 1.32 the `ImagePullBackOff` message is
`Back-off pulling image "X": <prev_error>`, so these patterns also match
during the backoff state. On older kubelets the bare message carries no
detail and we keep the existing fail-fast behaviour — matching
`back-off pulling` unconditionally would have caused a 120s wait for a
genuinely missing image instead of an immediate error (thanks Jens for
catching this).
`startup_timeout` still bounds the overall wait, and genuinely fatal pull
errors (`InvalidImageName`, `ErrImageNeverPull`, `manifest unknown`,
`unauthorized`, …) remain fatal.
(cherry picked from commit 938ccf3b2fc3238d36eca621a4580fd9424d161d)
---
.../providers/cncf/kubernetes/utils/pod_manager.py | 12 +++++
.../unit/cncf/kubernetes/utils/test_pod_manager.py | 56 ++++++++++++++++++++++
2 files changed, 68 insertions(+)
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py
index 2c2405b7fd4..a48354bc3c3 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py
@@ -203,6 +203,18 @@ def detect_pod_terminate_early_issues(pod: V1Pod) -> str |
None:
"temporarily unavailable",
"timeout",
"account limit",
+ # Upstream registry/auth outages (e.g. Docker Hub auth returning 5xx).
+ # kubelet will retry automatically; the caller's startup_timeout still
+ # bounds the total wait. These match both the underlying ErrImagePull
+ # message and the ImagePullBackOff message on kubelet >= 1.32, which
+ # appends the previous pull error to "Back-off pulling image ...".
+ # On older kubelets the bare ImagePullBackOff message carries no
+ # detail, so we fall back to the existing fail-fast path — matching
+ # "back-off pulling" unconditionally would cause a 120s wait for a
+ # genuinely missing image instead of failing fast.
+ "bad gateway",
+ "service unavailable",
+ "gateway timeout",
]
FATAL_STATES = ["InvalidImageName", "ErrImageNeverPull"]
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_pod_manager.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_pod_manager.py
index b5c57c49167..f1cbc9be8d6 100644
---
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_pod_manager.py
+++
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_pod_manager.py
@@ -36,6 +36,7 @@ from airflow.providers.cncf.kubernetes.utils.pod_manager
import (
PodManager,
PodPhase,
XComRetrievalError,
+ detect_pod_terminate_early_issues,
log_pod_event,
parse_log_line,
)
@@ -145,6 +146,61 @@ def test_log_pod_event_multiple_events():
assert mock_pod_manager.log.info.call_count == 2
+def _pod_with_waiting_container(reason: str, message: str):
+ pod = mock.MagicMock()
+ container_status = mock.MagicMock()
+ waiting_state = mock.MagicMock()
+ waiting_state.reason = reason
+ waiting_state.message = message
+ container_status.state.waiting = waiting_state
+ pod.status.container_statuses = [container_status]
+ return pod
+
+
[email protected](
+ ("reason", "message"),
+ [
+ ("ErrImagePull", "rpc error: 502 Bad Gateway from auth.docker.io"),
+ ("ErrImagePull", "registry returned 503 Service Unavailable"),
+ ("ErrImagePull", "got 504 Gateway Timeout"),
+ # kubelet >= 1.32 appends the previous pull error to the
+ # ImagePullBackOff message, so the 5xx patterns still match.
+ ("ImagePullBackOff", 'Back-off pulling image "ubuntu:latest": 502 Bad
Gateway'),
+ ("ErrImagePull", "too many requests"),
+ ("ErrImagePull", "pull QPS exceeded"),
+ ],
+)
+def
test_detect_pod_terminate_early_issues_returns_none_for_transient_messages(reason,
message):
+ """Transient registry/network errors should not trigger early termination.
+
+ kubelet retries image pulls automatically; `startup_timeout` bounds the
+ total wait. Returning None here keeps the monitoring loop going so those
+ retries have a chance to succeed.
+ """
+ pod = _pod_with_waiting_container(reason, message)
+ assert detect_pod_terminate_early_issues(pod) is None
+
+
[email protected](
+ ("reason", "message"),
+ [
+ ("InvalidImageName", "Failed to apply default image tag"),
+ ("ErrImageNeverPull", "Container image pull policy is Never"),
+ ("ErrImagePull", "manifest unknown"),
+ ("ImagePullBackOff", "unauthorized: authentication required"),
+ # Bare ImagePullBackOff message (kubelet < 1.32) — must fail fast so
+ # a genuinely missing image doesn't wait out startup_timeout.
+ ("ImagePullBackOff", 'Back-off pulling image "nonexistent:latest"'),
+ ],
+)
+def
test_detect_pod_terminate_early_issues_returns_error_for_fatal_messages(reason,
message):
+ pod = _pod_with_waiting_container(reason, message)
+ error = detect_pod_terminate_early_issues(pod)
+ assert error is not None
+ assert reason in error
+ assert message in error
+
+
class TestPodManager:
def setup_method(self):
self.mock_kube_client = mock.Mock()
