This is an automated email from the ASF dual-hosted git repository.
shahar1 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 d9dbe635255 Report Kubernetes API 403 errors from the deferrable pod
trigger (#76081)
d9dbe635255 is described below
commit d9dbe63525537d718903d8f6ee67079af9269834
Author: Henry Chen <[email protected]>
AuthorDate: Sun Nov 1 02:57:27 2026 +0800
Report Kubernetes API 403 errors from the deferrable pod trigger (#76081)
AsyncKubernetesHook caught urllib3's HTTPError, but it talks to the
cluster through kubernetes_asyncio over aiohttp, whose ApiException
descends from OpenApiException and shares no ancestry with HTTPError.
None of those handlers could run, so a 403 reached KubernetesPodTrigger
as a bare ApiException and fell through to its generic handler: the task
failed with an opaque message instead of the guidance about adding the
triggerer's service account to pod-launcher-role.
Only 403 is translated. Every other status stays an ApiException so that
generic_api_retry keeps deciding which ones are transient — wrapping
them would hide the status from its retry predicate and turn retryable
503s into immediate task failures.
---
.../providers/cncf/kubernetes/hooks/kubernetes.py | 23 +++++----
.../unit/cncf/kubernetes/hooks/test_kubernetes.py | 57 ++++++++++++++++++++++
2 files changed, 70 insertions(+), 10 deletions(-)
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
index b4809db6cf3..1216fd74429 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
@@ -32,10 +32,9 @@ from kubernetes import client, config, utils, watch
from kubernetes.client.models import V1Deployment
from kubernetes.config import ConfigException
from kubernetes_asyncio import client as async_client, config as async_config,
watch as async_watch
-from urllib3.exceptions import HTTPError
from airflow.models import Connection
-from airflow.providers.cncf.kubernetes.exceptions import KubernetesApiError,
KubernetesApiPermissionError
+from airflow.providers.cncf.kubernetes.exceptions import
KubernetesApiPermissionError
from airflow.providers.cncf.kubernetes.kube_client import (
_disable_verify_ssl,
_TimeoutAsyncK8sApiClient,
@@ -1093,10 +1092,12 @@ class AsyncKubernetesHook(KubernetesHook):
namespace=namespace,
)
return pod
- except HTTPError as e:
- if hasattr(e, "status") and e.status == 403:
+ except async_client.ApiException as e:
+ if e.status == 403:
raise KubernetesApiPermissionError("Permission denied
(403) from Kubernetes API.") from e
- raise KubernetesApiError from e
+ # Other statuses stay ApiException: generic_api_retry owns the
decision about which
+ # ones are transient, and wrapping them here would hide the
status from its predicate.
+ raise
@generic_api_retry
async def delete_pod(self, name: str, namespace: str,
grace_period_seconds: int | None = None):
@@ -1161,8 +1162,10 @@ class AsyncKubernetesHook(KubernetesHook):
raw_bytes = await raw_resp.read()
# CPU-bound decode/split, offloaded so it can't block the
triggerer event loop.
return await asyncio.to_thread(_split_log_bytes, raw_bytes)
- except HTTPError as e:
- raise KubernetesApiError from e
+ except async_client.ApiException as e:
+ if e.status == 403:
+ raise KubernetesApiPermissionError("Permission denied
(403) from Kubernetes API.") from e
+ raise
@generic_api_retry
async def get_pod_events(
@@ -1188,10 +1191,10 @@ class AsyncKubernetesHook(KubernetesHook):
events: CoreV1EventList = await
v1_api.list_namespaced_event(**kwargs)
return events
- except HTTPError as e:
- if hasattr(e, "status") and e.status == 403:
+ except async_client.ApiException as e:
+ if e.status == 403:
raise KubernetesApiPermissionError("Permission denied
(403) from Kubernetes API.") from e
- raise KubernetesApiError from e
+ raise
@generic_api_retry
async def watch_pod_events(
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py
index 4fed8799e57..c8df520d0ee 100644
---
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py
+++
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py
@@ -37,6 +37,7 @@ from kubernetes.config import ConfigException
from kubernetes_asyncio import client as async_client
from airflow.models import Connection
+from airflow.providers.cncf.kubernetes.exceptions import
KubernetesApiPermissionError
from airflow.providers.cncf.kubernetes.hooks.kubernetes import (
AsyncKubernetesHook,
KubernetesHook,
@@ -2026,6 +2027,62 @@ class TestAsyncKubernetesHook:
assert logs == ["decoded line"]
mock_to_thread.assert_awaited_once_with(_split_log_bytes, raw_bytes)
+ # The async hook talks to kubernetes_asyncio, whose ApiException does not
derive from urllib3's
+ # HTTPError, so these statuses only reach the handlers through the async
exception type.
+ ASYNC_API_CALLS = [
+ pytest.param(
+ "get_pod", "read_namespaced_pod", {"name": POD_NAME, "namespace":
NAMESPACE}, id="get_pod"
+ ),
+ pytest.param(
+ "read_logs", "read_namespaced_pod_log", {"name": POD_NAME,
"namespace": NAMESPACE}, id="read_logs"
+ ),
+ pytest.param(
+ "get_pod_events",
+ "list_namespaced_event",
+ {"name": POD_NAME, "namespace": NAMESPACE},
+ id="get_pod_events",
+ ),
+ ]
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(("hook_method", "api_method", "call_kwargs"),
ASYNC_API_CALLS)
+ async def test_async_api_403_raises_permission_error(
+ self, hook_method, api_method, call_kwargs, kube_config_loader
+ ):
+ hook = AsyncKubernetesHook(
+ conn_id=None,
+ in_cluster=False,
+ config_file=None,
+ cluster_context=None,
+ )
+
+ with mock.patch(self.KUBE_API.format(api_method), autospec=True) as
lib_method:
+ lib_method.side_effect = async_client.ApiException(status=403,
reason="Forbidden")
+
+ with pytest.raises(KubernetesApiPermissionError, match="Permission
denied \\(403\\)"):
+ await getattr(hook, hook_method)(**call_kwargs)
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(("hook_method", "api_method", "call_kwargs"),
ASYNC_API_CALLS)
+ async def test_async_api_other_status_stays_api_exception(
+ self, hook_method, api_method, call_kwargs, kube_config_loader
+ ):
+ """Non-403 statuses must reach generic_api_retry unwrapped so it can
judge transience itself."""
+ hook = AsyncKubernetesHook(
+ conn_id=None,
+ in_cluster=False,
+ config_file=None,
+ cluster_context=None,
+ )
+
+ with mock.patch(self.KUBE_API.format(api_method), autospec=True) as
lib_method:
+ lib_method.side_effect = async_client.ApiException(status=404,
reason="Not Found")
+
+ with pytest.raises(async_client.ApiException) as exc_info:
+ await getattr(hook, hook_method)(**call_kwargs)
+
+ assert exc_info.value.status == 404
+
@pytest.mark.asyncio
@mock.patch(KUBE_BATCH_API.format("read_namespaced_job_status"))
async def test_get_job_status(self, lib_method, kube_config_loader):
