feat(#1591): surface pending-login + crashing-agent banner warnings
This commit is contained in:
parent
5f57ea4ef1
commit
2157c3ae01
4 changed files with 188 additions and 1 deletions
|
|
@ -114,6 +114,14 @@ pub struct Coordinator {
|
||||||
/// watcher consults both this and the active map before declaring
|
/// watcher consults both this and the active map before declaring
|
||||||
/// a stop deliberate.
|
/// a stop deliberate.
|
||||||
recent_transient: Mutex<HashMap<String, (TransientKind, std::time::Instant)>>,
|
recent_transient: Mutex<HashMap<String, (TransientKind, std::time::Instant)>>,
|
||||||
|
/// Timestamps of recent unexpected container crashes, keyed by agent.
|
||||||
|
/// Fed by `crash_watch` each time it classifies a stop as a crash (so
|
||||||
|
/// a crash-looping container — which `Restart=on-failure` flips back
|
||||||
|
/// to running between polls — accumulates one entry per down-transition,
|
||||||
|
/// not just whatever its point-in-time state happens to be). Read by
|
||||||
|
/// the dashboard's `agents_crashing` banner warning via
|
||||||
|
/// `recent_crash_counts`, which prunes entries older than its window.
|
||||||
|
recent_crashes: Mutex<HashMap<String, Vec<std::time::Instant>>>,
|
||||||
/// Unified wire-facing event channel feeding the dashboard SSE
|
/// Unified wire-facing event channel feeding the dashboard SSE
|
||||||
/// stream. Carries broker messages (mirrored from `broker.subscribe`
|
/// stream. Carries broker messages (mirrored from `broker.subscribe`
|
||||||
/// by the forwarder task in `main.rs`) and dashboard-only mutation
|
/// by the forwarder task in `main.rs`) and dashboard-only mutation
|
||||||
|
|
@ -430,6 +438,7 @@ impl Coordinator {
|
||||||
agents: Mutex::new(HashMap::new()),
|
agents: Mutex::new(HashMap::new()),
|
||||||
transient: Mutex::new(HashMap::new()),
|
transient: Mutex::new(HashMap::new()),
|
||||||
recent_transient: Mutex::new(HashMap::new()),
|
recent_transient: Mutex::new(HashMap::new()),
|
||||||
|
recent_crashes: Mutex::new(HashMap::new()),
|
||||||
dashboard_events,
|
dashboard_events,
|
||||||
event_seq: AtomicU64::new(0),
|
event_seq: AtomicU64::new(0),
|
||||||
meta_updates_active: AtomicU64::new(0),
|
meta_updates_active: AtomicU64::new(0),
|
||||||
|
|
@ -1075,6 +1084,33 @@ impl Coordinator {
|
||||||
.collect()
|
.collect()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Record an unexpected crash for `agent`. Called by the crash
|
||||||
|
/// watcher whenever it classifies a container stop as a crash (not an
|
||||||
|
/// operator action). Append-only here; pruning happens lazily on read
|
||||||
|
/// in `recent_crash_counts`.
|
||||||
|
pub fn record_crash(&self, agent: &str) {
|
||||||
|
self.recent_crashes
|
||||||
|
.lock()
|
||||||
|
.unwrap()
|
||||||
|
.entry(agent.to_owned())
|
||||||
|
.or_default()
|
||||||
|
.push(std::time::Instant::now());
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Per-agent count of crashes within the last `window`. Lazily reaps
|
||||||
|
/// older timestamps and drops agents with none left, so the map stays
|
||||||
|
/// bounded and only lists agents actively crashing. Powers the
|
||||||
|
/// dashboard's `agents_crashing` banner warning.
|
||||||
|
pub fn recent_crash_counts(&self, window: std::time::Duration) -> HashMap<String, usize> {
|
||||||
|
let now = std::time::Instant::now();
|
||||||
|
let mut map = self.recent_crashes.lock().unwrap();
|
||||||
|
map.retain(|_, times| {
|
||||||
|
times.retain(|ts| now.duration_since(*ts) <= window);
|
||||||
|
!times.is_empty()
|
||||||
|
});
|
||||||
|
map.iter().map(|(k, v)| (k.clone(), v.len())).collect()
|
||||||
|
}
|
||||||
|
|
||||||
/// Set a transient state and return a guard that clears it on drop.
|
/// Set a transient state and return a guard that clears it on drop.
|
||||||
/// Use this from any path where the surrounding future could be
|
/// Use this from any path where the surrounding future could be
|
||||||
/// cancelled or panic between set and clear (HTTP handlers, spawned
|
/// cancelled or panic between set and clear (HTTP handlers, spawned
|
||||||
|
|
|
||||||
|
|
@ -95,6 +95,7 @@ fn emit_crash_transitions(coord: &Coordinator, prev: &HashSet<String>, current:
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
tracing::warn!(agent = %stopped, "container crash detected");
|
tracing::warn!(agent = %stopped, "container crash detected");
|
||||||
|
coord.record_crash(stopped);
|
||||||
coord.notify_manager(&hive_sh4re::HelperEvent::ContainerCrash {
|
coord.notify_manager(&hive_sh4re::HelperEvent::ContainerCrash {
|
||||||
agent: stopped.clone(),
|
agent: stopped.clone(),
|
||||||
note: Some("container stopped without an operator action".into()),
|
note: Some("container stopped without an operator action".into()),
|
||||||
|
|
|
||||||
|
|
@ -463,6 +463,13 @@ where
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Window over which container crashes count toward the `agents_crashing`
|
||||||
|
/// banner warning. Wide enough that a crash-looping container (restarted
|
||||||
|
/// by `Restart=on-failure` every few seconds) keeps the warning lit
|
||||||
|
/// between flaps, short enough that a single recovered crash clears within
|
||||||
|
/// minutes.
|
||||||
|
const CRASH_WARNING_WINDOW: std::time::Duration = std::time::Duration::from_mins(10);
|
||||||
|
|
||||||
async fn api_state(headers: HeaderMap, State(state): State<AppState>) -> axum::Json<StateSnapshot> {
|
async fn api_state(headers: HeaderMap, State(state): State<AppState>) -> axum::Json<StateSnapshot> {
|
||||||
let host = headers
|
let host = headers
|
||||||
.get("host")
|
.get("host")
|
||||||
|
|
@ -523,6 +530,18 @@ async fn api_state(headers: HeaderMap, State(state): State<AppState>) -> axum::J
|
||||||
.map(QuestionView::from_question)
|
.map(QuestionView::from_question)
|
||||||
.collect();
|
.collect();
|
||||||
|
|
||||||
|
// Banner warnings: host probes (disk) + agent-state (pending logins,
|
||||||
|
// crashing agents). Built before the response struct because the
|
||||||
|
// agent-state producer borrows `containers`, which moves in below.
|
||||||
|
let server_warnings = {
|
||||||
|
let mut w = crate::host_stats::server_warnings();
|
||||||
|
w.extend(crate::host_stats::agent_state_warnings(
|
||||||
|
&containers,
|
||||||
|
&state.coord.recent_crash_counts(CRASH_WARNING_WINDOW),
|
||||||
|
));
|
||||||
|
w
|
||||||
|
};
|
||||||
|
|
||||||
axum::Json(StateSnapshot {
|
axum::Json(StateSnapshot {
|
||||||
seq,
|
seq,
|
||||||
hostname,
|
hostname,
|
||||||
|
|
@ -564,7 +583,7 @@ async fn api_state(headers: HeaderMap, State(state): State<AppState>) -> axum::J
|
||||||
.ok()
|
.ok()
|
||||||
.filter(|s| !s.is_empty()),
|
.filter(|s| !s.is_empty()),
|
||||||
peer_hives: parse_peer_hives(),
|
peer_hives: parse_peer_hives(),
|
||||||
server_warnings: crate::host_stats::server_warnings(),
|
server_warnings,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -13,8 +13,12 @@
|
||||||
//! pressure, a failed unit, …) is a backend-only change — no frontend
|
//! pressure, a failed unit, …) is a backend-only change — no frontend
|
||||||
//! edit. Keep producers cheap; this runs on every `/api/state` assembly.
|
//! edit. Keep producers cheap; this runs on every `/api/state` assembly.
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
|
|
||||||
use serde::Serialize;
|
use serde::Serialize;
|
||||||
|
|
||||||
|
use crate::container_view::ContainerView;
|
||||||
|
|
||||||
/// One server-level warning for the dashboard's top-of-page banner.
|
/// One server-level warning for the dashboard's top-of-page banner.
|
||||||
#[derive(Debug, Clone, Serialize)]
|
#[derive(Debug, Clone, Serialize)]
|
||||||
pub struct ServerWarning {
|
pub struct ServerWarning {
|
||||||
|
|
@ -65,6 +69,68 @@ pub fn server_warnings() -> Vec<ServerWarning> {
|
||||||
out
|
out
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Agent-state warnings derived from the live container snapshot the
|
||||||
|
/// dashboard already holds: agents that need a claude login, and agents
|
||||||
|
/// that are crashing. Kept separate from [`server_warnings`] (host
|
||||||
|
/// probes) because the caller owns the container list + crash counts;
|
||||||
|
/// the dashboard concatenates both into one banner list.
|
||||||
|
///
|
||||||
|
/// `crash_counts` is `Coordinator::recent_crash_counts(window)` — agent →
|
||||||
|
/// number of crashes inside that window — so a crash-looping agent shows
|
||||||
|
/// its repeat count rather than a single point-in-time flap.
|
||||||
|
#[must_use]
|
||||||
|
pub fn agent_state_warnings<S: std::hash::BuildHasher>(
|
||||||
|
containers: &[ContainerView],
|
||||||
|
crash_counts: &HashMap<String, usize, S>,
|
||||||
|
) -> Vec<ServerWarning> {
|
||||||
|
let mut out = Vec::new();
|
||||||
|
|
||||||
|
// `needs_login` is already running-gated in `container_view::build_all`,
|
||||||
|
// so a stopped container never lights this.
|
||||||
|
let mut pending: Vec<&str> = containers
|
||||||
|
.iter()
|
||||||
|
.filter(|c| c.needs_login)
|
||||||
|
.map(|c| c.name.as_str())
|
||||||
|
.collect();
|
||||||
|
if !pending.is_empty() {
|
||||||
|
pending.sort_unstable();
|
||||||
|
out.push(ServerWarning {
|
||||||
|
kind: "pending_logins",
|
||||||
|
level: "warn",
|
||||||
|
message: format!(
|
||||||
|
"{n} agent{plural} need claude login: {list} \
|
||||||
|
— run `hivectl login <agent>` to authenticate",
|
||||||
|
n = pending.len(),
|
||||||
|
plural = if pending.len() == 1 { "" } else { "s" },
|
||||||
|
list = pending.join(", "),
|
||||||
|
),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
if !crash_counts.is_empty() {
|
||||||
|
let mut crashing: Vec<(&String, usize)> =
|
||||||
|
crash_counts.iter().map(|(a, n)| (a, *n)).collect();
|
||||||
|
crashing.sort_unstable_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(b.0)));
|
||||||
|
let list = crashing
|
||||||
|
.iter()
|
||||||
|
.map(|(a, n)| format!("{a} (×{n})"))
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
.join(", ");
|
||||||
|
out.push(ServerWarning {
|
||||||
|
kind: "agents_crashing",
|
||||||
|
level: "crit",
|
||||||
|
message: format!(
|
||||||
|
"{n} agent{plural} crashing: {list} \
|
||||||
|
— check the container journal (`hivectl logs <agent>`)",
|
||||||
|
n = crashing.len(),
|
||||||
|
plural = if crashing.len() == 1 { "" } else { "s" },
|
||||||
|
),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
||||||
/// Disk usage for the filesystem backing the host nix store — internal to
|
/// Disk usage for the filesystem backing the host nix store — internal to
|
||||||
/// the disk-pressure producer above.
|
/// the disk-pressure producer above.
|
||||||
struct DiskUsage {
|
struct DiskUsage {
|
||||||
|
|
@ -122,3 +188,68 @@ fn disk_usage(path: &str) -> Option<DiskUsage> {
|
||||||
used_pct,
|
used_pct,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
fn cv(name: &str, needs_login: bool) -> ContainerView {
|
||||||
|
ContainerView {
|
||||||
|
name: name.to_owned(),
|
||||||
|
container: format!("h-{name}"),
|
||||||
|
port: 0,
|
||||||
|
running: true,
|
||||||
|
needs_update: false,
|
||||||
|
needs_login,
|
||||||
|
deployed_sha: None,
|
||||||
|
pending_reminders: 0,
|
||||||
|
parent: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn no_agent_warnings_when_all_healthy() {
|
||||||
|
let containers = [cv("alice", false), cv("bob", false)];
|
||||||
|
assert!(agent_state_warnings(&containers, &HashMap::new()).is_empty());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn pending_logins_lists_sorted_agents() {
|
||||||
|
let containers = [cv("zoe", true), cv("amy", false), cv("bob", true)];
|
||||||
|
let w = agent_state_warnings(&containers, &HashMap::new());
|
||||||
|
assert_eq!(w.len(), 1);
|
||||||
|
assert_eq!(w[0].kind, "pending_logins");
|
||||||
|
assert_eq!(w[0].level, "warn");
|
||||||
|
// sorted, login-needing only, count reflected
|
||||||
|
assert!(w[0].message.starts_with("2 agents need claude login: bob, zoe"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn singular_grammar_for_one_agent() {
|
||||||
|
let containers = [cv("solo", true)];
|
||||||
|
let w = agent_state_warnings(&containers, &HashMap::new());
|
||||||
|
assert!(w[0].message.starts_with("1 agent needs claude login: solo"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn crashing_warning_is_crit_and_count_ordered() {
|
||||||
|
let crashes = HashMap::from([("flap".to_owned(), 5), ("blip".to_owned(), 1)]);
|
||||||
|
let w = agent_state_warnings(&[], &crashes);
|
||||||
|
assert_eq!(w.len(), 1);
|
||||||
|
assert_eq!(w[0].kind, "agents_crashing");
|
||||||
|
assert_eq!(w[0].level, "crit");
|
||||||
|
// higher crash count first
|
||||||
|
assert!(w[0].message.contains("flap (×5), blip (×1)"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn both_warnings_coexist() {
|
||||||
|
let containers = [cv("a", true)];
|
||||||
|
let crashes = HashMap::from([("b".to_owned(), 2)]);
|
||||||
|
let kinds: Vec<&str> = agent_state_warnings(&containers, &crashes)
|
||||||
|
.iter()
|
||||||
|
.map(|w| w.kind)
|
||||||
|
.collect();
|
||||||
|
assert_eq!(kinds, ["pending_logins", "agents_crashing"]);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue