From 919665fb16d2e0f19536e91250f2a957bc43a0c8 Mon Sep 17 00:00:00 2001 From: archipelago Date: Tue, 14 Jul 2026 05:39:55 -0400 Subject: [PATCH] fix(fleet): source Fleet view from trusted federation nodes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit telemetry.fleet-status only read the telemetry-fleet/ collector inbox — an opt-in anonymous-telemetry pipeline that federated peers never write to — so Fleet showed 'No nodes reporting' even with a healthy federation. Build the fleet list from nodes.json instead: every TRUSTED federated node is mapped from its last_state sync snapshot (cpu/mem/disk, uptime, apps, peer count) into the FleetNode shape the UI expects, keyed by DID. Observer ('peer') and untrusted nodes are excluded by design. Collector reports are still merged in for non-federated telemetry nodes. Co-Authored-By: Claude Fable 5 --- core/archipelago/src/api/rpc/analytics.rs | 167 ++++++++++++++-------- 1 file changed, 111 insertions(+), 56 deletions(-) diff --git a/core/archipelago/src/api/rpc/analytics.rs b/core/archipelago/src/api/rpc/analytics.rs index 4f0c5968..3532a812 100644 --- a/core/archipelago/src/api/rpc/analytics.rs +++ b/core/archipelago/src/api/rpc/analytics.rs @@ -326,75 +326,93 @@ impl RpcHandler { } /// Get all fleet nodes' latest reports. - /// Reads all {node_id}.json files from telemetry-fleet/ (excluding *-history.json). + /// + /// Primary source: TRUSTED federated nodes from nodes.json — their + /// `last_state` snapshot (kept fresh by federation state-sync) already + /// carries everything the Fleet UI renders. Observer ("peer") and + /// Untrusted nodes are deliberately excluded from Fleet. + /// + /// Secondary source: telemetry-fleet/*.json collector reports (opt-in + /// anonymous telemetry, includes this node's own report) — merged in for + /// back-compat with nodes that push telemetry but aren't federated. pub(super) async fn handle_telemetry_fleet_status(&self) -> Result { - let fleet_dir = self.config.data_dir.join("telemetry-fleet"); - if !fleet_dir.exists() { - return Ok(serde_json::json!({ "nodes": [] })); + let mut nodes: Vec = Vec::new(); + + // ── Trusted federation nodes ───────────────────────────────────── + let fed_nodes = crate::federation::load_nodes(&self.config.data_dir) + .await + .unwrap_or_default(); + for n in fed_nodes + .iter() + .filter(|n| n.trust_level == crate::federation::TrustLevel::Trusted) + { + let state = n.last_state.as_ref(); + let pct = |used: Option, total: Option| -> serde_json::Value { + match (used, total) { + (Some(u), Some(t)) if t > 0 => { + serde_json::json!((u as f64 / t as f64 * 100.0).round()) + } + _ => serde_json::json!(0), + } + }; + let apps = state.map(|s| s.apps.as_slice()).unwrap_or(&[]); + let reported_at = state + .map(|s| s.timestamp.clone()) + .or_else(|| n.last_seen.clone()) + .unwrap_or_else(|| n.added_at.clone()); + + let mut report = serde_json::json!({ + "node_id": n.did, + "node_name": state.and_then(|s| s.node_name.clone()).or_else(|| n.name.clone()), + "uptime_secs": state.and_then(|s| s.uptime_secs).unwrap_or(0), + "cpu_pct": state.and_then(|s| s.cpu_usage_percent).map(|v| v.round()).unwrap_or(0.0), + "mem_pct": pct(state.and_then(|s| s.mem_used_bytes), state.and_then(|s| s.mem_total_bytes)), + "disk_pct": pct(state.and_then(|s| s.disk_used_bytes), state.and_then(|s| s.disk_total_bytes)), + "container_count": apps.len(), + "running_count": apps.iter().filter(|a| a.status == "running").count(), + "federation_peers": state.map(|s| s.federated_peers.len()).unwrap_or(0), + "containers": apps.iter().map(|a| serde_json::json!({ + "id": a.id, + "state": a.status, + "version": a.version.clone().unwrap_or_default(), + })).collect::>(), + "reported_at": reported_at, + "trust_level": n.trust_level.to_string(), + "source": "federation", + }); + annotate_fleet_report(&mut report); + nodes.push(report); } - let mut nodes: Vec = Vec::new(); - let mut entries = tokio::fs::read_dir(&fleet_dir) - .await - .context("Failed to read telemetry-fleet directory")?; + // ── Opt-in telemetry collector reports ─────────────────────────── + let fleet_dir = self.config.data_dir.join("telemetry-fleet"); + if fleet_dir.exists() { + let mut entries = tokio::fs::read_dir(&fleet_dir) + .await + .context("Failed to read telemetry-fleet directory")?; - while let Some(entry) = entries.next_entry().await? { - let file_name = entry.file_name(); - let name = file_name.to_string_lossy(); - // Skip history files and non-JSON files - if name.ends_with("-history.json") || !name.ends_with(".json") { - continue; - } + while let Some(entry) = entries.next_entry().await? { + let file_name = entry.file_name(); + let name = file_name.to_string_lossy(); + // Skip history files and non-JSON files + if name.ends_with("-history.json") || !name.ends_with(".json") { + continue; + } - match tokio::fs::read_to_string(entry.path()).await { - Ok(data) => { - match serde_json::from_str::(&data) { + match tokio::fs::read_to_string(entry.path()).await { + Ok(data) => match serde_json::from_str::(&data) { Ok(mut report) => { - // Compute online/offline status from reported_at - let is_online = report - .get("reported_at") - .and_then(|v| v.as_str()) - .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) - .map(|dt| { - let age = chrono::Utc::now().signed_duration_since(dt); - age.num_minutes() < 30 - }) - .unwrap_or(false); - - // Compute human-readable last_seen - let last_seen = report - .get("reported_at") - .and_then(|v| v.as_str()) - .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) - .map(|dt| { - let age = chrono::Utc::now().signed_duration_since(dt); - let mins = age.num_minutes(); - if mins < 1 { - "just now".to_string() - } else if mins < 60 { - format!("{}m ago", mins) - } else if mins < 1440 { - format!("{}h ago", mins / 60) - } else { - format!("{}d ago", mins / 1440) - } - }) - .unwrap_or_else(|| "unknown".to_string()); - - if let Some(obj) = report.as_object_mut() { - obj.insert("online".to_string(), serde_json::json!(is_online)); - obj.insert("last_seen".to_string(), serde_json::json!(last_seen)); - } + annotate_fleet_report(&mut report); nodes.push(report); } Err(e) => { warn!(file = %name, error = %e, "Skipping corrupt fleet report"); } + }, + Err(e) => { + warn!(file = %name, error = %e, "Failed to read fleet report"); } } - Err(e) => { - warn!(file = %name, error = %e, "Failed to read fleet report"); - } } } @@ -531,3 +549,40 @@ fn local_server_url(host_ip: &str) -> Option { Some(format!("https://{host_ip}")) } } + +/// Stamp a fleet report with computed `online` and human-readable `last_seen` +/// derived from its `reported_at` timestamp (online = reported <30min ago). +fn annotate_fleet_report(report: &mut serde_json::Value) { + let reported = report + .get("reported_at") + .and_then(|v| v.as_str()) + .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()); + + let is_online = reported + .map(|dt| { + let age = chrono::Utc::now().signed_duration_since(dt); + age.num_minutes() < 30 + }) + .unwrap_or(false); + + let last_seen = reported + .map(|dt| { + let age = chrono::Utc::now().signed_duration_since(dt); + let mins = age.num_minutes(); + if mins < 1 { + "just now".to_string() + } else if mins < 60 { + format!("{}m ago", mins) + } else if mins < 1440 { + format!("{}h ago", mins / 60) + } else { + format!("{}d ago", mins / 1440) + } + }) + .unwrap_or_else(|| "unknown".to_string()); + + if let Some(obj) = report.as_object_mut() { + obj.insert("online".to_string(), serde_json::json!(is_online)); + obj.insert("last_seen".to_string(), serde_json::json!(last_seen)); + } +}