From 613f031a787f285988fcdc17c1970a9572dc08c2 Mon Sep 17 00:00:00 2001 From: Pal Lakatos-Toth Date: Fri, 26 Jun 2026 07:08:36 +0200 Subject: [PATCH] fix(controller): bound CRD watch timeout to stop reflector staleness on AKS (#451) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes #451 — on AKS, CR spec edits (InferencePolicy/ToolPolicy/EgressApproval/ …) intermittently stopped reaching the compiled ConfigMap until a controller restart. Root cause: every reconciler used `watcher::Config::default()`. That is NOT "no timeout" — kube-rs sends `timeoutSeconds=290` by default. The problem is the value: the Azure Standard LB idle timeout is 240s, so a quiet watch is silently dropped (no FIN/RST) at 240s, BEFORE the server's 290s close. kube-rs 3.1.0 then polls the dead stream with a bare `stream.next().await` (no client-side idle deadline) and hangs forever — the reflector store freezes while the 15s requeue keeps reconciling stale cached objects. Fix: a shared `watch_config::bounded()` that sets the watch timeout to 200s (< the 240s LB idle; kube-rs also rejects >=295s), applied to every primary and secondary (.owns/.watches) watch across all 11 CRD reconcilers so it can't diverge again. The server now closes each watch at 200s — that close is traffic that prevents the connection from ever sitting idle for 240s, and the watcher re-List/re-watches (refreshing the store) within a bounded window. Overridable via KARS_WATCH_TIMEOUT_SECS (clamped 30..=600) for other idle timeouts. This is the version-independent, low-risk prevention fix and works on the current kube-rs 3.1.0. Researched alternatives (documented in the module): - kube-rs 4.0 adds a client-side `next_with_idle_timeout` backstop (caps the freeze at ~295s instead of forever) but still ships NO HTTP/2 keepalive ping. - The architecturally superior, client-go-parity fix is HTTP/2 keepalive ping on the kube client (~45s dead-connection detection); it needs a custom hyper client stack + deps and is tracked as a defense-in-depth follow-up. Verification: clippy -D warnings clean, cargo fmt clean, 849 controller tests pass + 3 new watch_config unit tests (invariant: default < 240s; bounded() sets the timeout; env override clamped). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- controller/src/a2a_agent_reconciler.rs | 2 +- controller/src/egress_approval_reconciler.rs | 2 +- controller/src/inference_policy_reconciler.rs | 5 +- controller/src/kars_eval_reconciler.rs | 2 +- controller/src/kars_memory_reconciler.rs | 5 +- controller/src/kars_sre_action_reconciler.rs | 2 +- controller/src/main.rs | 1 + controller/src/mcp_server_reconciler.rs | 2 +- controller/src/pairing_reconciler.rs | 2 +- controller/src/reconciler/mod.rs | 4 +- controller/src/tool_policy_reconciler.rs | 5 +- controller/src/trust_graph_reconciler.rs | 2 +- controller/src/watch_config.rs | 111 ++++++++++++++++++ 13 files changed, 127 insertions(+), 18 deletions(-) create mode 100644 controller/src/watch_config.rs diff --git a/controller/src/a2a_agent_reconciler.rs b/controller/src/a2a_agent_reconciler.rs index a8fe7be5b..902843024 100644 --- a/controller/src/a2a_agent_reconciler.rs +++ b/controller/src/a2a_agent_reconciler.rs @@ -386,7 +386,7 @@ pub async fn run(client: Client) -> Result<()> { } } let ctx = Arc::new(Ctx { client }); - Controller::new(agents, kube::runtime::watcher::Config::default()) + Controller::new(agents, crate::watch_config::bounded()) .run( |x, ctx| async move { crate::metrics::observe_reconcile("A2AAgent", reconcile(x, ctx)).await diff --git a/controller/src/egress_approval_reconciler.rs b/controller/src/egress_approval_reconciler.rs index 2ffff1726..5472e631f 100644 --- a/controller/src/egress_approval_reconciler.rs +++ b/controller/src/egress_approval_reconciler.rs @@ -1007,7 +1007,7 @@ pub async fn run(client: Client) -> Result<()> { http, ttl_ceiling_seconds, }); - Controller::new(approvals, kube::runtime::watcher::Config::default()) + Controller::new(approvals, crate::watch_config::bounded()) .run( |x, ctx| async move { crate::metrics::observe_reconcile("EgressApproval", reconcile(x, ctx)).await diff --git a/controller/src/inference_policy_reconciler.rs b/controller/src/inference_policy_reconciler.rs index d502fcd9c..30e64feda 100644 --- a/controller/src/inference_policy_reconciler.rs +++ b/controller/src/inference_policy_reconciler.rs @@ -50,7 +50,6 @@ use kube::{ api::{Api, ListParams, ObjectMeta, Patch, PatchParams}, runtime::controller::{Action, Controller}, runtime::reflector::ObjectRef, - runtime::watcher, }; use serde_json::json; use std::collections::BTreeMap; @@ -838,10 +837,10 @@ pub async fn run(client: Client) -> Result<()> { // pure label-selector-only policies (no sandboxName + no // inferenceRef from sandbox) still rely on the periodic resync. let sandboxes: Api = Api::all(client); - Controller::new(policies, watcher::Config::default()) + Controller::new(policies, crate::watch_config::bounded()) .watches( sandboxes, - watcher::Config::default(), + crate::watch_config::bounded(), |sb: crate::crd::KarsSandbox| { let name = sb.spec.inference_ref.name.clone(); let ns = sb.namespace().unwrap_or_default(); diff --git a/controller/src/kars_eval_reconciler.rs b/controller/src/kars_eval_reconciler.rs index 5c4e26f79..11228c8cb 100644 --- a/controller/src/kars_eval_reconciler.rs +++ b/controller/src/kars_eval_reconciler.rs @@ -1310,7 +1310,7 @@ pub async fn run(client: Client) -> Result<()> { } } let ctx = Arc::new(Ctx { client }); - Controller::new(evals, kube::runtime::watcher::Config::default()) + Controller::new(evals, crate::watch_config::bounded()) .run( |x, ctx| async move { crate::metrics::observe_reconcile("KarsEval", reconcile(x, ctx)).await diff --git a/controller/src/kars_memory_reconciler.rs b/controller/src/kars_memory_reconciler.rs index 9fc53bcef..12fbe43fc 100644 --- a/controller/src/kars_memory_reconciler.rs +++ b/controller/src/kars_memory_reconciler.rs @@ -88,7 +88,6 @@ use kube::{ api::{Api, ListParams, ObjectMeta, Patch, PatchParams}, runtime::controller::{Action, Controller}, runtime::reflector::ObjectRef, - runtime::watcher, }; use serde_json::json; use std::collections::BTreeMap; @@ -876,10 +875,10 @@ pub async fn run(client: Client) -> Result<()> { // — without this we'd wait up to REQUEUE_OK (5 min) for the next // periodic resweep to flip NoSandboxesReferencing → RouterEnforcing. let sandboxes: Api = Api::all(client); - Controller::new(memories, watcher::Config::default()) + Controller::new(memories, crate::watch_config::bounded()) .watches( sandboxes, - watcher::Config::default(), + crate::watch_config::bounded(), |sb: crate::crd::KarsSandbox| { let ns = sb.namespace().unwrap_or_default(); sb.spec diff --git a/controller/src/kars_sre_action_reconciler.rs b/controller/src/kars_sre_action_reconciler.rs index 7156fbc13..d99e1a8fd 100644 --- a/controller/src/kars_sre_action_reconciler.rs +++ b/controller/src/kars_sre_action_reconciler.rs @@ -1035,7 +1035,7 @@ pub async fn run(client: Client) -> Result<()> { let api: Api = Api::all(client.clone()); let ctx = Arc::new(Ctx { client }); - Controller::new(api, kube::runtime::watcher::Config::default()) + Controller::new(api, crate::watch_config::bounded()) .run(reconcile, error_policy, ctx) .for_each(|res| async move { match res { diff --git a/controller/src/main.rs b/controller/src/main.rs index 78bd8139f..b1cf56516 100644 --- a/controller/src/main.rs +++ b/controller/src/main.rs @@ -66,6 +66,7 @@ mod tool_policy_reconciler; mod trust_graph; mod trust_graph_compile; mod trust_graph_reconciler; +mod watch_config; use anyhow::Result; use kube::Client; diff --git a/controller/src/mcp_server_reconciler.rs b/controller/src/mcp_server_reconciler.rs index 1c5705810..790756484 100644 --- a/controller/src/mcp_server_reconciler.rs +++ b/controller/src/mcp_server_reconciler.rs @@ -790,7 +790,7 @@ pub async fn run(client: Client) -> Result<()> { client: client.clone(), jwks_fetcher: Arc::new(HttpJwksFetcher::new()), }); - Controller::new(mcps, kube::runtime::watcher::Config::default()) + Controller::new(mcps, crate::watch_config::bounded()) .run( |x, ctx| async move { crate::metrics::observe_reconcile("McpServer", reconcile(x, ctx)).await diff --git a/controller/src/pairing_reconciler.rs b/controller/src/pairing_reconciler.rs index 6d20503be..3bb2b5712 100644 --- a/controller/src/pairing_reconciler.rs +++ b/controller/src/pairing_reconciler.rs @@ -159,7 +159,7 @@ pub async fn run(client: Client) -> Result<()> { let ctx = Arc::new(PairingContext { client }); - Controller::new(pairings, kube::runtime::watcher::Config::default()) + Controller::new(pairings, crate::watch_config::bounded()) .run( |x, ctx| async move { crate::metrics::observe_reconcile("KarsPairing", reconcile_pairing(x, ctx)).await diff --git a/controller/src/reconciler/mod.rs b/controller/src/reconciler/mod.rs index b6d8df06a..75c3926c0 100644 --- a/controller/src/reconciler/mod.rs +++ b/controller/src/reconciler/mod.rs @@ -3525,10 +3525,10 @@ pub async fn run(client: Client) -> Result<()> { agent_id_cache: Arc::new(crate::agent_id_provisioning::ProvisionerCache::new()), }); - Controller::new(sandboxes, kube::runtime::watcher::Config::default()) + Controller::new(sandboxes, crate::watch_config::bounded()) .watches( Api::::all(ctx.client.clone()), - kube::runtime::watcher::Config::default(), + crate::watch_config::bounded(), deployment_to_sandbox_ref, ) .run( diff --git a/controller/src/tool_policy_reconciler.rs b/controller/src/tool_policy_reconciler.rs index 08493713f..38a72cc8a 100644 --- a/controller/src/tool_policy_reconciler.rs +++ b/controller/src/tool_policy_reconciler.rs @@ -39,7 +39,6 @@ use kube::{ api::{Api, ListParams, ObjectMeta, Patch, PatchParams}, runtime::controller::{Action, Controller}, runtime::reflector::ObjectRef, - runtime::watcher, }; use serde_json::json; use std::collections::BTreeMap; @@ -765,10 +764,10 @@ pub async fn run(client: Client) -> Result<()> { // namespace that doesn't set toolPolicyRef explicitly implicitly // references it. let sandboxes: Api = Api::all(client); - Controller::new(tps, watcher::Config::default()) + Controller::new(tps, crate::watch_config::bounded()) .watches( sandboxes, - watcher::Config::default(), + crate::watch_config::bounded(), |sb: crate::crd::KarsSandbox| { let ns = sb.namespace().unwrap_or_default(); if ns.is_empty() { diff --git a/controller/src/trust_graph_reconciler.rs b/controller/src/trust_graph_reconciler.rs index dd31dab94..907876921 100644 --- a/controller/src/trust_graph_reconciler.rs +++ b/controller/src/trust_graph_reconciler.rs @@ -386,7 +386,7 @@ pub async fn run(client: Client) -> Result<()> { } } let ctx = Arc::new(Ctx { client }); - Controller::new(graphs, kube::runtime::watcher::Config::default()) + Controller::new(graphs, crate::watch_config::bounded()) .run( |x, ctx| async move { crate::metrics::observe_reconcile("TrustGraph", reconcile(x, ctx)).await diff --git a/controller/src/watch_config.rs b/controller/src/watch_config.rs new file mode 100644 index 000000000..dcc909ce3 --- /dev/null +++ b/controller/src/watch_config.rs @@ -0,0 +1,111 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! Shared watch configuration for every CRD `Controller`. +//! +//! ## The bug this fixes (issue #451) +//! +//! Every reconciler used `watcher::Config::default()`. That is **not** "no watch +//! timeout" — kube-rs always sends `timeoutSeconds=290` to the API server +//! (`WatchParams::populate_qp` defaults to 290 when `Config.timeout` is `None`). +//! The problem is the **value**: on AKS the controller-to-API-server path +//! traverses a load balancer / konnectivity tunnel with an idle timeout (Azure +//! Standard LB default: **4 minutes / 240s**). A watch with no events sends no +//! packets, so the LB **silently drops** the idle TCP connection (no FIN/RST) at +//! 240s — *before* the server's 290s close can be delivered. kube-rs 3.1.0 then +//! polls the dead stream with a bare `stream.next().await` (no client-side idle +//! deadline), which hangs **forever**. The reflector store freezes while the 15s +//! requeue keeps reconciling the stale cached objects, so CR spec edits never +//! reach the compiled ConfigMap until a controller restart forces a fresh `List`. +//! +//! (kube-rs 4.0 adds a client-side `next_with_idle_timeout` backstop that caps +//! the freeze at ~295s instead of forever, but still ships **no HTTP/2 keepalive +//! ping** — so the fix below remains valuable there too.) +//! +//! ## The fix +//! +//! Set the watch `timeout` **below** the environment's idle window. The API +//! server then closes each watch every `timeout` seconds; that close is *traffic* +//! on the connection, so it can never sit idle long enough to be dropped, and the +//! watcher re-`List`/re-`watch`es (refreshing the store) within a bounded window. +//! This *prevents* the silent drop, which is why the value must be +//! `< idle_timeout`: a value **above** it (e.g. the 290s default vs Azure's 240s) +//! does not help. +//! +//! Default: **200s** (comfortably under Azure's 240s LB default; kube-rs also +//! rejects `timeout >= 295`). Operators on a different idle timeout can override +//! via `KARS_WATCH_TIMEOUT_SECS`. +//! +//! This is the version-independent, low-risk primary fix. The architecturally +//! superior, client-go-parity defense (HTTP/2 keepalive ping on the kube client, +//! ~45s dead-connection detection) is tracked as a follow-up — it needs a custom +//! hyper client stack and additional dependencies. + +use kube::runtime::watcher; + +/// Env var to override the watch timeout (seconds). Must be below the +/// environment's LB / proxy idle timeout for the silent-drop prevention to work. +pub const WATCH_TIMEOUT_ENV: &str = "KARS_WATCH_TIMEOUT_SECS"; + +/// Default watch timeout in seconds — under Azure Standard LB's 240s idle +/// default, with headroom for jitter. +pub const DEFAULT_WATCH_TIMEOUT_SECS: u32 = 200; + +/// Resolve the configured watch timeout, honoring `KARS_WATCH_TIMEOUT_SECS`. +/// Values are clamped to a sane range (30s..=600s) so a typo can't disable the +/// protection or hammer the API server. +pub fn watch_timeout_secs() -> u32 { + std::env::var(WATCH_TIMEOUT_ENV) + .ok() + .and_then(|s| s.trim().parse::().ok()) + .map(|v| v.clamp(30, 600)) + .unwrap_or(DEFAULT_WATCH_TIMEOUT_SECS) +} + +/// A `watcher::Config` with a bounded, idle-drop-safe timeout. Use this for +/// **every** primary and secondary (`.owns()` / `.watches()`) watch instead of +/// `watcher::Config::default()` so a silently-dropped watch can never freeze the +/// reflector store (issue #451). +pub fn bounded() -> watcher::Config { + watcher::Config::default().timeout(watch_timeout_secs()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_is_under_azure_lb_idle_window() { + // The whole mechanism depends on the timeout firing BEFORE the 240s + // Azure LB idle drop. Guard the invariant. + const _: () = assert!( + DEFAULT_WATCH_TIMEOUT_SECS < 240, + "watch timeout must be below the 240s Azure LB idle default", + ); + const _FLOOR: () = assert!(DEFAULT_WATCH_TIMEOUT_SECS >= 30); + } + + #[test] + fn bounded_config_sets_the_timeout() { + let cfg = bounded(); + assert_eq!(cfg.timeout, Some(watch_timeout_secs())); + assert!(cfg.timeout.is_some(), "bounded() must set a watch timeout"); + } + + #[test] + fn env_override_is_clamped() { + // Below floor → clamped to 30. + unsafe { std::env::set_var(WATCH_TIMEOUT_ENV, "1") }; + assert_eq!(watch_timeout_secs(), 30); + // Above ceiling → clamped to 600. + unsafe { std::env::set_var(WATCH_TIMEOUT_ENV, "99999") }; + assert_eq!(watch_timeout_secs(), 600); + // In range → honored. + unsafe { std::env::set_var(WATCH_TIMEOUT_ENV, "150") }; + assert_eq!(watch_timeout_secs(), 150); + // Garbage → default. + unsafe { std::env::set_var(WATCH_TIMEOUT_ENV, "not-a-number") }; + assert_eq!(watch_timeout_secs(), DEFAULT_WATCH_TIMEOUT_SECS); + unsafe { std::env::remove_var(WATCH_TIMEOUT_ENV) }; + } +}