Reuse async Kubernetes API client across trigger polls

[xvega]

I've been chasing triggerer issues on our production deployment: sustained triggers.blocked_main_thread bursts whenever a wave of deferrable KubernetesPodOperator/GKEStartPodOperator tasks started or finished, with the async thread blocked 2–10s at a time. #69661 helped the log-parsing path, but the metric barely moved. After reading the source and profiling the TriggerRunner with py-spy during blocking windows, the loop thread showed up inside ssl.create_default_context() reached from AsyncKubernetesHook.get_conn(): every API method builds a fresh ApiClient per call, which parses the CA bundle synchronously on the event loop and opens a new aiohttp session, a full TCP+TLS handshake per status poll (default 2s per trigger) with zero reuse. The GKE hook additionally fetched a new OAuth token and wrote the cluster CA to a temp file each call.

The change itself is simple: instead of building a new API client for every call, the hook now builds one and keeps it for its whole lifetime. For in-cluster and static-auth kubeconfigs Airflow loads the configuration once per hook, and anything that does rotate (the in-cluster service-account token) is read from the configuration per request rather than baked into the client, so one client can serve the hook's lifetime. Exec-based auth (EKS/GKE kubeconfigs) is the exception and keeps the old per-call behavior on purpose, reloading the config on every call is exactly what refreshes its short-lived credentials, so those clients can't be reused. To pair every cached client with an owner, hooks get an idempotent close() and the triggers call it from cleanup(), so the client lives exactly as long as its trigger. The GKE hook gets the same treatment plus token caching: it re-sets the bearer header every time the connection is used, so tokens refresh on schedule instead of being fetched from Google on every call (and since all that was left of its _load_config() was building the client, it's now called _build_client()).

A few things follow from the caching that are worth knowing. If you instantiate these hooks yourself or subclass the triggers and override cleanup(), call hook.close() when you're done, otherwise the client is only cleaned up at garbage collection and aiohttp logs an "unclosed session" warning; calling the hook again after close() just builds a fresh client. mTLS client certs from static kubeconfigs are now read once per hook rather than on every call, which I think is fine for hooks that live only as long as a trigger (if cert rotation mid-trigger ever becomes a real problem, rebuilding the client on auth errors would be the fix). The pre-existing gap left by #65212 the default-kubeconfig path sets _config_loaded without exec-auth detection, is untouched. And one expectation to set: each new trigger still builds its one client on the event loop, so blocked_main_thread should drop dramatically but won't be exactly zero when a batch of triggers starts.

To make sure this was really the problem, I reproduced it locally on a minikube cluster: real pod triggers polling real pods, with Airflow's own block_watchdog running, counting how many SSL contexts and API clients got created. I ran the exact same load twice, once with the old per-call get_conn patched back in as a baseline, once with this change. The baseline created a new client (and SSL context) for every single API call; with this change each trigger creates exactly one and releases it on cleanup. Unit tests in both providers assert the same behavior.

related: #69661

Was generative AI tooling used to co-author this PR?

No

[Miretpl]

Could you adjust your PR description to match our project guidelines - https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst?

[Miretpl]

I would appreciate a review from a Kubernetes Google maintainer on that part of the code.

[xvega]

Could you adjust your PR description to match our project guidelines - https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst?

Besides the AI disclosure block what's missing or not part of the guideline?

[Miretpl]

Besides the AI disclosure block what's missing or not part of the guideline?

Only this was missing here. Thanks for adding it. I've rerun the whole CI as there were a couple of unrelated failures. Let's see how it will be now.