This is an automated email from the ASF dual-hosted git repository.
henry3260 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 5925de6d01e Fail fast when an init container image cannot be pulled
(#76393)
5925de6d01e is described below
commit 5925de6d01e6802616be2c7f0ea206d4cafe54f6
Author: Henry Chen <[email protected]>
AuthorDate: Sun Nov 8 19:34:28 2026 +0800
Fail fast when an init container image cannot be pulled (#76393)
* Fail fast when an init container image cannot be pulled
An unpullable init container image left the pod Pending until
startup_timeout expired, because the main container only reports
PodInitializing while init containers run. Deployments hit the full
120s wait for a typo in an init container image instead of an
immediate, actionable error.
* Name the stuck container in the image pull error
The raised error did not name the stuck container, and the failure
diagnostics omitted init containers and run behind a non-default flag, so
by default a reader had to work out the container from other output, such
as the kubelet message or the raw pod data.
* Keep a restarting sidecar from failing a pod that is already running
Native sidecars are init containers with ``restartPolicy: Always``, so
unlike
other init containers they restart, and can go back to waiting, long after
start-up. Scanning ``init_container_statuses`` unconditionally let a sidecar
that fails an image pull mid-run end the monitoring loop while the phase was
still ``Running``; without istio or ``do_xcom_push``, ``cleanup`` then
raised a
failure for a task whose work had already succeeded. The re-defer guard in
``trigger_reentry`` reads the same check, so a deferred task whose trigger
errored failed there instead of deferring again.
The pod's ``Initialized`` condition separates the two cases. An init
container
that cannot pull at start-up keeps it false, so failing fast still applies.
Once any main container has started, kubelet keeps it true even while a
sidecar restarts, so from then on only the main containers are checked.
---
.../providers/cncf/kubernetes/operators/pod.py | 5 +-
.../providers/cncf/kubernetes/utils/pod_manager.py | 41 ++++---
.../unit/cncf/kubernetes/operators/test_pod.py | 42 +++++++
.../unit/cncf/kubernetes/utils/test_pod_manager.py | 133 +++++++++++++++++----
4 files changed, 180 insertions(+), 41 deletions(-)
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
index f9716532765..65c57cafd24 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
@@ -1488,7 +1488,10 @@ class KubernetesPodOperator(BaseOperator):
pod_reason = getattr(remote_pod.status, "reason", None)
self.log.info("Pod phase: %s, reason: %s", pod_phase, pod_reason)
- container_statuses = getattr(remote_pod.status,
"container_statuses", None) or []
+ container_statuses = [
+ *(getattr(remote_pod.status, "init_container_statuses", None)
or []),
+ *(getattr(remote_pod.status, "container_statuses", None) or
[]),
+ ]
for status in container_statuses:
name = status.name
state = status.state
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 4b9c683f30e..8f18ec7792a 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
@@ -254,30 +254,39 @@ def detect_pod_terminate_early_issues(pod: V1Pod) -> str
| None:
FATAL_STATES = ["InvalidImageName", "ErrImageNeverPull"]
TRANSIENT_STATES = ["ErrImagePull", "ImagePullBackOff"]
- ERROR_MESSAGE = "Image cannot be pulled, unable to start:
{reason}\n{message}"
+ ERROR_MESSAGE = "Image for container '{name}' cannot be pulled, unable to
start: {reason}\n{message}"
pod_status = pod.status
- if not pod_status.container_statuses:
- return None
-
- for container_status in pod_status.container_statuses:
+ # ``init_container_statuses`` also lists native sidecars -- init
containers with
+ # ``restartPolicy: Always`` -- which can fail a restart mid-run. Scan init
containers
+ # only until ``Initialized`` is true: one that cannot pull keeps it false,
and kubelet
+ # keeps it true once any main container has started, even while a sidecar
restarts.
+ initialized = any(
+ condition.type == "Initialized" and condition.status == "True"
+ for condition in (pod_status.conditions or [])
+ )
+ container_statuses = [
+ # Init containers first: while one of them is stuck pulling, the main
+ # container only reports ``PodInitializing``, which matches no state
below.
+ *([] if initialized else (pod_status.init_container_statuses or [])),
+ *(pod_status.container_statuses or []),
+ ]
+ for container_status in container_statuses:
container_state: V1ContainerState = container_status.state
container_waiting: V1ContainerStateWaiting | None =
container_state.waiting
if not container_waiting:
continue
- if container_waiting.reason in FATAL_STATES:
- return ERROR_MESSAGE.format(
- reason=container_waiting.reason,
message=container_waiting.message or ""
- )
+ reason = container_waiting.reason
+ message = container_waiting.message or ""
- if container_waiting.reason in TRANSIENT_STATES:
- message_lower = (container_waiting.message or "").lower()
- is_transient = any(pattern in message_lower for pattern in
TRANSIENT_ERROR_PATTERNS)
- if not is_transient:
- return ERROR_MESSAGE.format(
- reason=container_waiting.reason,
message=container_waiting.message or ""
- )
+ # The message can quote the image name, so a transient pattern can
match by
+ # accident; it only ever excuses a TRANSIENT_STATES reason.
+ if reason in FATAL_STATES or (
+ reason in TRANSIENT_STATES
+ and not any(pattern in message.lower() for pattern in
TRANSIENT_ERROR_PATTERNS)
+ ):
+ return ERROR_MESSAGE.format(name=container_status.name,
reason=reason, message=message)
return None
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py
index 5f3219001b4..a151c8ae718 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py
@@ -19,6 +19,7 @@ from __future__ import annotations
import asyncio
import datetime
import json
+import logging
import re
from contextlib import contextmanager, nullcontext
from typing import TYPE_CHECKING
@@ -4641,3 +4642,44 @@ def test_read_pod_events(mock_log, mock_pod_manager):
mock_event_error.reason,
mock_event_error.message,
)
+
+
+def _build_waiting_container_status(name: str, reason: str, message: str = "")
-> k8s.V1ContainerStatus:
+ return k8s.V1ContainerStatus(
+ name=name,
+ image="alpine",
+ image_id="1",
+ ready=False,
+ restart_count=0,
+
state=k8s.V1ContainerState(waiting=k8s.V1ContainerStateWaiting(reason=reason,
message=message)),
+ )
+
+
+@patch(
+
"airflow.providers.cncf.kubernetes.operators.pod.KubernetesPodOperator.pod_manager",
+ spec=PodManager,
+)
+@patch(
+
"airflow.providers.cncf.kubernetes.operators.pod.KubernetesPodOperator.log",
+ spec=logging.Logger,
+)
+def test_read_pod_container_states_names_the_stuck_init_container(mock_log,
mock_pod_manager):
+ mock_pod_manager.read_pod.return_value = k8s.V1Pod(
+ status=k8s.V1PodStatus(
+ phase=PodPhase.PENDING,
+ init_container_statuses=[
+ _build_waiting_container_status("init", "ImagePullBackOff",
"Back-off pulling image")
+ ],
+ container_statuses=[_build_waiting_container_status("base",
"PodInitializing")],
+ )
+ )
+
+ operator = KubernetesPodOperator(task_id="test-task")
+ operator._read_pod_container_states(k8s.V1Pod())
+
+ mock_log.warning.assert_any_call(
+ "Container '%s': state='WAITING', reason='%s', message='%s'",
+ "init",
+ "ImagePullBackOff",
+ "Back-off pulling image",
+ )
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 17b1a7c80f9..fdc4b057509 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
@@ -26,7 +26,19 @@ from unittest.mock import MagicMock, PropertyMock, patch
import pendulum
import pytest
import time_machine
-from kubernetes.client import CoreV1Event, CoreV1EventList, V1ObjectMeta,
V1ObjectReference, V1Pod
+from kubernetes.client import (
+ CoreV1Event,
+ CoreV1EventList,
+ V1ContainerState,
+ V1ContainerStateTerminated,
+ V1ContainerStateWaiting,
+ V1ContainerStatus,
+ V1ObjectMeta,
+ V1ObjectReference,
+ V1Pod,
+ V1PodCondition,
+ V1PodStatus,
+)
from kubernetes.client.rest import ApiException
from urllib3.exceptions import HTTPError as BaseHTTPError
@@ -194,15 +206,62 @@ 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
+def _build_waiting_container_status(name: str, reason: str, message: str = "")
-> V1ContainerStatus:
+ return V1ContainerStatus(
+ name=name,
+ image="alpine",
+ image_id="1",
+ ready=False,
+ restart_count=0,
+ state=V1ContainerState(waiting=V1ContainerStateWaiting(reason=reason,
message=message)),
+ )
+
+
+def _build_pod_with_waiting_container(reason: str, message: str, *, init: bool
= False) -> V1Pod:
+ stuck = _build_waiting_container_status("init" if init else "base",
reason, message)
+ if not init:
+ return V1Pod(status=V1PodStatus(container_statuses=[stuck]))
+ return V1Pod(
+ status=V1PodStatus(
+ init_container_statuses=[stuck],
+ container_statuses=[_build_waiting_container_status("base",
"PodInitializing")],
+ )
+ )
+
+
+def _build_terminated_container_status(name: str) -> V1ContainerStatus:
+ return V1ContainerStatus(
+ name=name,
+ image="alpine",
+ image_id="1",
+ ready=False,
+ restart_count=0,
+
state=V1ContainerState(terminated=V1ContainerStateTerminated(exit_code=0)),
+ )
+
+
+def _build_pod_with_sidecar_pull_failure(*, initialized: bool) -> V1Pod:
+ """Build a pod whose native sidecar cannot pull its image.
+
+ With ``initialized=True`` the base container has already run, which keeps
+ ``Initialized`` true while the sidecar restarts.
+ """
+ return V1Pod(
+ status=V1PodStatus(
+ phase=PodPhase.RUNNING if initialized else PodPhase.PENDING,
+ conditions=[V1PodCondition(type="Initialized", status="True" if
initialized else "False")],
+ init_container_statuses=[
+ _build_waiting_container_status(
+ "sidecar", "ImagePullBackOff", 'Back-off pulling image
"sidecar:latest"'
+ )
+ ],
+ container_statuses=[
+ _build_terminated_container_status("base")
+ if initialized
+ else _build_waiting_container_status("base", "PodInitializing")
+ ],
+ )
+ )
@pytest.mark.parametrize(
@@ -225,14 +284,17 @@ def
test_detect_pod_terminate_early_issues_returns_none_for_transient_messages(r
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)
+ pod = _build_pod_with_waiting_container(reason, message)
assert detect_pod_terminate_early_issues(pod) is None
[email protected]("init", [False, True], ids=["main_container",
"init_container"])
@pytest.mark.parametrize(
("reason", "message"),
[
("InvalidImageName", "Failed to apply default image tag"),
+ # The image name contains the transient pattern "timeout".
+ ("InvalidImageName", 'Failed to apply default image tag
"timeout-worker:Bad Tag"'),
("ErrImageNeverPull", "Container image pull policy is Never"),
("ErrImagePull", "manifest unknown"),
("ImagePullBackOff", "unauthorized: authentication required"),
@@ -241,14 +303,38 @@ def
test_detect_pod_terminate_early_issues_returns_none_for_transient_messages(r
("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)
+def
test_detect_pod_terminate_early_issues_returns_error_for_fatal_messages(reason,
message, init):
+ pod = _build_pod_with_waiting_container(reason, message, init=init)
error = detect_pod_terminate_early_issues(pod)
assert error is not None
assert reason in error
assert message in error
[email protected](
+ ("initialized", "expected_error_contains"),
+ [
+ pytest.param(False, "Image for container 'sidecar' cannot be pulled",
id="still_initializing"),
+ pytest.param(True, None, id="already_initialized"),
+ ],
+)
+def test_detect_pod_terminate_early_issues_gates_sidecars_on_initialized(
+ initialized, expected_error_contains
+):
+ """A sidecar stuck pulling only justifies early termination while it gates
start-up.
+
+ Once the pod is ``Initialized`` init containers no longer gate start-up,
so a sidecar
+ failing a restart must not end the wait: the main container's work can
still succeed.
+ """
+ pod = _build_pod_with_sidecar_pull_failure(initialized=initialized)
+ error = detect_pod_terminate_early_issues(pod)
+ if expected_error_contains is None:
+ assert error is None
+ else:
+ assert error is not None
+ assert expected_error_contains in error
+
+
class TestPodManager:
def setup_method(self):
self.mock_kube_client = mock.Mock()
@@ -1028,11 +1114,13 @@ class TestPodManager:
waiting_state.reason = fail_reason
waiting_state.message = "Test error"
container_statuse.state.waiting = waiting_state
+ container_statuse.name = "base"
pod_response.status.container_statuses = [container_statuse]
self.mock_kube_client.read_namespaced_pod.return_value = pod_response
expected_msg = (
- f"Image cannot be pulled, unable to start:
{waiting_state.reason}\n{waiting_state.message}"
+ "Image for container 'base' cannot be pulled, unable to start: "
+ f"{waiting_state.reason}\n{waiting_state.message}"
)
mock_pod = MagicMock()
with pytest.raises(AirflowException, match=expected_msg):
@@ -1734,19 +1822,16 @@ class TestAsyncPodManager:
self.mock_async_hook.get_pod.assert_called()
@pytest.mark.asyncio
+ @pytest.mark.parametrize("init", [False, True], ids=["main_container",
"init_container"])
@pytest.mark.parametrize("fail_reason", ["ErrImagePull",
"ImagePullBackOff", "InvalidImageName"])
- async def test_start_pod_raises_fast_error_on_image_error(self,
fail_reason):
- pod_response = mock.MagicMock()
- pod_response.status.phase = "Pending"
- container_status = mock.MagicMock()
- waiting_state = mock.MagicMock()
- waiting_state.reason = fail_reason
- waiting_state.message = "Test error"
- container_status.state.waiting = waiting_state
- pod_response.status.container_statuses = [container_status]
+ async def test_start_pod_raises_fast_error_on_image_error(self,
fail_reason, init):
+ pod_response = _build_pod_with_waiting_container(fail_reason, "Test
error", init=init)
+ pod_response.status.phase = PodPhase.PENDING
self.mock_async_hook.get_pod.return_value = pod_response
+ stuck_container = "init" if init else "base"
expected_msg = (
- f"Image cannot be pulled, unable to start:
{waiting_state.reason}\n{waiting_state.message}"
+ f"Image for container '{stuck_container}' cannot be pulled, "
+ f"unable to start: {fail_reason}\nTest error"
)
mock_pod = mock.MagicMock()
with pytest.raises(AirflowException, match=expected_msg):
