lfg2025/archy
1
0
Fork 0
Code Issues 20 Pull Requests Actions Packages Projects Releases 61 Wiki Activity

fix(fleet): source Fleet view from trusted federation nodes

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
archipelago 2026-07-14 05:39:55 -04:00
parent 03418e5d26
commit 919665fb16
1 changed files with 111 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

167
core/archipelago/src/api/rpc/analytics.rs
View File

@ -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<serde_json::Value> {
let fleet_dir = self.config.data_dir.join("telemetry-fleet");
if !fleet_dir.exists() {
return Ok(serde_json::json!({ "nodes": [] }));
let mut nodes: Vec<serde_json::Value> = 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<u64>, total: Option<u64>| -> 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::<Vec<_>>(),
"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<serde_json::Value> = 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::<serde_json::Value>(&data) {
match tokio::fs::read_to_string(entry.path()).await {
Ok(data) => match serde_json::from_str::<serde_json::Value>(&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<String> {
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));
}
}