diff --git a/controller/src/a2a_agent_reconciler.rs b/controller/src/a2a_agent_reconciler.rs index a8fe7be5..90284302 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 2ffff172..5472e631 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 d502fcd9..30e64fed 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 5c4e26f7..11228c8c 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 9fc53bce..12fbe43f 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 7156fbc1..d99e1a8f 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 78bd8139..b1cf5651 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 1c570581..79075648 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 6d20503b..3bb2b571 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 b6d8df06..75c3926c 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 08493713..38a72cc8 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 dd31dab9..90787692 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 00000000..dcc909ce --- /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) }; + } +}