This commit is contained in:
@@ -0,0 +1,405 @@
|
||||
//! monitor: real-time TUI of the managed processes (issue #50). Lists every
|
||||
//! running agent with CPU %, RSS, uptime; alerts on configurable thresholds
|
||||
//! (settings.monitor_thresholds) with a desktop notification + journal event.
|
||||
//! `--json` prints one JSON document per tick for scripts.
|
||||
|
||||
use super::*;
|
||||
use crate::config::MonitorThresholds;
|
||||
use crate::events::{Event, EventKind};
|
||||
use anyhow::{Context, Result};
|
||||
use serde::Serialize;
|
||||
use std::process::Command;
|
||||
use std::time::Duration;
|
||||
|
||||
/// One measured process row.
|
||||
#[derive(Debug, Clone, Default, Serialize)]
|
||||
pub struct AgentMetric {
|
||||
pub agent: String,
|
||||
pub pid: u32,
|
||||
pub cpu_pct: f64,
|
||||
pub mem_mb: f64,
|
||||
pub uptime_s: u64,
|
||||
/// "running" when the process answers, "dead" otherwise.
|
||||
pub state: String,
|
||||
}
|
||||
|
||||
/// Raw reading of a process (platform-dependent).
|
||||
#[derive(Debug, Clone, Copy, Default)]
|
||||
pub struct RawReading {
|
||||
/// Total CPU time consumed so far (seconds).
|
||||
pub cpu_total_s: f64,
|
||||
/// Resident set size in bytes.
|
||||
pub rss_bytes: u64,
|
||||
}
|
||||
|
||||
/// Read the raw counters of a process. Windows: PowerShell Get-Process
|
||||
/// (CPU = total seconds, WorkingSet64 = RSS). Unix: ps.
|
||||
fn raw_reading(pid: u32) -> Option<RawReading> {
|
||||
#[cfg(windows)]
|
||||
{
|
||||
let out = Command::new("powershell")
|
||||
.args([
|
||||
"-NoProfile",
|
||||
"-Command",
|
||||
&format!(
|
||||
"Get-Process -Id {pid} | Select-Object -ExpandProperty CPU; Get-Process -Id {pid} | Select-Object -ExpandProperty WorkingSet64"
|
||||
),
|
||||
])
|
||||
.output()
|
||||
.ok()?;
|
||||
if !out.status.success() {
|
||||
return None;
|
||||
}
|
||||
let text = String::from_utf8_lossy(&out.stdout);
|
||||
let mut lines = text.lines().filter(|l| !l.trim().is_empty());
|
||||
let cpu: f64 = lines.next()?.trim().parse().ok()?;
|
||||
let rss: u64 = lines.next()?.trim().parse().ok()?;
|
||||
Some(RawReading {
|
||||
cpu_total_s: cpu,
|
||||
rss_bytes: rss,
|
||||
})
|
||||
}
|
||||
#[cfg(unix)]
|
||||
{
|
||||
let out = Command::new("ps")
|
||||
.args(["-o", "time=,rss=", "-p", &pid.to_string()])
|
||||
.output()
|
||||
.ok()?;
|
||||
if !out.status.success() {
|
||||
return None;
|
||||
}
|
||||
let text = String::from_utf8_lossy(&out.stdout);
|
||||
let mut parts = text.split_whitespace();
|
||||
let time = parts.next()?;
|
||||
let rss_kb: u64 = parts.next()?.parse().ok()?;
|
||||
Some(RawReading {
|
||||
cpu_total_s: parse_ps_time(time),
|
||||
rss_bytes: rss_kb * 1024,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse `ps` cumulative time "MM:SS" or "HH:MM:SS" into seconds.
|
||||
fn parse_ps_time(t: &str) -> f64 {
|
||||
let parts: Vec<f64> = t
|
||||
.split(':')
|
||||
.filter_map(|p| p.parse().ok())
|
||||
.collect();
|
||||
match parts.as_slice() {
|
||||
[m, s] => m * 60.0 + s,
|
||||
[h, m, s] => h * 3600.0 + m * 60.0 + s,
|
||||
_ => 0.0,
|
||||
}
|
||||
}
|
||||
|
||||
/// Turn raw readings (taken `interval_s` apart) into a metric row.
|
||||
/// Pure: unit tested.
|
||||
pub fn to_metric(
|
||||
agent: &str,
|
||||
pid: u32,
|
||||
before: Option<RawReading>,
|
||||
after: RawReading,
|
||||
started_at: Option<&str>,
|
||||
interval_s: f64,
|
||||
) -> AgentMetric {
|
||||
let cpu_pct = match before {
|
||||
Some(b) if interval_s > 0.0 => {
|
||||
let delta = (after.cpu_total_s - b.cpu_total_s).max(0.0);
|
||||
(delta / interval_s * 100.0).min(100.0 * 64.0) // multi-core cap, generous
|
||||
}
|
||||
_ => 0.0,
|
||||
};
|
||||
let uptime_s = started_at
|
||||
.and_then(|t| chrono::DateTime::parse_from_rfc3339(t).ok())
|
||||
.map(|t| (chrono::Utc::now() - t.with_timezone(&chrono::Utc)).num_seconds().max(0) as u64)
|
||||
.unwrap_or(0);
|
||||
AgentMetric {
|
||||
agent: agent.to_string(),
|
||||
pid,
|
||||
cpu_pct,
|
||||
mem_mb: after.rss_bytes as f64 / (1024.0 * 1024.0),
|
||||
uptime_s,
|
||||
state: "running".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Decide which alerts fire for a metric against the thresholds. Pure.
|
||||
pub fn alerts_for(m: &AgentMetric, t: &MonitorThresholds) -> Vec<String> {
|
||||
let mut out = Vec::new();
|
||||
if let Some(cpu) = t.cpu_pct {
|
||||
if cpu > 0.0 && m.cpu_pct > cpu {
|
||||
out.push(format!(
|
||||
"{} cpu {:.1}% > {cpu}%",
|
||||
m.agent, m.cpu_pct
|
||||
));
|
||||
}
|
||||
}
|
||||
if let Some(mem) = t.mem_mb {
|
||||
if mem > 0 && m.mem_mb > mem as f64 {
|
||||
out.push(format!(
|
||||
"{} rss {:.0} MB > {mem} MB",
|
||||
m.agent, m.mem_mb
|
||||
));
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Collect the metrics of every managed agent with a live PID.
|
||||
pub fn collect(app: &App, interval_s: f64) -> Vec<AgentMetric> {
|
||||
let mut out = Vec::new();
|
||||
let Ok(sf) = app.state.load() else {
|
||||
return out;
|
||||
};
|
||||
for (name, entry) in &sf.installed {
|
||||
let Some(pid) = entry.pid else { continue };
|
||||
let Some(after) = raw_reading(pid) else {
|
||||
out.push(AgentMetric {
|
||||
agent: name.clone(),
|
||||
pid,
|
||||
state: "dead".to_string(),
|
||||
..Default::default()
|
||||
});
|
||||
continue;
|
||||
};
|
||||
let before = if interval_s > 0.0 {
|
||||
std::thread::sleep(Duration::from_secs_f64(interval_s));
|
||||
raw_reading(pid)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
out.push(to_metric(
|
||||
name,
|
||||
pid,
|
||||
before,
|
||||
after,
|
||||
entry.started_at.as_deref(),
|
||||
interval_s,
|
||||
));
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Emit one JSON document (one row per process) — script-friendly.
|
||||
fn print_json_rows(rows: &[AgentMetric]) {
|
||||
crate::output::print_json(&rows);
|
||||
}
|
||||
|
||||
/// Run the monitor: TUI by default, --json for scripts.
|
||||
pub fn run(app: &App, interval: Option<u64>, json: bool) -> Result<i32> {
|
||||
let secs = interval.unwrap_or(2).max(1) as f64;
|
||||
let thresholds = app
|
||||
.config
|
||||
.settings
|
||||
.monitor_thresholds
|
||||
.clone()
|
||||
.unwrap_or_default();
|
||||
if json {
|
||||
// One JSON document per tick, then sleep; Ctrl-C stops.
|
||||
loop {
|
||||
let rows = collect(app, 0.0);
|
||||
print_json_rows(&rows);
|
||||
for m in &rows {
|
||||
fire_alerts(app, m, &thresholds);
|
||||
}
|
||||
std::thread::sleep(Duration::from_secs_f64(secs));
|
||||
}
|
||||
}
|
||||
tui_loop(app, secs, &thresholds)
|
||||
}
|
||||
|
||||
/// Fire a desktop notification + journal event when a threshold is crossed.
|
||||
/// Alerts are rate-limited by the caller (one per tick).
|
||||
fn fire_alerts(app: &App, m: &AgentMetric, t: &MonitorThresholds) {
|
||||
for alert in alerts_for(m, t) {
|
||||
let _ = crate::process::desktop_notify("agent-manager monitor", &alert);
|
||||
app.emit(
|
||||
&Event::now(EventKind::Alert)
|
||||
.with_agent(m.agent.clone())
|
||||
.with_pid(m.pid)
|
||||
.with_reason(format!("monitor: {alert}")),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// ratatui loop: list of managed processes, refresh every `secs`, q/Esc quits.
|
||||
fn tui_loop(app: &App, secs: f64, thresholds: &MonitorThresholds) -> Result<i32> {
|
||||
use crossterm::event::{self, Event, KeyCode, KeyEventKind};
|
||||
use ratatui::layout::{Constraint, Layout};
|
||||
use ratatui::style::{Color, Modifier, Style};
|
||||
use ratatui::text::{Line, Span};
|
||||
use ratatui::widgets::{Block, Borders, List, ListItem};
|
||||
use ratatui::DefaultTerminal;
|
||||
|
||||
let mut terminal = ratatui::init();
|
||||
let result = (|| -> Result<i32> {
|
||||
loop {
|
||||
let rows = collect(app, secs);
|
||||
for m in &rows {
|
||||
fire_alerts(app, m, thresholds);
|
||||
}
|
||||
terminal.draw(|f| {
|
||||
let chunks = Layout::vertical([
|
||||
Constraint::Length(1),
|
||||
Constraint::Min(0),
|
||||
Constraint::Length(1),
|
||||
])
|
||||
.split(f.area());
|
||||
let title = Line::from(vec![
|
||||
Span::styled(
|
||||
" am monitor ",
|
||||
Style::default()
|
||||
.fg(Color::Cyan)
|
||||
.add_modifier(Modifier::BOLD),
|
||||
),
|
||||
Span::styled(
|
||||
format!("{} managed process(es)", rows.len()),
|
||||
Style::default().fg(Color::Gray),
|
||||
),
|
||||
]);
|
||||
f.render_widget(
|
||||
Block::default().borders(Borders::TOP).title(title),
|
||||
chunks[0],
|
||||
);
|
||||
let items: Vec<ListItem> = rows
|
||||
.iter()
|
||||
.map(|m| {
|
||||
let state = if m.state == "dead" {
|
||||
Style::default().fg(Color::Red)
|
||||
} else {
|
||||
Style::default().fg(Color::Green)
|
||||
};
|
||||
ListItem::new(Line::from(vec![
|
||||
Span::styled(
|
||||
format!(
|
||||
"{:<20} pid {:>7} cpu {:>6.1}% rss {:>8.0} MB up {}",
|
||||
m.agent,
|
||||
m.pid,
|
||||
m.cpu_pct,
|
||||
m.mem_mb,
|
||||
human_uptime(m.uptime_s),
|
||||
),
|
||||
state,
|
||||
),
|
||||
]))
|
||||
})
|
||||
.collect();
|
||||
f.render_widget(
|
||||
List::new(items).block(Block::default().borders(Borders::ALL)),
|
||||
chunks[1],
|
||||
);
|
||||
let hint = Line::from(Span::styled(
|
||||
" q / Esc: quit ",
|
||||
Style::default().fg(Color::DarkGray),
|
||||
));
|
||||
f.render_widget(hint, chunks[2]);
|
||||
})?;
|
||||
if event::poll(Duration::from_millis(100))? {
|
||||
if let Event::Key(k) = event::read()? {
|
||||
if k.kind == KeyEventKind::Press
|
||||
&& matches!(k.code, KeyCode::Char('q') | KeyCode::Esc)
|
||||
{
|
||||
return Ok(0);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
})();
|
||||
ratatui::restore();
|
||||
result
|
||||
}
|
||||
|
||||
fn human_uptime(secs: u64) -> String {
|
||||
if secs < 60 {
|
||||
format!("{secs}s")
|
||||
} else if secs < 3600 {
|
||||
format!("{}m{:02}s", secs / 60, secs % 60)
|
||||
} else {
|
||||
format!("{}h{:02}m", secs / 3600, (secs % 3600) / 60)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn to_metric_computes_cpu_and_mem() {
|
||||
let before = RawReading {
|
||||
cpu_total_s: 10.0,
|
||||
rss_bytes: 2 * 1024 * 1024,
|
||||
};
|
||||
let after = RawReading {
|
||||
cpu_total_s: 12.0,
|
||||
rss_bytes: 4 * 1024 * 1024,
|
||||
};
|
||||
let m = to_metric("a", 42, Some(before), after, None, 2.0);
|
||||
assert_eq!(m.pid, 42);
|
||||
assert!((m.cpu_pct - 100.0).abs() < 1e-6, "{}", m.cpu_pct);
|
||||
assert!((m.mem_mb - 4.0).abs() < 1e-6);
|
||||
assert_eq!(m.state, "running");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn to_metric_first_sample_zero_cpu() {
|
||||
let after = RawReading {
|
||||
cpu_total_s: 5.0,
|
||||
rss_bytes: 1024 * 1024,
|
||||
};
|
||||
let m = to_metric("a", 1, None, after, None, 2.0);
|
||||
assert_eq!(m.cpu_pct, 0.0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn to_metric_parses_uptime() {
|
||||
let now = chrono::Utc::now();
|
||||
let started = now - chrono::Duration::minutes(5);
|
||||
let after = RawReading {
|
||||
cpu_total_s: 1.0,
|
||||
rss_bytes: 1024 * 1024,
|
||||
};
|
||||
let m = to_metric("a", 1, None, after, Some(&started.to_rfc3339()), 0.0);
|
||||
assert!((260..=320).contains(&m.uptime_s), "{}", m.uptime_s);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn alerts_fire_on_thresholds() {
|
||||
let m = AgentMetric {
|
||||
agent: "a".into(),
|
||||
pid: 1,
|
||||
cpu_pct: 95.0,
|
||||
mem_mb: 2000.0,
|
||||
uptime_s: 0,
|
||||
state: "running".into(),
|
||||
};
|
||||
let t = MonitorThresholds {
|
||||
cpu_pct: Some(80.0),
|
||||
mem_mb: Some(1024),
|
||||
};
|
||||
let alerts = alerts_for(&m, &t);
|
||||
assert_eq!(alerts.len(), 2);
|
||||
// Below thresholds → nothing.
|
||||
let quiet = AgentMetric {
|
||||
cpu_pct: 10.0,
|
||||
mem_mb: 100.0,
|
||||
..m.clone()
|
||||
};
|
||||
assert!(alerts_for(&quiet, &t).is_empty());
|
||||
// 0 disables the threshold.
|
||||
let off = MonitorThresholds {
|
||||
cpu_pct: Some(0.0),
|
||||
mem_mb: Some(0),
|
||||
};
|
||||
assert!(alerts_for(&m, &off).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_ps_time_formats() {
|
||||
assert!((parse_ps_time("01:30") - 90.0).abs() < 1e-9);
|
||||
assert!((parse_ps_time("1:02:03") - 3723.0).abs() < 1e-9);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user