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

fix: report collected fleet metrics and distinguish unavailable readings

Browse Source
This commit is contained in:
archipelago
2026-10-05 20:31:40 -04:00
parent cfd9a596c0
commit 041f1fa2d3
11 changed files with 338 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/archipelago/src/api/rpc/analytics.rs
+8 -7
View File
@@ -352,20 +352,19 @@ impl RpcHandler {
(Some(u), Some(t)) if t > 0 => {
serde_json::json!((u as f64 / t as f64 * 100.0).round())
}
_ => serde_json::json!(0),
_ => serde_json::Value::Null,
}
};
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());
.or_else(|| n.last_seen.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),
"uptime_secs": state.and_then(|s| s.uptime_secs),
"cpu_pct": state.and_then(|s| s.cpu_usage_percent).filter(|v| v.is_finite() && (0.0..=100.0).contains(v)).map(|v| v.round()),
"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(),
@@ -561,7 +560,7 @@ fn annotate_fleet_report(report: &mut serde_json::Value) {
let is_online = reported
.map(|dt| {
let age = chrono::Utc::now().signed_duration_since(dt);
age.num_minutes() < 30
age.num_seconds() >= -60 && age.num_seconds() < 1800
})
.unwrap_or(false);
@@ -569,7 +568,9 @@ fn annotate_fleet_report(report: &mut serde_json::Value) {
.map(|dt| {
let age = chrono::Utc::now().signed_duration_since(dt);
let mins = age.num_minutes();
if mins < 1 {
if age.num_seconds() < -60 {
"unknown (clock ahead)".to_string()
} else if mins < 1 {
"just now".to_string()
} else if mins < 60 {
format!("{}m ago", mins)
core/archipelago/src/api/rpc/federation/handlers.rs
+19 -6
View File
@@ -572,14 +572,27 @@ impl RpcHandler {
None
};
// Reuse the minute collector instead of running expensive probes for
// every peer. An absent/stalled collector is unknown, never zero load.
let now = chrono::Utc::now().timestamp();
let latest = self.metrics_store.latest().await.filter(|sample| {
(0..=180).contains(&now.saturating_sub(sample.timestamp))
});
let metrics = latest.as_ref().map(|sample| &sample.system);
let uptime = tokio::fs::read_to_string("/proc/uptime")
.await
.ok()
.and_then(|s| s.split_whitespace().next()?.parse::<f64>().ok())
.filter(|v| v.is_finite() && *v >= 0.0)
.map(|v| v as u64);
let state = federation::build_local_state(
apps,
0.0,
0,
0,
0,
0,
0,
metrics.map(|m| m.cpu_percent).filter(|v| v.is_finite() && (0.0..=100.0).contains(v)),
metrics.filter(|m| m.mem_total_bytes > 0).map(|m| m.mem_used_bytes),
metrics.filter(|m| m.mem_total_bytes > 0).map(|m| m.mem_total_bytes),
metrics.filter(|m| m.disk_total_bytes > 0).map(|m| m.disk_used_bytes),
metrics.filter(|m| m.disk_total_bytes > 0).map(|m| m.disk_total_bytes),
uptime,
tor_active,
server_name,
nostr_npub,
core/archipelago/src/api/rpc/federation/handshake_tests.rs
+74
View File
@@ -158,3 +158,77 @@ async fn managed_relay_receives_approval_rejection_and_cancellation() {
}
}
}
#[tokio::test]
async fn federation_metrics_are_collected_values_or_unknown_never_placeholders() {
for age in [None, Some(0), Some(181), Some(-120)] {
let tmp = tempfile::tempdir().unwrap();
let mut config = crate::config::Config::default();
config.data_dir = tmp.path().to_path_buf();
let metrics = Arc::new(crate::monitoring::MetricsStore::new());
if let Some(age) = age {
metrics
.push(
serde_json::from_value(serde_json::json!({
"timestamp": chrono::Utc::now().timestamp() - age,
"system": {"cpu_percent": 37.5, "mem_used_bytes": 200,
"mem_total_bytes": 800, "disk_used_bytes": 600,
"disk_total_bytes": 1000, "net_rx_bytes": 0, "net_tx_bytes": 0,
"load_avg_1": 0.0, "load_avg_5": 0.0, "load_avg_15": 0.0},
"containers": [], "rpc_latency_ms": 0.0, "ws_connections": 0
}))
.unwrap(),
)
.await;
}
let handler = crate::api::rpc::RpcHandler::new(
config,
Arc::new(crate::state::StateManager::new()),
metrics,
crate::session::SessionStore::new_for_tests(tmp.path().join("sessions.json")),
None,
None,
)
.await
.unwrap();
let snapshot = handler.handle_federation_get_state().await.unwrap();
if age == Some(0) {
assert_eq!(snapshot["cpu_usage_percent"], 37.5);
assert_eq!(snapshot["mem_used_bytes"], 200);
assert_eq!(snapshot["disk_total_bytes"], 1000);
} else {
for field in [
"cpu_usage_percent",
"mem_used_bytes",
"mem_total_bytes",
"disk_used_bytes",
"disk_total_bytes",
] {
assert!(
snapshot.get(field).is_none_or(|v| v.is_null()),
"{age:?}: {field}"
);
}
}
let peer = serde_json::from_value(serde_json::json!({
"did": "did:key:test", "pubkey": "11".repeat(32), "onion": "test.onion",
"trust_level": "trusted", "added_at": chrono::Utc::now().to_rfc3339(),
"last_state": snapshot
}))
.unwrap();
crate::federation::save_nodes(tmp.path(), &[peer])
.await
.unwrap();
let fleet = handler.handle_telemetry_fleet_status().await.unwrap();
let report = &fleet["nodes"][0];
if age == Some(0) {
assert_eq!(report["cpu_pct"], 38.0);
assert_eq!(report["mem_pct"], 25.0);
assert_eq!(report["disk_pct"], 60.0);
} else {
for field in ["cpu_pct", "mem_pct", "disk_pct"] {
assert!(report[field].is_null(), "{age:?}: {field}");
}
}
}
}
core/archipelago/src/federation/sync.rs
+24 -24
View File
@@ -230,12 +230,12 @@ async fn merge_transitive_peers(
#[allow(clippy::too_many_arguments)]
pub fn build_local_state(
apps: Vec<AppStatus>,
cpu: f64,
mem_used: u64,
mem_total: u64,
disk_used: u64,
disk_total: u64,
uptime: u64,
cpu: Option<f64>,
mem_used: Option<u64>,
mem_total: Option<u64>,
disk_used: Option<u64>,
disk_total: Option<u64>,
uptime: Option<u64>,
tor_active: bool,
server_name: Option<String>,
nostr_npub: Option<String>,
@@ -261,12 +261,12 @@ pub fn build_local_state(
timestamp: chrono::Utc::now().to_rfc3339(),
node_name: server_name,
apps,
cpu_usage_percent: Some(cpu),
mem_used_bytes: Some(mem_used),
mem_total_bytes: Some(mem_total),
disk_used_bytes: Some(disk_used),
disk_total_bytes: Some(disk_total),
uptime_secs: Some(uptime),
cpu_usage_percent: cpu,
mem_used_bytes: mem_used,
mem_total_bytes: mem_total,
disk_used_bytes: disk_used,
disk_total_bytes: disk_total,
uptime_secs: uptime,
tor_active: Some(tor_active),
nostr_npub,
own_fips_npub,
@@ -355,12 +355,12 @@ mod tests {
status: "running".to_string(),
version: Some("0.18".to_string()),
}],
25.5,
2_000_000_000,
8_000_000_000,
100_000_000_000,
500_000_000_000,
3600,
Some(25.5),
Some(2_000_000_000),
Some(8_000_000_000),
Some(100_000_000_000),
Some(500_000_000_000),
Some(3600),
true,
Some("Test Node".to_string()),
None,
@@ -430,12 +430,12 @@ mod tests {
];
let state = build_local_state(
vec![],
0.0,
0,
0,
0,
0,
0,
None,
None,
None,
None,
None,
None,
true,
None,
None,