Repository navigation
Conversation
The deferrable KubernetesPodOperator streams pod events from an async generator that was decorated with @generic_api_retry. tenacity only recognises coroutine functions, so it treated the async generator function as a plain synchronous callable and wrapped the construction of the generator rather than its iteration. Construction cannot fail, so the retry never fired: the first 5xx ended event logging for the rest of that pod's life. Connection-level failures were not caught at all, even though a dropped socket is the most likely way a watch stream kept open for hours ends. The retry has to live inside the generator rather than around it. Resuming a watch needs the last observed resourceVersion, which only the reconnect loop can see, and retrying from the outside would re-deliver events that had already been yielded. generic_api_retry now rejects async generator functions outright so the same mistake fails at import time instead of silently doing nothing. Dropping the decorator also exposed the generator's real signature to mypy, which found AsyncPodManager handing a user-settable float straight to the watch API's integer timeoutSeconds. The Kubernetes API server would reject a fractional value, and truncating it to 0 would ask for a watch with no timeout at all.
boring-cyborg
Bot
added
area:providers
provider:cncf-kubernetes
shahar1
left a comment
shahar1
left a comment
There was a problem hiding this comment.
Moving the retry inside the generator is the right fix, and failures before response headers now reconnect. But a stream that dies after headers raises ClientPayloadError or TimeoutError, and neither is retried, so the dropped-socket case the description leads with still ends the watch. That one is worth closing before merge; the rest are smaller.
The mechanism was checked by running a real kubernetes_asyncio watch against a local socket server, and the branch's own code in a scratch checkout. Details are inline.
Smaller observations
- nit, Testing: the new tests build the pod and event with
mock.MagicMock()/mock.Mock()and no spec. Rule: "Flag anymock.Mock()ormock.MagicMock()withoutspecorautospec. Unspec'd mocks silently accept any attribute access, hiding real bugs."
The other inline comments are four minors and one nit.
This review was drafted by an AI-assisted tool and
confirmed by an Apache Airflow maintainer. The findings
below are observations, not blockers; an Apache Airflow
maintainer — a real person — will take the next look at the
PR. If you think a finding is mis-applied, please reply on
the PR and a maintainer will weigh in.More on how Apache Airflow handles maintainer review:
Contributing guide.
|
|
||
| # 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], ...] = ( |
There was a problem hiding this 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 widensgeneric_api_retryand 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.
|
|
||
| # A delivered event means the connection works; don't let unrelated | ||
| # blips hours apart add up to an exhausted budget. | ||
| transient_retries = 0 |
There was a problem hiding this 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.
| # 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: |
There was a problem hiding this 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.
| # 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 |
There was a problem hiding this 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.
| # 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), |
There was a problem hiding this 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.
| 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. |
There was a problem hiding this 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".
Why
AsyncKubernetesHook.watch_pod_eventsis an async generator, and@generic_api_retryon it never retried anything. tenacity treats an async generator function as a plain sync callable, so it wrapped the construction of the generator instead of its iteration — and construction cannot fail.So the first 5xx from the events endpoint ended pod event logging for the rest of that pod's life. A dropped socket was worse: the handler only caught
ApiException, soaiohttp.ClientConnectionErrorwas not caught at all, even though that is how a watch stream held open for hours usually ends.The task still succeeds, because the watch is best-effort. What you lose is the pod events in the task log — the thing you read when a pod fails to start.
The retry has to be inside the generator: resuming a watch needs the last
resourceVersion, which only the reconnect loop can see, and retrying from outside would re-deliver events that were already yielded.What
hooks/kubernetes.py— drop the dead decorator and reconnect inside the existing loop instead, for the errors_should_retry_apialready accepts. Budget isAPI_RETRIES - 1reconnects with the same backoff bounds the decorator used, and it resets whenever an event arrives so unrelated blips hours apart don't accumulate. The handler now also catchesTRANSIENT_CONNECTION_ERRORS. 403, 404 and 410 behave as before, and a non-transient status still fails fast.kubernetes_helper_functions.py—generic_api_retrynow raisesTypeErroron an async generator function, so the same mistake fails at import time.watch_pod_eventswas its only async generator.utils/pod_manager.py— the dead decorator had hidden the generator's signature from mypy, which was masking a floatstartup_check_intervalbeing passed as the watch's integertimeoutSeconds. Now sends whole seconds, floored at 1 since 0 means no timeout.Was generative AI tooling used to co-author this PR?