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

Archipelago — open-source initial import

Browse Source
This commit is contained in:
Archipelago
2026-08-12 10:55:50 +00:00
commit b67e1527a2
2068 changed files with 472303 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/archipelago/src/monitoring/alerts.rs
+252
View File
@@ -0,0 +1,252 @@
use crate::monitoring::store::MetricsStore;
use crate::monitoring::types::*;
use tracing::{info, warn};
const ALERT_RULES_FILE: &str = "alert-rules.json";
impl AlertRule {
pub(crate) fn default_rules() -> Vec<AlertRule> {
vec![
AlertRule {
kind: AlertRuleKind::DiskUsage,
threshold: 80.0,
enabled: true,
description: "Disk usage exceeds threshold".to_string(),
},
AlertRule {
kind: AlertRuleKind::RamUsage,
threshold: 80.0,
enabled: true,
description: "Total memory usage exceeds threshold".to_string(),
},
AlertRule {
kind: AlertRuleKind::CpuLoad,
threshold: 4.0,
enabled: true,
description: "CPU load exceeds 4x core count for 5 minutes".to_string(),
},
AlertRule {
kind: AlertRuleKind::ContainerCrash,
threshold: 1.0,
enabled: true,
description: "Container stopped unexpectedly".to_string(),
},
AlertRule {
kind: AlertRuleKind::BackendErrorSpike,
threshold: 500.0,
enabled: true,
description: "RPC latency exceeds threshold (ms)".to_string(),
},
AlertRule {
kind: AlertRuleKind::SslCertExpiry,
threshold: 30.0,
enabled: true,
description: "SSL certificate expires within N days".to_string(),
},
]
}
}
/// Alert-related methods on MetricsStore.
impl MetricsStore {
/// Get the current alert rules.
pub async fn get_alert_rules(&self) -> Vec<AlertRule> {
self.alert_rules.read().await.clone()
}
/// Update an alert rule by kind and persist to disk.
pub async fn update_alert_rule(
&self,
kind: &AlertRuleKind,
enabled: Option<bool>,
threshold: Option<f64>,
) {
let mut rules = self.alert_rules.write().await;
if let Some(rule) = rules.iter_mut().find(|r| &r.kind == kind) {
if let Some(e) = enabled {
rule.enabled = e;
}
if let Some(t) = threshold {
rule.threshold = t;
}
}
// Persist to disk so changes survive restarts
if let Some(ref dir) = self.data_dir {
if let Err(e) = save_alert_rules(dir, &rules).await {
warn!("Failed to persist alert rules: {}", e);
}
}
}
/// Get fired alert history.
pub async fn get_fired_alerts(&self, last_n: usize) -> Vec<FiredAlert> {
let buf = self.fired_alerts.read().await;
let start = buf.len().saturating_sub(last_n);
buf.iter().skip(start).cloned().collect()
}
/// Acknowledge a fired alert by id.
pub async fn acknowledge_alert(&self, alert_id: &str) -> bool {
let mut buf = self.fired_alerts.write().await;
if let Some(alert) = buf.iter_mut().find(|a| a.id == alert_id) {
alert.acknowledged = true;
true
} else {
false
}
}
/// Evaluate alert rules against a snapshot and return any new alerts.
pub async fn check_alerts(&self, snapshot: &MetricSnapshot) -> Vec<FiredAlert> {
let rules = self.alert_rules.read().await;
let mut new_alerts = Vec::new();
let ts = snapshot.timestamp;
for rule in rules.iter() {
if !rule.enabled {
continue;
}
match rule.kind {
AlertRuleKind::DiskUsage => {
if snapshot.system.disk_total_bytes > 0 {
let pct = (snapshot.system.disk_used_bytes as f64
/ snapshot.system.disk_total_bytes as f64)
* 100.0;
if pct > rule.threshold {
new_alerts.push(FiredAlert {
id: format!("disk-{}", ts),
kind: AlertRuleKind::DiskUsage,
message: format!(
"Disk usage at {:.1}% (threshold: {:.0}%)",
pct, rule.threshold
),
value: pct,
threshold: rule.threshold,
timestamp: ts,
acknowledged: false,
});
}
}
}
AlertRuleKind::RamUsage => {
if snapshot.system.mem_total_bytes > 0 {
let pct = (snapshot.system.mem_used_bytes as f64
/ snapshot.system.mem_total_bytes as f64)
* 100.0;
if pct > rule.threshold {
new_alerts.push(FiredAlert {
id: format!("ram-{}", ts),
kind: AlertRuleKind::RamUsage,
message: format!(
"RAM usage at {:.1}% (threshold: {:.0}%)",
pct, rule.threshold
),
value: pct,
threshold: rule.threshold,
timestamp: ts,
acknowledged: false,
});
}
}
}
AlertRuleKind::CpuLoad => {
// Alert if 5-min load average exceeds threshold * core count
let cores = std::thread::available_parallelism()
.map(|n| n.get() as f64)
.unwrap_or(4.0);
let max_load = rule.threshold * cores;
if snapshot.system.load_avg_5 > max_load {
new_alerts.push(FiredAlert {
id: format!("cpu-{}", ts),
kind: AlertRuleKind::CpuLoad,
message: format!(
"CPU load at {:.1} (threshold: {:.0} = {:.0}x {} cores)",
snapshot.system.load_avg_5, max_load, rule.threshold, cores as u32
),
value: snapshot.system.load_avg_5,
threshold: max_load,
timestamp: ts,
acknowledged: false,
});
}
}
AlertRuleKind::BackendErrorSpike => {
if snapshot.rpc_latency_ms > rule.threshold {
new_alerts.push(FiredAlert {
id: format!("latency-{}", ts),
kind: AlertRuleKind::BackendErrorSpike,
message: format!(
"RPC latency at {:.0}ms (threshold: {:.0}ms)",
snapshot.rpc_latency_ms, rule.threshold
),
value: snapshot.rpc_latency_ms,
threshold: rule.threshold,
timestamp: ts,
acknowledged: false,
});
}
}
// ContainerCrash and SslCertExpiry are checked via dedicated paths
_ => {}
}
}
// Store fired alerts
if !new_alerts.is_empty() {
let mut buf = self.fired_alerts.write().await;
for alert in &new_alerts {
if buf.len() >= MAX_ALERT_HISTORY {
buf.pop_front();
}
buf.push_back(alert.clone());
}
}
new_alerts
}
}
/// Load alert rules from disk, falling back to defaults if file missing or corrupt.
pub(crate) async fn load_alert_rules(data_dir: &std::path::Path) -> Vec<AlertRule> {
let path = data_dir.join(ALERT_RULES_FILE);
match tokio::fs::read_to_string(&path).await {
Ok(content) => match serde_json::from_str::<Vec<AlertRule>>(&content) {
Ok(saved) => {
// Merge with defaults: use saved enabled/threshold, add any new rule kinds
let defaults = AlertRule::default_rules();
let mut merged = Vec::new();
for default in &defaults {
if let Some(saved_rule) = saved.iter().find(|r| r.kind == default.kind) {
merged.push(AlertRule {
kind: default.kind.clone(),
threshold: saved_rule.threshold,
enabled: saved_rule.enabled,
description: default.description.clone(),
});
} else {
merged.push(default.clone());
}
}
info!("Loaded alert rules from {}", path.display());
merged
}
Err(e) => {
warn!("Failed to parse alert rules ({}), using defaults", e);
AlertRule::default_rules()
}
},
Err(_) => AlertRule::default_rules(),
}
}
/// Save alert rules to disk.
pub(crate) async fn save_alert_rules(
data_dir: &std::path::Path,
rules: &[AlertRule],
) -> anyhow::Result<()> {
tokio::fs::create_dir_all(data_dir).await?;
let content = serde_json::to_string_pretty(rules)?;
tokio::fs::write(data_dir.join(ALERT_RULES_FILE), content).await?;
Ok(())
}
core/archipelago/src/monitoring/collector.rs
+393
View File
@@ -0,0 +1,393 @@
use super::{ContainerMetrics, MetricSnapshot, SystemMetrics};
use anyhow::{Context, Result};
/// Collect a full metrics snapshot from the system.
pub async fn collect_snapshot() -> Result<MetricSnapshot> {
let timestamp = chrono::Utc::now().timestamp();
let (cpu, mem, disk, net, load) = tokio::join!(
read_cpu_usage(),
read_meminfo(),
read_disk_usage(),
read_network_totals(),
read_loadavg(),
);
let cpu = cpu.unwrap_or(0.0);
let (mem_used, mem_total) = mem.unwrap_or((0, 0));
let (disk_used, disk_total) = disk.unwrap_or((0, 0));
let (net_rx, net_tx) = net.unwrap_or((0, 0));
let (l1, l5, l15) = load.unwrap_or((0.0, 0.0, 0.0));
let system = SystemMetrics {
cpu_percent: cpu,
mem_used_bytes: mem_used,
mem_total_bytes: mem_total,
disk_used_bytes: disk_used,
disk_total_bytes: disk_total,
net_rx_bytes: net_rx,
net_tx_bytes: net_tx,
load_avg_1: l1,
load_avg_5: l5,
load_avg_15: l15,
};
let containers = read_container_stats().await.unwrap_or_default();
Ok(MetricSnapshot {
timestamp,
system,
containers,
rpc_latency_ms: 0.0, // filled in by MetricsStore::push
ws_connections: 0, // filled in by MetricsStore::push
})
}
/// Compute CPU usage by sampling /proc/stat twice with a 250ms gap.
async fn read_cpu_usage() -> Result<f64> {
let snap1 = read_cpu_jiffies().await?;
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
let snap2 = read_cpu_jiffies().await?;
let total_delta = snap2.0.saturating_sub(snap1.0);
let idle_delta = snap2.1.saturating_sub(snap1.1);
if total_delta == 0 {
return Ok(0.0);
}
let usage = 100.0 * (1.0 - (idle_delta as f64 / total_delta as f64));
Ok((usage * 10.0).round() / 10.0)
}
/// Returns (total_jiffies, idle_jiffies) from /proc/stat.
async fn read_cpu_jiffies() -> Result<(u64, u64)> {
let content = tokio::fs::read_to_string("/proc/stat")
.await
.context("Failed to read /proc/stat")?;
let cpu_line = content
.lines()
.next()
.ok_or_else(|| anyhow::anyhow!("Empty /proc/stat"))?;
let vals: Vec<u64> = cpu_line
.split_whitespace()
.skip(1)
.filter_map(|v| v.parse().ok())
.collect();
if vals.len() < 4 {
anyhow::bail!("Not enough fields in /proc/stat cpu line");
}
let idle = vals[3];
let total: u64 = vals.iter().sum();
Ok((total, idle))
}
/// Read memory used/total from /proc/meminfo.
async fn read_meminfo() -> Result<(u64, u64)> {
let content = tokio::fs::read_to_string("/proc/meminfo")
.await
.context("Failed to read /proc/meminfo")?;
let mut total_kb: u64 = 0;
let mut available_kb: u64 = 0;
for line in content.lines() {
if let Some(val) = line.strip_prefix("MemTotal:") {
total_kb = parse_kb(val)?;
} else if let Some(val) = line.strip_prefix("MemAvailable:") {
available_kb = parse_kb(val)?;
}
}
Ok((
total_kb.saturating_sub(available_kb) * 1024,
total_kb * 1024,
))
}
fn parse_kb(val: &str) -> Result<u64> {
val.trim()
.trim_end_matches("kB")
.trim()
.parse::<u64>()
.context("parse meminfo kB value")
}
/// Read disk used/total via `df`, preferring the encrypted data partition.
async fn read_disk_usage() -> Result<(u64, u64)> {
let target = if std::path::Path::new("/var/lib/archipelago").exists() {
"/var/lib/archipelago"
} else {
"/"
};
let output = tokio::process::Command::new("df")
.args(["--block-size=1", "--output=used,size", target])
.output()
.await
.context("Failed to run df")?;
if !output.status.success() {
anyhow::bail!("df failed: {}", String::from_utf8_lossy(&output.stderr));
}
let stdout = String::from_utf8(output.stdout).context("df output not utf8")?;
let data_line = stdout
.lines()
.nth(1)
.ok_or_else(|| anyhow::anyhow!("No data line from df"))?;
let mut parts = data_line.split_whitespace();
let used: u64 = parts
.next()
.ok_or_else(|| anyhow::anyhow!("Missing used"))?
.parse()
.context("parse df used")?;
let total: u64 = parts
.next()
.ok_or_else(|| anyhow::anyhow!("Missing total"))?
.parse()
.context("parse df total")?;
Ok((used, total))
}
/// Read load averages from /proc/loadavg.
async fn read_loadavg() -> Result<(f64, f64, f64)> {
let content = tokio::fs::read_to_string("/proc/loadavg")
.await
.context("Failed to read /proc/loadavg")?;
let mut parts = content.split_whitespace();
let l1: f64 = parts
.next()
.ok_or_else(|| anyhow::anyhow!("Missing load1"))?
.parse()
.context("parse load1")?;
let l5: f64 = parts
.next()
.ok_or_else(|| anyhow::anyhow!("Missing load5"))?
.parse()
.context("parse load5")?;
let l15: f64 = parts
.next()
.ok_or_else(|| anyhow::anyhow!("Missing load15"))?
.parse()
.context("parse load15")?;
Ok((l1, l5, l15))
}
/// Read total network RX/TX bytes from /proc/net/dev (sum of all real interfaces).
async fn read_network_totals() -> Result<(u64, u64)> {
let content = tokio::fs::read_to_string("/proc/net/dev")
.await
.context("Failed to read /proc/net/dev")?;
let mut rx_total: u64 = 0;
let mut tx_total: u64 = 0;
for line in content.lines().skip(2) {
let line = line.trim();
if line.is_empty() {
continue;
}
// Split on colon to separate interface name from data
let (iface, data) = match line.split_once(':') {
Some((i, d)) => (i.trim(), d),
None => continue,
};
// Skip loopback
if iface == "lo" {
continue;
}
let parts: Vec<&str> = data.split_whitespace().collect();
// Fields: rx_bytes rx_packets rx_errs ... (8 fields) tx_bytes tx_packets ...
if parts.len() >= 10 {
if let Ok(rx) = parts[0].parse::<u64>() {
rx_total += rx;
}
if let Ok(tx) = parts[8].parse::<u64>() {
tx_total += tx;
}
}
}
Ok((rx_total, tx_total))
}
/// Get per-container resource stats via `podman stats --no-stream --format json`.
async fn read_container_stats() -> Result<Vec<ContainerMetrics>> {
let output = tokio::process::Command::new("podman")
.args(["stats", "--no-stream", "--format", "json"])
.output()
.await
.context("Failed to run podman stats")?;
if !output.status.success() {
anyhow::bail!(
"podman stats failed: {}",
String::from_utf8_lossy(&output.stderr)
);
}
let stdout = String::from_utf8_lossy(&output.stdout);
let entries: Vec<serde_json::Value> = serde_json::from_str(&stdout).unwrap_or_default();
Ok(entries
.iter()
.filter_map(|e| {
let name = e
.get("name")
.or_else(|| e.get("Name"))
.and_then(|v| v.as_str())?
.to_string();
Some(ContainerMetrics {
name,
cpu_percent: parse_percent_field(e, "cpu_percent")
.or_else(|| parse_percent_field(e, "CPUPerc"))
.unwrap_or(0.0),
mem_used_bytes: parse_bytes_field(e, "mem_usage")
.or_else(|| parse_bytes_field(e, "MemUsage"))
.unwrap_or(0),
mem_limit_bytes: parse_bytes_field(e, "mem_limit")
.or_else(|| parse_bytes_field(e, "MemLimit"))
.unwrap_or(0),
net_rx_bytes: parse_bytes_field(e, "net_input")
.or_else(|| parse_bytes_field(e, "NetInput"))
.unwrap_or(0),
net_tx_bytes: parse_bytes_field(e, "net_output")
.or_else(|| parse_bytes_field(e, "NetOutput"))
.unwrap_or(0),
block_read_bytes: parse_bytes_field(e, "block_input")
.or_else(|| parse_bytes_field(e, "BlockInput"))
.unwrap_or(0),
block_write_bytes: parse_bytes_field(e, "block_output")
.or_else(|| parse_bytes_field(e, "BlockOutput"))
.unwrap_or(0),
})
})
.collect())
}
/// Parse a percentage field that may be a number or a string like "12.5%".
fn parse_percent_field(obj: &serde_json::Value, key: &str) -> Option<f64> {
let val = obj.get(key)?;
if let Some(n) = val.as_f64() {
return Some(n);
}
val.as_str()?
.trim_end_matches('%')
.trim()
.parse::<f64>()
.ok()
}
/// Parse a bytes field that may be a number or a human-readable string.
fn parse_bytes_field(obj: &serde_json::Value, key: &str) -> Option<u64> {
let val = obj.get(key)?;
if let Some(n) = val.as_u64() {
return Some(n);
}
parse_human_bytes(val.as_str()?)
}
/// Parse human-readable byte strings like "1.5GiB", "256MiB", "100MiB / 16GiB".
fn parse_human_bytes(s: &str) -> Option<u64> {
let s = s.trim();
if s == "--" || s.is_empty() {
return Some(0);
}
// Handle "X / Y" format — take only the first part
let s = s.split('/').next()?.trim();
let (num_str, multiplier) = if let Some(n) = s.strip_suffix("GiB") {
(n, 1024u64 * 1024 * 1024)
} else if let Some(n) = s.strip_suffix("MiB") {
(n, 1024u64 * 1024)
} else if let Some(n) = s.strip_suffix("KiB") {
(n, 1024u64)
} else if let Some(n) = s.strip_suffix("GB") {
(n, 1_000_000_000u64)
} else if let Some(n) = s.strip_suffix("MB") {
(n, 1_000_000u64)
} else if let Some(n) = s.strip_suffix("kB") {
(n, 1000u64)
} else if let Some(n) = s.strip_suffix('B') {
(n, 1u64)
} else {
(s, 1u64)
};
let num: f64 = num_str.trim().parse().ok()?;
Some((num * multiplier as f64) as u64)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_parse_human_bytes_gib() {
assert_eq!(parse_human_bytes("1GiB"), Some(1073741824));
assert_eq!(parse_human_bytes("1.5GiB"), Some(1610612736));
}
#[test]
fn test_parse_human_bytes_mib() {
assert_eq!(parse_human_bytes("256MiB"), Some(268435456));
}
#[test]
fn test_parse_human_bytes_kib() {
assert_eq!(parse_human_bytes("1024KiB"), Some(1048576));
}
#[test]
fn test_parse_human_bytes_si() {
assert_eq!(parse_human_bytes("1000MB"), Some(1000000000));
assert_eq!(parse_human_bytes("100B"), Some(100));
}
#[test]
fn test_parse_human_bytes_empty() {
assert_eq!(parse_human_bytes("--"), Some(0));
assert_eq!(parse_human_bytes(""), Some(0));
}
#[test]
fn test_parse_human_bytes_slash_format() {
assert_eq!(parse_human_bytes("100MiB / 16GiB"), Some(104857600));
}
#[test]
fn test_parse_percent_string() {
let obj = serde_json::json!({"cpu": "12.5%"});
assert_eq!(parse_percent_field(&obj, "cpu"), Some(12.5));
}
#[test]
fn test_parse_percent_number() {
let obj = serde_json::json!({"cpu": 12.5});
assert_eq!(parse_percent_field(&obj, "cpu"), Some(12.5));
}
#[test]
fn test_parse_percent_missing() {
let obj = serde_json::json!({"other": 1});
assert_eq!(parse_percent_field(&obj, "cpu"), None);
}
#[test]
fn test_parse_bytes_field_number() {
let obj = serde_json::json!({"mem": 1048576});
assert_eq!(parse_bytes_field(&obj, "mem"), Some(1048576));
}
#[test]
fn test_parse_bytes_field_string() {
let obj = serde_json::json!({"mem": "256MiB"});
assert_eq!(parse_bytes_field(&obj, "mem"), Some(268435456));
}
}
core/archipelago/src/monitoring/mod.rs
+58
View File
@@ -0,0 +1,58 @@
pub(crate) mod alerts;
pub mod collector;
mod notifications;
pub mod store;
mod telemetry;
pub mod types;
// Re-export public types for external consumers
pub use store::MetricsStore;
pub use telemetry::spawn_telemetry_reporter;
pub use types::*;
use std::path::PathBuf;
use std::sync::Arc;
use tracing::{debug, warn};
/// Spawn the background metrics collector (runs every 300 seconds / 5 minutes).
/// Evaluates alert rules on each snapshot and dispatches notifications.
/// Note: health_monitor.rs handles container state polling at 120s intervals.
/// This collector handles system-level metrics (CPU, disk, network) and only
/// calls podman stats every 5 minutes to avoid duplicate subprocess overhead.
pub fn spawn_metrics_collector(
store: Arc<MetricsStore>,
state: Option<Arc<crate::state::StateManager>>,
data_dir: Option<PathBuf>,
) {
tokio::spawn(async move {
// Wait 60s for system to stabilize after boot
tokio::time::sleep(std::time::Duration::from_secs(60)).await;
let mut interval = tokio::time::interval(std::time::Duration::from_secs(300));
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
interval.tick().await;
match collector::collect_snapshot().await {
Ok(snapshot) => {
let alerts = store.check_alerts(&snapshot).await;
store.push(snapshot).await;
debug!("Metrics snapshot collected");
if !alerts.is_empty() {
if let Some(ref state_mgr) = state {
notifications::push_alert_notifications(state_mgr, &alerts).await;
}
if let Some(ref dir) = data_dir {
notifications::deliver_alert_webhooks(dir, &alerts).await;
}
}
}
Err(e) => {
warn!("Failed to collect metrics: {}", e);
}
}
}
});
}
core/archipelago/src/monitoring/notifications.rs
+138
View File
@@ -0,0 +1,138 @@
use crate::monitoring::types::{AlertRuleKind, FiredAlert};
use crate::webhooks::{self, WebhookEvent, WebhookPayload};
use chrono::Utc;
use std::collections::HashSet;
use std::path::Path;
use std::sync::Arc;
use tracing::info;
const NOTIFICATION_MAX_AGE_SECS: i64 = 30 * 60;
/// Push fired alerts as notifications to the state manager (broadcast via WebSocket).
pub(crate) async fn push_alert_notifications(
state_mgr: &Arc<crate::state::StateManager>,
alerts: &[FiredAlert],
) {
let (mut data, _rev) = state_mgr.get_snapshot().await;
prune_stale_alert_notifications(&mut data.notifications, alerts);
for alert in alerts {
let level = match alert.kind {
AlertRuleKind::DiskUsage | AlertRuleKind::RamUsage => {
if alert.value > 95.0 {
crate::data_model::NotificationLevel::Error
} else {
crate::data_model::NotificationLevel::Warning
}
}
AlertRuleKind::ContainerCrash => crate::data_model::NotificationLevel::Error,
_ => crate::data_model::NotificationLevel::Warning,
};
let notification = crate::data_model::Notification {
id: alert.id.clone(),
level,
title: format!("{:?} Alert", alert.kind),
message: alert.message.clone(),
timestamp: Utc::now().to_rfc3339(),
app_id: None,
};
data.notifications.push(notification);
}
// Keep max 20 notifications
while data.notifications.len() > 20 {
data.notifications.remove(0);
}
state_mgr.update_data(data).await;
info!("Fired {} alert(s)", alerts.len());
}
fn prune_stale_alert_notifications(
notifications: &mut Vec<crate::data_model::Notification>,
alerts: &[FiredAlert],
) {
let now = Utc::now();
let active_ids: HashSet<&str> = alerts.iter().map(|alert| alert.id.as_str()).collect();
notifications.retain(|notification| {
if active_ids.contains(notification.id.as_str()) {
return false;
}
if notification.app_id.is_some() || notification.id.starts_with("health-") {
return true;
}
match chrono::DateTime::parse_from_rfc3339(&notification.timestamp) {
Ok(ts) => {
now.signed_duration_since(ts.with_timezone(&Utc))
.num_seconds()
<= NOTIFICATION_MAX_AGE_SECS
}
Err(_) => false,
}
});
}
/// Deliver webhook notifications for alerts that map to webhook events.
pub(crate) async fn deliver_alert_webhooks(data_dir: &Path, alerts: &[FiredAlert]) {
for alert in alerts {
let event = match alert.kind {
AlertRuleKind::DiskUsage => Some(WebhookEvent::DiskWarning),
AlertRuleKind::ContainerCrash => Some(WebhookEvent::ContainerCrash),
_ => None,
};
if let Some(event) = event {
let payload = WebhookPayload {
event,
title: format!("{:?} Alert", alert.kind),
message: alert.message.clone(),
timestamp: Utc::now().to_rfc3339(),
node_id: String::new(),
details: Some(serde_json::json!({
"value": alert.value,
"threshold": alert.threshold,
})),
};
webhooks::send_webhook(data_dir, payload).await;
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::data_model::{Notification, NotificationLevel};
fn notification(id: &str, timestamp: String, app_id: Option<&str>) -> Notification {
Notification {
id: id.to_string(),
level: NotificationLevel::Warning,
title: "DiskUsage Alert".to_string(),
message: "Disk warning".to_string(),
timestamp,
app_id: app_id.map(str::to_string),
}
}
#[test]
fn prune_stale_alert_notifications_removes_duplicate_and_old_generic_alerts() {
let active_alert = FiredAlert {
id: "alert-active".to_string(),
kind: AlertRuleKind::DiskUsage,
message: "Disk warning".to_string(),
value: 90.0,
threshold: 85.0,
timestamp: Utc::now().timestamp(),
acknowledged: false,
};
let old_timestamp = (Utc::now() - chrono::Duration::minutes(45)).to_rfc3339();
let fresh_timestamp = (Utc::now() - chrono::Duration::minutes(5)).to_rfc3339();
let mut notifications = vec![
notification("alert-active", fresh_timestamp.clone(), None),
notification("alert-old", old_timestamp, None),
notification("alert-fresh", fresh_timestamp.clone(), None),
notification("health-indeedhub-1", fresh_timestamp, Some("indeedhub")),
];
prune_stale_alert_notifications(&mut notifications, &[active_alert]);
let ids: Vec<&str> = notifications.iter().map(|n| n.id.as_str()).collect();
assert_eq!(ids, vec!["alert-fresh", "health-indeedhub-1"]);
}
}
core/archipelago/src/monitoring/store.rs
+322
View File
@@ -0,0 +1,322 @@
use crate::monitoring::alerts::load_alert_rules;
use crate::monitoring::types::*;
use std::collections::VecDeque;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU32, Ordering};
use tokio::sync::RwLock;
/// Thread-safe metrics store with ring buffers at two resolutions.
pub struct MetricsStore {
minute_data: RwLock<VecDeque<MetricSnapshot>>,
quarter_hour_data: RwLock<VecDeque<MetricSnapshot>>,
minute_count: RwLock<u32>,
rpc_latency: RwLock<(f64, u64)>,
ws_connections: AtomicU32,
pub(crate) alert_rules: RwLock<Vec<AlertRule>>,
pub(crate) fired_alerts: RwLock<VecDeque<FiredAlert>>,
pub(crate) data_dir: Option<PathBuf>,
}
impl MetricsStore {
#[cfg(test)]
pub fn new() -> Self {
Self {
minute_data: RwLock::new(VecDeque::with_capacity(MAX_1MIN_ENTRIES)),
quarter_hour_data: RwLock::new(VecDeque::with_capacity(MAX_15MIN_ENTRIES)),
minute_count: RwLock::new(0),
rpc_latency: RwLock::new((0.0, 0)),
ws_connections: AtomicU32::new(0),
alert_rules: RwLock::new(AlertRule::default_rules()),
fired_alerts: RwLock::new(VecDeque::with_capacity(MAX_ALERT_HISTORY)),
data_dir: None,
}
}
/// Create a MetricsStore that persists alert rules to disk.
pub async fn with_data_dir(data_dir: PathBuf) -> Self {
let rules = load_alert_rules(&data_dir).await;
Self {
minute_data: RwLock::new(VecDeque::with_capacity(MAX_1MIN_ENTRIES)),
quarter_hour_data: RwLock::new(VecDeque::with_capacity(MAX_15MIN_ENTRIES)),
minute_count: RwLock::new(0),
rpc_latency: RwLock::new((0.0, 0)),
ws_connections: AtomicU32::new(0),
alert_rules: RwLock::new(rules),
fired_alerts: RwLock::new(VecDeque::with_capacity(MAX_ALERT_HISTORY)),
data_dir: Some(data_dir),
}
}
/// Record a new metric snapshot (called every minute by collector).
pub async fn push(&self, mut snapshot: MetricSnapshot) {
// Fill in RPC latency from accumulated samples
{
let mut latency = self.rpc_latency.write().await;
if latency.1 > 0 {
snapshot.rpc_latency_ms = (latency.0 / latency.1 as f64 * 10.0).round() / 10.0;
*latency = (0.0, 0);
}
}
// Fill in current WS connection count
snapshot.ws_connections = self.ws_connections.load(Ordering::Relaxed);
// Push to 1-minute ring buffer
{
let mut buf = self.minute_data.write().await;
if buf.len() >= MAX_1MIN_ENTRIES {
buf.pop_front();
}
buf.push_back(snapshot.clone());
}
// Every 15 minutes, push to quarter-hour ring buffer
{
let mut count = self.minute_count.write().await;
*count += 1;
if *count >= 15 {
*count = 0;
let mut buf = self.quarter_hour_data.write().await;
if buf.len() >= MAX_15MIN_ENTRIES {
buf.pop_front();
}
buf.push_back(snapshot);
}
}
}
/// Record an RPC request latency sample (milliseconds).
pub async fn record_rpc_latency(&self, latency_ms: f64) {
let mut data = self.rpc_latency.write().await;
data.0 += latency_ms;
data.1 += 1;
}
/// Increment WebSocket connection count (called on connect).
pub fn increment_ws(&self) {
self.ws_connections.fetch_add(1, Ordering::Relaxed);
}
/// Decrement WebSocket connection count (called on disconnect).
pub fn decrement_ws(&self) {
// Use saturating semantics to avoid underflow
let _ = self
.ws_connections
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
if v > 0 {
Some(v - 1)
} else {
Some(0)
}
});
}
/// Get the latest snapshot.
pub async fn latest(&self) -> Option<MetricSnapshot> {
self.minute_data.read().await.back().cloned()
}
/// Get minute-resolution data for the last N minutes.
pub async fn history_minutes(&self, last_n: usize) -> Vec<MetricSnapshot> {
let buf = self.minute_data.read().await;
let start = buf.len().saturating_sub(last_n);
buf.iter().skip(start).cloned().collect()
}
/// Get quarter-hour-resolution data for the last N entries.
pub async fn history_quarter_hours(&self, last_n: usize) -> Vec<MetricSnapshot> {
let buf = self.quarter_hour_data.read().await;
let start = buf.len().saturating_sub(last_n);
buf.iter().skip(start).cloned().collect()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_metrics_store_new() {
let store = MetricsStore::new();
assert_eq!(store.ws_connections.load(Ordering::Relaxed), 0);
}
#[test]
fn test_ws_connection_tracking() {
let store = MetricsStore::new();
store.increment_ws();
store.increment_ws();
assert_eq!(store.ws_connections.load(Ordering::Relaxed), 2);
store.decrement_ws();
assert_eq!(store.ws_connections.load(Ordering::Relaxed), 1);
store.decrement_ws();
assert_eq!(store.ws_connections.load(Ordering::Relaxed), 0);
// Decrement below zero should stay at 0
store.decrement_ws();
assert_eq!(store.ws_connections.load(Ordering::Relaxed), 0);
}
#[tokio::test]
async fn test_push_and_latest() {
let store = MetricsStore::new();
assert!(store.latest().await.is_none());
let snapshot = MetricSnapshot {
timestamp: 1000,
system: SystemMetrics {
cpu_percent: 25.0,
mem_used_bytes: 1_000_000,
mem_total_bytes: 4_000_000,
disk_used_bytes: 500_000,
disk_total_bytes: 1_000_000,
net_rx_bytes: 100,
net_tx_bytes: 200,
load_avg_1: 1.0,
load_avg_5: 0.5,
load_avg_15: 0.3,
},
containers: vec![],
rpc_latency_ms: 0.0,
ws_connections: 0,
};
store.push(snapshot).await;
let latest = store.latest().await.unwrap();
assert_eq!(latest.timestamp, 1000);
assert_eq!(latest.system.cpu_percent, 25.0);
}
#[tokio::test]
async fn test_rpc_latency_recording() {
let store = MetricsStore::new();
store.record_rpc_latency(10.0).await;
store.record_rpc_latency(20.0).await;
store.record_rpc_latency(30.0).await;
let snapshot = MetricSnapshot {
timestamp: 2000,
system: SystemMetrics {
cpu_percent: 0.0,
mem_used_bytes: 0,
mem_total_bytes: 0,
disk_used_bytes: 0,
disk_total_bytes: 0,
net_rx_bytes: 0,
net_tx_bytes: 0,
load_avg_1: 0.0,
load_avg_5: 0.0,
load_avg_15: 0.0,
},
containers: vec![],
rpc_latency_ms: 0.0,
ws_connections: 0,
};
store.push(snapshot).await;
let latest = store.latest().await.unwrap();
assert_eq!(latest.rpc_latency_ms, 20.0); // average of 10+20+30 = 20
}
#[tokio::test]
async fn test_history_minutes() {
let store = MetricsStore::new();
for i in 0..5 {
let snapshot = MetricSnapshot {
timestamp: i * 60,
system: SystemMetrics {
cpu_percent: i as f64,
mem_used_bytes: 0,
mem_total_bytes: 0,
disk_used_bytes: 0,
disk_total_bytes: 0,
net_rx_bytes: 0,
net_tx_bytes: 0,
load_avg_1: 0.0,
load_avg_5: 0.0,
load_avg_15: 0.0,
},
containers: vec![],
rpc_latency_ms: 0.0,
ws_connections: 0,
};
store.push(snapshot).await;
}
let history = store.history_minutes(3).await;
assert_eq!(history.len(), 3);
assert_eq!(history[0].timestamp, 120);
assert_eq!(history[2].timestamp, 240);
}
#[tokio::test]
async fn test_ring_buffer_eviction() {
let store = MetricsStore::new();
// Push more than MAX_1MIN_ENTRIES
for i in 0..(MAX_1MIN_ENTRIES + 10) {
let snapshot = MetricSnapshot {
timestamp: i as i64,
system: SystemMetrics {
cpu_percent: 0.0,
mem_used_bytes: 0,
mem_total_bytes: 0,
disk_used_bytes: 0,
disk_total_bytes: 0,
net_rx_bytes: 0,
net_tx_bytes: 0,
load_avg_1: 0.0,
load_avg_5: 0.0,
load_avg_15: 0.0,
},
containers: vec![],
rpc_latency_ms: 0.0,
ws_connections: 0,
};
store.push(snapshot).await;
}
let all = store.history_minutes(MAX_1MIN_ENTRIES + 100).await;
assert_eq!(all.len(), MAX_1MIN_ENTRIES);
// Oldest entry should be 10 (first 10 were evicted)
assert_eq!(all[0].timestamp, 10);
}
#[tokio::test]
async fn test_quarter_hour_downsampling() {
let store = MetricsStore::new();
// Push exactly 15 entries to trigger one quarter-hour sample
for i in 0..15 {
let snapshot = MetricSnapshot {
timestamp: i * 60,
system: SystemMetrics {
cpu_percent: 50.0,
mem_used_bytes: 0,
mem_total_bytes: 0,
disk_used_bytes: 0,
disk_total_bytes: 0,
net_rx_bytes: 0,
net_tx_bytes: 0,
load_avg_1: 0.0,
load_avg_5: 0.0,
load_avg_15: 0.0,
},
containers: vec![],
rpc_latency_ms: 0.0,
ws_connections: 0,
};
store.push(snapshot).await;
}
let qh = store.history_quarter_hours(10).await;
assert_eq!(qh.len(), 1);
assert_eq!(qh[0].timestamp, 14 * 60); // The 15th entry (index 14)
}
#[test]
fn test_constants() {
assert_eq!(MAX_1MIN_ENTRIES, 1440);
assert_eq!(MAX_15MIN_ENTRIES, 672);
}
}
core/archipelago/src/monitoring/telemetry.rs
+310
View File
@@ -0,0 +1,310 @@
use crate::monitoring::store::MetricsStore;
use std::path::PathBuf;
use std::sync::Arc;
use tracing::{debug, info, warn};
/// Spawn the periodic telemetry reporter (runs every 15 minutes when opt-in enabled).
/// Collects anonymous health data and saves to disk. Posts to central collector if configured.
pub fn spawn_telemetry_reporter(
store: Arc<MetricsStore>,
state: Option<Arc<crate::state::StateManager>>,
data_dir: PathBuf,
) {
tokio::spawn(async move {
// Wait 60s for system to fully stabilize
tokio::time::sleep(std::time::Duration::from_secs(60)).await;
let mut interval = tokio::time::interval(std::time::Duration::from_secs(900)); // 15 min
loop {
interval.tick().await;
// Check if telemetry is opted in
let config_path = data_dir.join("analytics-config.json");
let enabled = match tokio::fs::read_to_string(&config_path).await {
Ok(data) => serde_json::from_str::<serde_json::Value>(&data)
.ok()
.and_then(|c| c["enabled"].as_bool())
.unwrap_or(false),
Err(_) => false,
};
if !enabled {
debug!("Telemetry disabled — skipping report");
continue;
}
match build_telemetry_report(&store, &state, &data_dir).await {
Ok(report) => {
// Save latest report to disk
let report_path = data_dir.join("telemetry-latest.json");
if let Ok(json) = serde_json::to_string_pretty(&report) {
let _ = tokio::fs::write(&report_path, &json).await;
}
// Always save to local fleet directory so this node appears
// in its own fleet view
save_report_to_fleet_dir(&data_dir, &report).await;
// POST to central collector if configured
let collector_url = std::env::var("TELEMETRY_COLLECTOR_URL").ok();
if let Some(url) = collector_url {
match post_telemetry_report(&url, &report).await {
Ok(_) => info!("Telemetry report sent to collector"),
Err(e) => warn!("Failed to send telemetry report: {}", e),
}
} else {
debug!("Telemetry report saved locally (no collector URL configured)");
}
}
Err(e) => {
warn!("Failed to build telemetry report: {}", e);
}
}
}
});
}
/// Build an anonymous telemetry report from current system state.
async fn build_telemetry_report(
store: &Arc<MetricsStore>,
state: &Option<Arc<crate::state::StateManager>>,
data_dir: &std::path::Path,
) -> anyhow::Result<serde_json::Value> {
// Anonymous node ID — truncated SHA-256 hash of pubkey
let (node_id, node_name, version, container_count, running_count, peer_count, containers) =
if let Some(ref sm) = state {
let (data, _) = sm.get_snapshot().await;
let id = {
use sha2::{Digest, Sha256};
let mut h = Sha256::new();
h.update(data.server_info.pubkey.as_bytes());
hex::encode(h.finalize())[..16].to_string()
};
let containers: Vec<serde_json::Value> = data
.package_data
.iter()
.map(|(id, pkg)| {
serde_json::json!({
"id": id,
"state": format!("{:?}", pkg.state),
"version": pkg.manifest.version,
})
})
.collect();
let running = data
.package_data
.values()
.filter(|p| matches!(p.state, crate::data_model::PackageState::Running))
.count();
(
id,
data.server_info
.name
.clone()
.filter(|n| !n.trim().is_empty()),
data.server_info.version.clone(),
data.package_data.len(),
running,
data.peer_health.len(),
containers,
)
} else {
(
"unknown".to_string(),
None,
"unknown".to_string(),
0,
0,
0,
Vec::new(),
)
};
// System info
let cpu_cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(0);
let uptime_secs = tokio::fs::read_to_string("/proc/uptime")
.await
.ok()
.and_then(|s| s.split_whitespace().next()?.parse::<f64>().ok())
.map(|f| f as u64)
.unwrap_or(0);
let hostname = system_hostname().await;
let server_url = local_server_url(data_dir).await;
// Latest metrics snapshot
let latest = store.latest().await;
let (cpu_pct, mem_pct, disk_pct): (f64, f64, f64) = latest
.map(|s| {
let mem_total = s.system.mem_total_bytes as f64;
let disk_total = s.system.disk_total_bytes as f64;
(
s.system.cpu_percent,
if mem_total > 0.0 {
(s.system.mem_used_bytes as f64 / mem_total) * 100.0
} else {
0.0
},
if disk_total > 0.0 {
(s.system.disk_used_bytes as f64 / disk_total) * 100.0
} else {
0.0
},
)
})
.unwrap_or((0.0, 0.0, 0.0));
// Recent alerts
let recent_alerts: Vec<serde_json::Value> = store
.get_fired_alerts(10)
.await
.into_iter()
.map(|a| {
serde_json::json!({
"rule": format!("{:?}", a.kind),
"message": a.message,
"timestamp": a.timestamp,
})
})
.collect();
let _ = data_dir; // used for future per-app telemetry
Ok(serde_json::json!({
"node_id": node_id,
"node_name": node_name,
"hostname": hostname,
"server_url": server_url,
"version": version,
"uptime_secs": uptime_secs,
"cpu_cores": cpu_cores,
"cpu_pct": (cpu_pct * 10.0).round() / 10.0,
"mem_pct": (mem_pct * 10.0).round() / 10.0,
"disk_pct": (disk_pct * 10.0).round() / 10.0,
"containers": containers,
"container_count": container_count,
"running_count": running_count,
"federation_peers": peer_count,
"recent_alerts": recent_alerts,
"reported_at": chrono::Utc::now().to_rfc3339(),
}))
}
async fn system_hostname() -> Option<String> {
let output = tokio::process::Command::new("hostname")
.output()
.await
.ok()?;
if !output.status.success() {
return None;
}
let hostname = String::from_utf8_lossy(&output.stdout).trim().to_string();
(!hostname.is_empty()).then_some(hostname)
}
async fn local_server_url(data_dir: &std::path::Path) -> Option<String> {
let _ = data_dir;
let output = tokio::process::Command::new("hostname")
.arg("-I")
.output()
.await
.ok()?;
if !output.status.success() {
return None;
}
let ip = String::from_utf8_lossy(&output.stdout)
.split_whitespace()
.find(|ip| !ip.starts_with("127.") && ip.contains('.'))?
.to_string();
Some(format!("https://{ip}"))
}
/// POST a telemetry report to the central collector.
async fn post_telemetry_report(url: &str, report: &serde_json::Value) -> anyhow::Result<()> {
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(10))
.build()?;
let payload = serde_json::json!({
"method": "telemetry.ingest",
"params": report,
});
let response = client
.post(url)
.header("Content-Type", "application/json")
.header("User-Agent", "Archipelago-Telemetry/1.0")
.json(&payload)
.send()
.await?;
if !response.status().is_success() {
anyhow::bail!("Collector returned {}", response.status());
}
let status = response.status();
let body: serde_json::Value = response.json().await.unwrap_or_default();
if let Some(error) = body.get("error") {
anyhow::bail!("Collector RPC error: {}", error);
}
if body.get("result").is_none() {
anyhow::bail!("Collector returned {} without RPC result", status);
}
Ok(())
}
/// Save a telemetry report into the local fleet directory.
/// This makes the node's own report visible in the fleet dashboard.
async fn save_report_to_fleet_dir(data_dir: &std::path::Path, report: &serde_json::Value) {
let node_id = match report.get("node_id").and_then(|v| v.as_str()) {
Some(id) if !id.is_empty() => id,
_ => {
warn!("Telemetry report missing node_id — skipping fleet save");
return;
}
};
let fleet_dir = data_dir.join("telemetry-fleet");
if let Err(e) = tokio::fs::create_dir_all(&fleet_dir).await {
warn!("Failed to create telemetry-fleet directory: {}", e);
return;
}
// Write latest report (overwrites previous)
let latest_path = fleet_dir.join(format!("{}.json", node_id));
match serde_json::to_string_pretty(report) {
Ok(json) => {
if let Err(e) = tokio::fs::write(&latest_path, &json).await {
warn!("Failed to write fleet report for {}: {}", node_id, e);
}
}
Err(e) => {
warn!("Failed to serialize fleet report: {}", e);
return;
}
}
// Append to history file (cap at 200 entries)
let history_path = fleet_dir.join(format!("{}-history.json", node_id));
let mut history: Vec<serde_json::Value> = match tokio::fs::read_to_string(&history_path).await {
Ok(data) => serde_json::from_str(&data).unwrap_or_default(),
Err(_) => Vec::new(),
};
history.push(report.clone());
if history.len() > 200 {
let start = history.len() - 200;
history = history.split_off(start);
}
match serde_json::to_string_pretty(&history) {
Ok(json) => {
if let Err(e) = tokio::fs::write(&history_path, &json).await {
warn!("Failed to write fleet history for {}: {}", node_id, e);
}
}
Err(e) => {
warn!("Failed to serialize fleet history: {}", e);
}
}
debug!(
"Saved own telemetry report to fleet directory (node_id={})",
node_id
);
}
core/archipelago/src/monitoring/types.rs
+81
View File
@@ -0,0 +1,81 @@
use serde::{Deserialize, Serialize};
/// Maximum entries at 1-minute resolution (24 hours = 1440 minutes)
pub const MAX_1MIN_ENTRIES: usize = 1440;
/// Maximum entries at 15-minute resolution (7 days = 672 quarter-hours)
pub const MAX_15MIN_ENTRIES: usize = 672;
/// Maximum number of fired alerts to keep in history.
pub const MAX_ALERT_HISTORY: usize = 100;
/// A single metrics snapshot collected at a point in time.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MetricSnapshot {
pub timestamp: i64,
pub system: SystemMetrics,
pub containers: Vec<ContainerMetrics>,
pub rpc_latency_ms: f64,
pub ws_connections: u32,
}
/// System-wide resource metrics.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SystemMetrics {
pub cpu_percent: f64,
pub mem_used_bytes: u64,
pub mem_total_bytes: u64,
pub disk_used_bytes: u64,
pub disk_total_bytes: u64,
pub net_rx_bytes: u64,
pub net_tx_bytes: u64,
pub load_avg_1: f64,
pub load_avg_5: f64,
pub load_avg_15: f64,
}
/// Per-container resource metrics.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ContainerMetrics {
pub name: String,
pub cpu_percent: f64,
pub mem_used_bytes: u64,
pub mem_limit_bytes: u64,
pub net_rx_bytes: u64,
pub net_tx_bytes: u64,
pub block_read_bytes: u64,
pub block_write_bytes: u64,
}
/// Types of alert rules the system can evaluate.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum AlertRuleKind {
DiskUsage,
RamUsage,
CpuLoad,
ContainerCrash,
BackendErrorSpike,
SslCertExpiry,
}
/// A configured alert rule with threshold and enabled state.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AlertRule {
pub kind: AlertRuleKind,
pub threshold: f64,
pub enabled: bool,
pub description: String,
}
/// A fired alert instance.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FiredAlert {
pub id: String,
pub kind: AlertRuleKind,
pub message: String,
pub value: f64,
pub threshold: f64,
pub timestamp: i64,
pub acknowledged: bool,
}