lfg2025/archy
1
0
Fork 0
Code Issues 27 Pull Requests 4 Actions Packages Projects Releases 74 Wiki Activity

backend: harden rootless app lifecycle orchestration

Browse Source
This commit is contained in:
archipelago
2026-06-11 00:24:32 -04:00
parent 09ec64932f
commit c393b96da3
56 changed files with 7543 additions and 1994 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/archipelago/src/monitoring/notifications.rs
+74 -2
View File
@@ -1,15 +1,20 @@
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 => {
@@ -27,7 +32,7 @@ pub(crate) async fn push_alert_notifications(
level,
title: format!("{:?} Alert", alert.kind),
message: alert.message.clone(),
timestamp: chrono::Utc::now().to_rfc3339(),
timestamp: Utc::now().to_rfc3339(),
app_id: None,
};
data.notifications.push(notification);
@@ -40,6 +45,30 @@ pub(crate) async fn push_alert_notifications(
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 {
@@ -53,7 +82,7 @@ pub(crate) async fn deliver_alert_webhooks(data_dir: &Path, alerts: &[FiredAlert
event,
title: format!("{:?} Alert", alert.kind),
message: alert.message.clone(),
timestamp: chrono::Utc::now().to_rfc3339(),
timestamp: Utc::now().to_rfc3339(),
node_id: String::new(),
details: Some(serde_json::json!({
"value": alert.value,
@@ -64,3 +93,46 @@ pub(crate) async fn deliver_alert_webhooks(data_dir: &Path, alerts: &[FiredAlert
}
}
}
#[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/telemetry.rs
+56 -24
View File
@@ -71,30 +71,49 @@ async fn build_telemetry_report(
data_dir: &std::path::Path,
) -> anyhow::Result<serde_json::Value> {
// Anonymous node ID — truncated SHA-256 hash of pubkey
let (node_id, version, container_count, running_count, peer_count) = 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 (node_id, 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.version.clone(),
data.package_data.len(),
running,
data.peer_health.len(),
containers,
)
} else {
(
"unknown".to_string(),
"unknown".to_string(),
0,
0,
0,
Vec::new(),
)
};
let running = data
.package_data
.values()
.filter(|p| matches!(p.state, crate::data_model::PackageState::Running))
.count();
(
id,
data.server_info.version.clone(),
data.package_data.len(),
running,
data.peer_health.len(),
)
} else {
("unknown".to_string(), "unknown".to_string(), 0, 0, 0)
};
// System info
let cpu_cores = std::thread::available_parallelism()
@@ -153,6 +172,7 @@ async fn build_telemetry_report(
"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,
@@ -166,16 +186,28 @@ async fn post_telemetry_report(url: &str, report: &serde_json::Value) -> anyhow:
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(report)
.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(())
}