shahar1 commented on code in PR #73399:
URL: https://github.com/apache/airflow/pull/73399#discussion_r4409346741
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -921,16 +922,12 @@ async def _load_config(self):
"""
Load Kubernetes configuration.
- For static auth (token, certificate), configuration is loaded once per
hook instance
- and cached. For exec-based auth (EKS, GKE), the config is reloaded on
every call so
- that short-lived tokens are always refreshed.
+ Reload credentials on every access while retaining the client for
non-exec auth.
Review Comment:
**minor** (question): this re-runs the full kubeconfig load on every
`get_conn()`. For `kube_config_path` and the default kubeconfig,
`kubernetes_asyncio.config.load_kube_config()` reads and YAML-parses the file
synchronously (`KubeConfigMerger.load_config`: `open()` + `yaml.safe_load`) on
the triggerer event loop, once per API call. The `kube_config` extra branch
also writes a new temp file per call. #60532 added the cache partly to avoid
repeated config loading during watch reconnects.
This PR is a performance change, so could you share before/after numbers
(for example, many pod triggers on a `kube_config_path` connection) showing the
SSL saving outweighs the per-call reload? If it does not, one option is to keep
the reload but only for what can rotate, for example re-reading when the
kubeconfig file's mtime changes.
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -1068,14 +1065,27 @@ async def _get_field(self, field_name):
@contextlib.asynccontextmanager
async def get_conn(self) -> AsyncGenerator[async_client.ApiClient, None]:
- kube_client = None
+ await self._load_config()
+ if self._client_cacheable:
+ # Reuse one client per hook: each construction runs
ssl.create_default_context()
+ # on the event loop and opens a new connection pool. Owners
release it via
+ # close(); triggers do so in cleanup().
+ if self._cached_kube_client is None:
+ self._cached_kube_client =
_TimeoutAsyncK8sApiClient(configuration=self.client_configuration)
+ yield self._cached_kube_client
+ return
+ # Exec-based auth is not cacheable: build and close a client per call.
+ kube_client =
_TimeoutAsyncK8sApiClient(configuration=self.client_configuration)
try:
- await self._load_config()
- kube_client =
_TimeoutAsyncK8sApiClient(configuration=self.client_configuration)
yield kube_client
finally:
- if kube_client is not None:
- await kube_client.close()
+ await kube_client.close()
+
+ async def close(self) -> None:
Review Comment:
**minor** — `providers/AGENTS.md`: "When a provider change needs a
user-visible note (typically a breaking change or important behavior change
that warrants explanation), update the provider's `docs/changelog.rst` directly
in the same PR, just below the `Changelog` header — exactly as the in-file
`NOTE TO CONTRIBUTORS` block describes."
`AsyncKubernetesHook` is a public hook. Before this PR, each `get_conn()`
closed its own client. Now a non-exec hook keeps the `aiohttp` session open
until `close()` is called. In-tree triggers handle that in `cleanup()`, but
third-party code using the hook directly will leave sessions open (`Unclosed
client session` warnings, held sockets). The PR description says direct users
must call `await hook.close()`; please put that in
`providers/cncf/kubernetes/docs/changelog.rst` and in the `AsyncKubernetesHook`
class docstring too.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py:
##########
@@ -911,13 +914,15 @@ async def
test_cleanup_deletes_pod_when_safe_to_cancel(self, mock_hook):
on_finish_action="delete_pod",
)
mock_hook.delete_pod = mock.AsyncMock()
+ mock_hook.close = mock.AsyncMock()
Review Comment:
**nit** — "Flag any `mock.Mock()` or `mock.MagicMock()` without `spec` or
`autospec`. Unspec'd mocks silently accept any attribute access, hiding real
bugs." The other tests in this PR use `spec=AsyncKubernetesHook.close`:
```suggestion
mock_hook.close = mock.AsyncMock(spec=AsyncKubernetesHook.close)
```
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py:
##########
@@ -1096,10 +1103,47 @@ async def
test_cleanup_skips_deletion_when_safe_to_cancel_raises(self, mock_hook
on_kill_action="delete_pod",
on_finish_action="delete_pod",
)
+ mock_hook.close = mock.AsyncMock(spec=AsyncKubernetesHook.close)
with mock.patch(
f"{TRIGGER_PATH}.safe_to_cancel",
new_callable=mock.AsyncMock,
side_effect=Exception("API down"),
):
await trigger.cleanup()
mock_hook.delete_pod.assert_not_called()
+ mock_hook.close.assert_awaited_once()
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ ("fired_event", "on_kill_action", "safe_result", "delete_error"),
+ [
+ pytest.param(True, "delete_pod", True, None, id="event-fired"),
+ pytest.param(False, "keep_pod", True, None, id="keep-pod"),
+ pytest.param(False, "delete_pod", False, None, id="restart"),
Review Comment:
**nit** — "The target is exactly 100% coverage of what the PR changes — no
more, no less." The `restart`, `state-error` and `pod-deleted` cases repeat the
`close.assert_awaited_once()` checks this PR also added to
`test_cleanup_does_not_delete_during_triggerer_restart`,
`test_cleanup_skips_deletion_when_safe_to_cancel_raises` and
`test_cleanup_deletes_pod_when_safe_to_cancel_and_closes_hook`. Those are
skipped on Airflow 3.3+, while this test patches `AIRFLOW_V_3_3_PLUS`, so this
is the better home. Consider dropping the `close` additions from the three
older tests.
--
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]
