diff --git a/bin/fxt/src/tui/data_fetcher.rs b/bin/fxt/src/tui/data_fetcher.rs index 232ab76c0..f19665421 100644 --- a/bin/fxt/src/tui/data_fetcher.rs +++ b/bin/fxt/src/tui/data_fetcher.rs @@ -25,7 +25,7 @@ use crate::proto::{broker_gateway, data_acquisition, ml, monitoring, risk, tradi use super::state::{ AccountData, AlertData, CircuitBreakerData, ConnectionStatus, DataCacheData, DataFeedData, - ExecutionData, GpuData, ModelStatusData, OrderEventData, PodGroupData, PositionData, + ExecutionData, GpuData, ModelStatusData, OrderEventData, PodData, PodGroupData, PositionData, RiskAlertData, RiskData, ServiceData, SystemResources, TrainingSessionData, }; @@ -128,6 +128,7 @@ impl DataFetcher { self.spawn_risk_alerts(client); self.spawn_model_status(client); self.spawn_alerts(client); + self.spawn_pods_stream(client); } // ── Stream spawners ───────────────────────────────────────────── @@ -823,6 +824,69 @@ impl DataFetcher { } }); } + + /// Kubernetes pod statuses — streams from SubscribeClusterPods, grouped by + /// service for the pods cockpit. + fn spawn_pods_stream(&self, client: &FoxhuntClient) { + let mut mon = client.monitoring(); + let tx = self.tx.clone(); + let cancel = self.cancel.clone(); + let connected_sent = self.connected_sent.clone(); + + tokio::spawn(async move { + let mut backoff = Backoff::new(); + loop { + let req = Request::new(monitoring::SubscribeClusterPodsRequest { + interval_seconds: 5, + }); + match mon.subscribe_cluster_pods(req).await { + Ok(resp) => { + backoff.reset(); + let mut stream = resp.into_inner(); + loop { + tokio::select! { + _ = cancel.cancelled() => return, + msg = stream.message() => { + match msg { + Ok(Some(m)) => { + if !connected_sent.swap(true, Ordering::Relaxed) { + let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected)); + } + let groups = group_pods(m.pods); + let _ = tx.send(StateUpdate::Pods(groups)); + } + Ok(None) => { + debug!("ClusterPods stream ended"); + break; + } + Err(e) => { + warn!("ClusterPods stream error: {e}"); + break; + } + } + } + } + } + } + Err(e) => { + debug!("ClusterPods connect failed: {e}"); + let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Reconnecting { + attempt: backoff.attempts, + })); + } + } + + if backoff.attempts >= MAX_RECONNECT_ATTEMPTS { + warn!("ClusterPods exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up"); + return; + } + tokio::select! { + _ = cancel.cancelled() => return, + _ = tokio::time::sleep(backoff.next_delay()) => {} + } + } + }); + } } // --------------------------------------------------------------------------- @@ -1130,6 +1194,45 @@ fn convert_download_job(j: &data_acquisition::DownloadJobSummary) -> DataFeedDat } } +/// Group a flat list of [`PodInfo`] into [`PodGroupData`] buckets keyed by +/// service name. Pods with an empty service label are bucketed under `(other)`. +fn group_pods(pods: Vec) -> Vec { + use std::collections::BTreeMap; + + let mut groups: BTreeMap> = BTreeMap::new(); + for p in pods { + let svc = if p.service.is_empty() { + "(other)".into() + } else { + p.service.clone() + }; + groups.entry(svc).or_default().push(PodData { + name: p.name, + service: p.service, + status: p.status, + restarts: p.restarts, + age: p.age, + node: p.node, + ready: p.ready, + version: p.version, + }); + } + + groups + .into_iter() + .map(|(service, pods)| { + let ready_count = pods.iter().filter(|p| p.ready).count(); + let total_count = pods.len(); + PodGroupData { + service, + pods, + ready_count, + total_count, + } + }) + .collect() +} + /// Build aggregate [`DataCacheData`] from download job summaries. /// /// Uses completed job count as a proxy for total records (no per-record