shahar1 commented on code in PR #76276:
URL: https://github.com/apache/airflow/pull/76276#discussion_r4494413503
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -1292,9 +1313,29 @@ async def watch_pod_events(
last_rv = None
continue
- # Other API errors are either transient or
configuration/programming errors.
- # Re-raise so generic_api_retry can apply centralized
retry/backoff for
- # transient failures, and fail fast for non-retryable ones.
+ # One short-lived 5xx or dropped socket should not end event
logging for the
+ # rest of the pod's life. The budget matches
stop_after_attempt(API_RETRIES),
+ # which counts the first try, so API_RETRIES - 1 reconnects
follow it.
+ # get_pod carries its own @generic_api_retry, so a transient
error arriving
+ # from it has already been retried and signals a sustained
outage; letting it
+ # consume this budget too is what stops the watch instead of
looping forever.
+ if _should_retry_api(e) and transient_retries < API_RETRIES -
1:
Review Comment:
**minor: 429 ignores `Retry-After`**
`generic_api_retry` documents "Honors Retry-After on 429" through
`WaitRetryAfterOrExponential`, but this loop only uses the exponential part. On
this branch, a 429 with `Retry-After: 7` sleeps 1.0s. That is the one response
where the server says how long to back off.
Suggestion: move the header parsing out of `WaitRetryAfterOrExponential`
into a small helper that both paths use.
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -65,6 +72,13 @@
JOB_STATUS_CONDITION_TYPES = JOB_FINAL_STATUS_CONDITION_TYPES | {"Suspended"}
+# Hoisted out of the ``except`` clause in watch_pod_events: mypy rejects
star-unpacking a tuple
+# of exception types inside an except expression.
+WATCH_RETRYABLE_ERRORS: tuple[type[BaseException], ...] = (
Review Comment:
**major: real mid-stream drops are not retried**
> **Description doesn't match code**: PR description describes something
different from what the code actually does.
The description says a dropped socket surfaces as
`aiohttp.ClientConnectionError`. That holds only before response headers
arrive. Results from a real `kubernetes_asyncio` `Watch.stream` (aiohttp
3.14.5) against a local socket server:
| How the stream dies | Raised | Retried here? |
|---|---|---|
| Connection closed before response headers | `ServerDisconnectedError` (a
`ClientConnectionError`) | yes |
| FIN or RST after headers, mid-body | `ClientPayloadError` (a
`ClientError`, not a `ClientConnectionError`) | no |
| Half-open: server silent until the client timeout, before or after headers
| `TimeoutError` | no |
`Watch.next` re-raises `TimeoutError` when `timeout_seconds` is set, so it
does not reconnect on its own. On this branch, both exceptions propagate out of
`watch_pod_events` on the first occurrence. `AsyncPodManager.watch_pod_events`
then logs "Stopped watching" and event logging ends, the same as before this PR.
The new test raises a synthetic `ClientConnectionError("socket reset")`, so
it doesn't catch this.
Suggestion: treat both as transient. There are two ways:
- Add them to `TRANSIENT_CONNECTION_ERRORS`, whose comment already lists
"read timeout". This is the smallest change, but it also widens
`generic_api_retry` and the executor's handling.
- Keep it local to the watch: add them to this tuple, and extend the check
to `_should_retry_api(e) or isinstance(e, (ClientPayloadError, TimeoutError))`.
Either way, add both to the
`test_async_watch_pod_events_reconnects_on_transient_error` parametrization.
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -1292,9 +1313,29 @@ async def watch_pod_events(
last_rv = None
continue
- # Other API errors are either transient or
configuration/programming errors.
- # Re-raise so generic_api_retry can apply centralized
retry/backoff for
- # transient failures, and fail fast for non-retryable ones.
+ # One short-lived 5xx or dropped socket should not end event
logging for the
+ # rest of the pod's life. The budget matches
stop_after_attempt(API_RETRIES),
+ # which counts the first try, so API_RETRIES - 1 reconnects
follow it.
+ # get_pod carries its own @generic_api_retry, so a transient
error arriving
Review Comment:
**minor: a sustained outage now costs 25 reads per pod**
With the default `api_retries` of 5, each of the 5 outer attempts runs
`get_pod` with its own 5 attempts. By my reading (not measured), that is 25
`read_namespaced_pod` calls and about 90s of retrying per pod, where it was 5
before this PR.
The docstring on `AsyncPodManager.watch_pod_events` (from #75016) still says
"any error logs one warning and stops the watch, so a struggling API server is
not polled again for this pod". That no longer describes what happens, so it
needs updating at minimum.
Could a failure from `get_pod` stop the watch straight away? It has already
used its own retries, so the outer budget would then apply only to the watch
request, and an outage would stay at 5 reads.
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -1260,13 +1278,16 @@ async def watch_pod_events(
if event.metadata.resource_version:
last_rv = event.metadata.resource_version
+ # A delivered event means the connection works; don't
let unrelated
+ # blips hours apart add up to an exhausted budget.
+ transient_retries = 0
Review Comment:
**minor: the budget reset is untested and misses quiet watches**
> **Flag any changed or added behaviour without a corresponding test, and
flag tests that a reviewer cannot fail by reverting the PR's change.**
Deleting this line leaves all 13 `watch_pod_events` hook tests passing.
The reset also only fires on an event, so a watch that ends cleanly with no
events doesn't reset the budget. In a test on this branch, I repeated one 503
followed by three clean empty streams, and the fifth 503 raised. The impact is
small with the default 120s startup timeout, and grows with a long
`startup_timeout_seconds`.
Suggestion: also reset when a stream ends cleanly (an `else:` on the `try`),
and add a test for the reset.
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -1292,9 +1313,29 @@ async def watch_pod_events(
last_rv = None
continue
- # Other API errors are either transient or
configuration/programming errors.
- # Re-raise so generic_api_retry can apply centralized
retry/backoff for
- # transient failures, and fail fast for non-retryable ones.
+ # One short-lived 5xx or dropped socket should not end event
logging for the
+ # rest of the pod's life. The budget matches
stop_after_attempt(API_RETRIES),
Review Comment:
**nit: the watch covers pod startup, not the pod's life**
> **Description doesn't match code**: PR description describes something
different from what the code actually does.
Both callers (`KubernetesPodOperator.await_pod_start` and
`KubernetesPodTrigger._wait_for_pod_start`) cancel the events task once pod
start completes, and each stream lasts `startup_check_interval` (default 5s).
So "the rest of the pod's life" here, "hours apart" at L1281, and "held open
for hours" in the description overstate it. Worth correcting, since these
comments are where someone would learn how long the watch runs.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py:
##########
@@ -1635,6 +1639,138 @@ async def async_generator_with_error(*_, **__):
assert events[0] == mock_event
assert hook._event_polling_fallback is True
+ @staticmethod
+ def _failing_stream(exc):
+ """A ``Watch.stream`` replacement that raises on first iteration."""
+
+ async def stream(*_, **__):
+ raise exc
+ yield # pragma: no cover - makes this an async generator
+
+ return stream()
+
+ @staticmethod
+ def _event_stream(event):
+ async def stream(*_, **__):
+ yield {"object": event}
+
+ return stream()
+
+ def _watch_returning(self, mock_watch_class, streams):
+ mock_watch = mock.Mock(spec=["stream", "stop"])
+ mock_watch.stream = mock.Mock(side_effect=streams)
+ mock_watch_class.return_value = mock_watch
+ return mock_watch
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ "exception",
+ [
+ pytest.param(async_client.ApiException(status=503),
id="transient_status"),
+ pytest.param(ClientConnectionError("socket reset"),
id="dropped_connection"),
+ ],
+ )
+ @mock.patch(HOOK_MODULE + ".asyncio.sleep", new_callable=mock.AsyncMock)
+ @mock.patch(KUBE_ASYNC_HOOK.format("get_pod"))
+ @mock.patch("kubernetes_asyncio.watch.Watch")
+ async def test_async_watch_pod_events_reconnects_on_transient_error(
+ self, mock_watch_class, mock_get_pod, mock_sleep, exception,
kube_config_loader
+ ):
+ mock_pod = mock.MagicMock()
+ mock_pod.status.phase = "Running"
+ mock_get_pod.return_value = mock_pod
+
+ mock_event = mock.Mock()
+ mock_event.metadata.resource_version = "99"
+ mock_watch = self._watch_returning(
+ mock_watch_class,
+ [self._failing_stream(exception), self._event_stream(mock_event)],
+ )
+
+ hook = AsyncKubernetesHook(
+ conn_id=None,
+ in_cluster=False,
+ config_file=None,
+ cluster_context=None,
+ )
+
+ events = []
+ async for event in hook.watch_pod_events(name=POD_NAME,
namespace=NAMESPACE):
+ events.append(event)
+ break
+
+ assert events == [mock_event]
+ assert mock_watch.stream.call_count == 2
+ mock_sleep.assert_awaited_once_with(API_RETRY_WAIT_MIN)
+ # The failed attempt must not restart the stream from scratch.
Review Comment:
**minor: this assertion can't detect a restart from scratch**
> **Flag any changed or added behaviour without a corresponding test, and
flag tests that a reviewer cannot fail by reverting the PR's change.**
The failing stream raises before it yields, so `last_rv` is still `None`,
and `None` is also what a from-scratch restart sends. Resuming from the last
`resourceVersion` is the reason the description gives for retrying inside the
generator. The code does resume correctly: a stream that yields rv 99 and then
raises 503 reconnects with `"99"`. But no test pins that.
Suggestion: yield an event before raising, then assert the second call's
`resource_version` is `"99"`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
