diff --git a/cmux-tui/crates/cmux-host/src/cloud/mod.rs b/cmux-tui/crates/cmux-host/src/cloud/mod.rs index 9e5477dabcf0..8905bb93cbaa 100644 --- a/cmux-tui/crates/cmux-host/src/cloud/mod.rs +++ b/cmux-tui/crates/cmux-host/src/cloud/mod.rs @@ -15,10 +15,14 @@ //! (`acp_activity` socket lines, held 90 s unless renewed), `activity` //! advertised only while the stream is live and `activity-turns` with it //! only while acpmux's lines arrive; +//! - workload: every 30 s a probe of the machine's processes using CPU, +//! listening sockets and timers that are not the system's ([`workload`]), +//! reported as `activity.workload` with `activity-workload` while it reads; +//! the Cloud reads a machine as idle only when it counts none; //! - resume: the supervisor's `Resumed` event (clock-set wake, id //! confirmed) sends one report named `resume`. //! -//! Modules: [`wire`] (contracts, pure), [`sender`] (the two senders, sans +//! Modules: [`wire`] (contracts, pure), [`workload`] (the workload probe), [`sender`] (the two senders, sans //! I/O), [`session`] (inputs to requests, sans I/O), [`client`] (bind and //! auth over the [`client::Http`] and [`client::Store`] traits), [`github`] //! (GitHub tokens for the machine's git and `gh` over the agent socket), @@ -61,6 +65,7 @@ pub mod trajectory; #[cfg(unix)] pub mod trajectory_thread; pub mod wire; +pub mod workload; #[cfg(all(test, unix))] mod github_tests; diff --git a/cmux-tui/crates/cmux-host/src/cloud/role.rs b/cmux-tui/crates/cmux-host/src/cloud/role.rs index 4c6eb87fae95..c12ed820918b 100644 --- a/cmux-tui/crates/cmux-host/src/cloud/role.rs +++ b/cmux-tui/crates/cmux-host/src/cloud/role.rs @@ -492,6 +492,9 @@ struct Worker { relay: Option, bind_retry_at: Option, bind_backoff: Backoff, + /// The workload probe (`workload.rs`): its memory and when it runs next (while bound). + workload: super::workload::Tracker, + workload_at: Option, rng: SystemRandom, epoch: Instant, } @@ -541,6 +544,8 @@ impl Worker { relay: None, bind_retry_at: None, bind_backoff: Backoff::new(2_000, 600_000), + workload: super::workload::Tracker::default(), + workload_at: None, rng: SystemRandom::new(), epoch: Instant::now(), }) @@ -567,6 +572,7 @@ impl Worker { self.bind_retry_at, self.activity.retry_at, self.bootstrap_retry_at, + self.workload_at, ] .into_iter() .flatten() @@ -963,6 +969,9 @@ impl Worker { self.activity = self.activity_reset(); self.connect_activity(); self.drive(reqs); + // A new session knows no workload yet: the first probe now, the first result one interval on. + self.workload = super::workload::Tracker::default(); + self.workload_at = Some(self.now()); } fn activity_reset(&self) -> Activity { @@ -1068,6 +1077,9 @@ impl Worker { if self.activity.retry_at.is_some_and(|at| at <= now) { self.connect_activity(); } + if self.workload_at.is_some_and(|at| at <= self.now()) { + self.probe_workload(); + } let now = self.now(); if let Some(live) = self.live.as_mut() { let reqs = live.session.fire(now); @@ -1075,6 +1087,27 @@ impl Worker { } } + /// One workload probe (`workload.rs`); the next one an interval later. A probe that does not + /// read makes the workload unknown, and the report then drops `activity-workload`. + fn probe_workload(&mut self) { + let now = self.now(); + self.workload_at = Some(now + super::workload::PROBE_INTERVAL_MS); + let Some(live) = self.live.as_mut() else { return }; + let seen = match super::workload::read_snapshot(self.paths.root(), now) { + // The first probe has no CPU baseline yet: it changes nothing. + Some(snapshot) => match self.workload.observe(&snapshot, wall_ms()) { + Some(workload) => Some(workload), + None => return, + }, + None => { + self.workload = super::workload::Tracker::default(); + None + } + }; + let reqs = live.session.workload(seen, now); + self.drive(reqs); + } + /// Sends each request and feeds its answer back until nothing is due. fn drive(&mut self, mut queue: Vec) { while let Some(req) = queue.pop() { diff --git a/cmux-tui/crates/cmux-host/src/cloud/session.rs b/cmux-tui/crates/cmux-host/src/cloud/session.rs index 87f26164c091..8cd0c0ec6ad7 100644 --- a/cmux-tui/crates/cmux-host/src/cloud/session.rs +++ b/cmux-tui/crates/cmux-host/src/cloud/session.rs @@ -12,6 +12,7 @@ use super::wire::{ ACP_ACTIVITY_TTL_MS, ATTENTION_EMIT_OP, Activity, ActivityChange, DaemonInfo, EVENT_EMIT_OP, activity_from_daemon, }; +use super::workload::Workload; use serde_json::json; pub struct Session { @@ -27,6 +28,8 @@ pub struct Session { acp: AcpActivity, /// The daemon's activity stream is live. stream_live: bool, + /// The workload probe's last reported result (`workload.rs`); None while it is unknown. + workload: Option, } /// What the machine's acpmux daemons last said: each one's sessions at work and until when it @@ -149,6 +152,7 @@ impl Session { own: Activity::default(), acp: AcpActivity::default(), stream_live: false, + workload: None, } } @@ -195,11 +199,7 @@ impl Session { /// the activity change together, so the report that goes carries both (cmux#33331: an /// expiry's report never still says the count is known). fn acp_refresh(&mut self, now: u64) -> Vec { - let daemon = self.base.with_activity( - self.stream_live, - self.acp.known(), - self.acp.phases().is_some(), - ); + let daemon = self.daemon_now(); let block_changed = &daemon != self.reporter.daemon(); self.reporter.replace_daemon(daemon); let mut out = self.acp_changed(now); @@ -225,14 +225,43 @@ impl Session { self.reporter.set_activity(activity, report, now).into_iter().collect() } - /// The activity a report carries: the daemon's plus acpmux's. + /// The activity a report carries: the daemon's plus acpmux's, and the workload. fn activity(&self) -> Activity { - self.own.with_acp( + let mut activity = self.own.with_acp( self.acp.sessions(), self.acp.last_action_at, self.acp.phases(), self.acp.last_person_prompt_at, - ) + ); + activity.workload = self.workload.as_ref().map(Workload::to_json); + activity + } + + /// The daemon block as it stands: `activity` while the stream is live, `activity-turns` with + /// it while acpmux's count is known, `activity-phases` with that while its phases are, and + /// `activity-workload` with `activity` while the workload probe reads. + fn daemon_now(&self) -> DaemonInfo { + self.base + .with_activity(self.stream_live, self.acp.known(), self.acp.phases().is_some()) + .with_workload(self.workload.is_some()) + } + + /// One workload probe (`workload.rs`; None: it did not read). A report goes when the machine + /// turns busy or quiet, when its listening ports or timers change, when a quiet period starts + /// anew, and when the probe starts or stops reading. Which processes use CPU while it is busy + /// rides along with the next report. + pub fn workload(&mut self, seen: Option, now: u64) -> Vec { + let next = match (seen, &self.workload) { + (Some(seen), Some(cur)) if seen.key() == cur.key() => return Vec::new(), + (seen, _) => seen, + }; + if next.is_none() && self.workload.is_none() { + return Vec::new(); + } + self.workload = next; + self.reporter.replace_daemon(self.daemon_now()); + let activity = self.activity(); + self.reporter.set_activity(activity, true, now).into_iter().collect() } /// The activity stream connected or dropped: `activity` is advertised @@ -245,11 +274,7 @@ impl Session { /// The daemon block as it stands: `activity` while the stream is live, `activity-turns` with /// it while acpmux's count is known. fn daemon_block(&mut self, now: u64) -> Vec { - let daemon = self.base.with_activity( - self.stream_live, - self.acp.known(), - self.acp.phases().is_some(), - ); + let daemon = self.daemon_now(); if &daemon == self.reporter.daemon() { return Vec::new(); } @@ -737,6 +762,70 @@ mod tests { assert!(read_phases(&json!({"turns": 0}), 0).is_none(), "no block"); } + /// The workload probe's result reaches the report as `activity.workload` with + /// `activity-workload` (only with `activity`); a report goes when the machine turns busy or + /// quiet, not when only which process uses CPU changes; a probe that stops reading withdraws + /// the capability, so the Cloud reads the machine as busy. + #[test] + fn the_workload_reaches_the_report_and_only_a_change_of_state_is_reported() { + use super::super::wire::ACTIVITY_WORKLOAD_CAPABILITY; + let mut s = session(); + let send = |s: &mut Session, mut r: Vec, now: u64| { + r.extend(s.fire(now)); + for req in &r { + s.answered(req, &ok(), now, 0.5); + } + r + }; + let start = s.start(0); + send(&mut s, start, 0); + let cap = |r: &OpRequest| { + r.params["daemon"]["capabilities"] + .as_array() + .unwrap() + .contains(&json!(ACTIVITY_WORKLOAD_CAPABILITY)) + }; + let busy = |comm: &str, at: u64| Workload { + busy_processes: vec![comm.into()], + last_busy_at: Some(at), + ..Workload::default() + }; + // Without the activity stream the block rides along, never the capability. + let r = s.workload(Some(busy("node", 1)), 20_000); + let r = send(&mut s, r, 20_000); + assert_eq!(r.len(), 1); + assert!(!cap(&r[0])); + assert_eq!(r[0].params["activity"]["workload"]["busy_processes"], 1); + let r = s.activity_stream(true, 40_000); + let r = send(&mut s, r, 40_000); + assert!(cap(&r[0])); + assert_eq!(r[0].params["activity"]["workload"]["process_names"], json!(["node"])); + + // Another process now uses CPU: still busy, no report. + assert!(s.workload(Some(busy("cargo", 2)), 60_000).is_empty()); + // Quiet: a report, with the last busy time. + let quiet = Workload { last_busy_at: Some(2), ..Workload::default() }; + let r = s.workload(Some(quiet.clone()), 80_000); + let r = send(&mut s, r, 80_000); + assert_eq!(r.len(), 1); + assert_eq!(r[0].params["activity"]["workload"]["busy_processes"], 0); + assert_eq!(r[0].params["activity"]["workload"]["last_busy_at"], 2); + assert!(s.workload(Some(quiet), 100_000).is_empty(), "nothing changed"); + // A port opens: a report. + let port = + Workload { listening_ports: vec![3000], last_busy_at: Some(3), ..Workload::default() }; + let r = s.workload(Some(port), 120_000); + let r = send(&mut s, r, 120_000); + assert_eq!(r[0].params["activity"]["workload"]["ports"], json!([3000])); + // The probe stops reading: the capability and the block go. + let r = s.workload(None, 140_000); + let r = send(&mut s, r, 140_000); + assert_eq!(r.len(), 1); + assert!(!cap(&r[0])); + assert!(r[0].params["activity"].get("workload").is_none()); + assert!(s.workload(None, 160_000).is_empty()); + } + /// Relayed only when every daemon says so; turns from every daemon. #[test] fn the_machine_phases_join_every_daemon() { diff --git a/cmux-tui/crates/cmux-host/src/cloud/wire.rs b/cmux-tui/crates/cmux-host/src/cloud/wire.rs index 755412fc6472..250d96f3e6d4 100644 --- a/cmux-tui/crates/cmux-host/src/cloud/wire.rs +++ b/cmux-tui/crates/cmux-host/src/cloud/wire.rs @@ -50,6 +50,12 @@ pub const ACTIVITY_TURNS_CAPABILITY: &str = "activity-turns"; /// work; acpmux `daemon/cloud_activity.rs` names the phases). The Cloud sleeps a machine inside /// its turns only from a report that advertises this (cmux#34298). pub const ACTIVITY_PHASES_CAPABILITY: &str = "activity-phases"; +/// Advertised with `activity` while the agent's workload probe reads (`workload.rs`): the +/// report's `activity.workload` counts the processes using CPU, the listening sockets and the +/// timers that are not the system's. The Cloud reads a machine as idle only from a report that +/// advertises this and counts none of them, so a dev server, a build or a cron job an agent left +/// running is never paused. +pub const ACTIVITY_WORKLOAD_CAPABILITY: &str = "activity-workload"; /// How long acpmux's count of sessions at work holds without a new `acp_activity` line /// (acpmux renews it every 20 s, `daemon/cloud_activity.rs`): a /// daemon that died mid-turn does not keep the machine awake for good. @@ -212,13 +218,31 @@ impl DaemonInfo { DaemonInfo { version: self.version.clone(), capabilities } } + /// The same block with `activity-workload` on (the workload is known and `activity` is on) + /// or off. [`Self::with_activity`] drops it: apply this after. + pub fn with_workload(&self, on: bool) -> DaemonInfo { + let mut capabilities: Vec = self + .capabilities + .iter() + .filter(|c| *c != ACTIVITY_WORKLOAD_CAPABILITY) + .cloned() + .collect(); + if on && capabilities.iter().any(|c| c == ACTIVITY_CAPABILITY) { + capabilities.push(ACTIVITY_WORKLOAD_CAPABILITY.to_owned()); + } + DaemonInfo { version: self.version.clone(), capabilities } + } + pub fn to_json(&self) -> Value { json!({ "version": self.version, "capabilities": self.capabilities }) } } fn is_activity_capability(c: &str) -> bool { - c == ACTIVITY_CAPABILITY || c == ACTIVITY_TURNS_CAPABILITY || c == ACTIVITY_PHASES_CAPABILITY + c == ACTIVITY_CAPABILITY + || c == ACTIVITY_TURNS_CAPABILITY + || c == ACTIVITY_PHASES_CAPABILITY + || c == ACTIVITY_WORKLOAD_CAPABILITY } /// The daemon block from one `identify` answer's `data`: version plus the @@ -263,6 +287,9 @@ pub struct Activity { /// the Cloud keeps the machine running for a while after it (cmux-platform `IdlePolicy.ts`, /// the post-chat warm window). pub last_person_prompt_at: Option, + /// The agent's workload probe (`workload.rs`), while it reads + /// ([`ACTIVITY_WORKLOAD_CAPABILITY`]). + pub workload: Option, } impl Activity { @@ -281,6 +308,9 @@ impl Activity { if let Some(at) = self.last_person_prompt_at { out.insert("last_person_prompt_at".to_owned(), json!(at)); } + if let Some(workload) = &self.workload { + out.insert("workload".to_owned(), workload.clone()); + } Value::Object(out) } } @@ -302,6 +332,7 @@ impl Activity { last_agent_action_at: self.last_agent_action_at.max(acp_action_at), phases: acp_phases, last_person_prompt_at: acp_prompt_at, + workload: self.workload.clone(), } } } diff --git a/cmux-tui/crates/cmux-host/src/cloud/workload.rs b/cmux-tui/crates/cmux-host/src/cloud/workload.rs new file mode 100644 index 000000000000..b44082ec0e21 --- /dev/null +++ b/cmux-tui/crates/cmux-host/src/cloud/workload.rs @@ -0,0 +1,495 @@ +//! The machine's own workload for the Cloud's idle decision: processes using CPU, sockets +//! listening and timers (cron, systemd timers, at jobs) that are not the system's. +//! +//! The Cloud pauses a machine it reads as idle. Until this, idle meant no input and no agent +//! turn, so a pause froze what an agent left running: a dev server, a build, a cron job +//! (exe.dev, : "Every time we talked about quiescing, we +//! realized it would break cron jobs."). The report's `activity.workload` (capability +//! `activity-workload`) now says what runs, and cmux-platform reads a machine as idle only when +//! it counts nothing: +//! +//! - `busy_processes`: processes that used more than [`CPU_BUSY_PERCENT`] of one CPU between +//! two probes ([`PROBE_INTERVAL_MS`] apart). A process's reaped children count toward it, so +//! a `make` whose compilers each live a second still reads busy. +//! - `listening_ports`: TCP sockets in LISTEN (IPv4 and IPv6, loopback included: a dev server on +//! 127.0.0.1 is still a dev server). A socket whose owner cannot be found counts. +//! - `timers`: user crontabs, `/etc/cron.d` entries, at jobs, transient (`systemd-run`) timers +//! and timer units under `/etc/systemd/system` or a user's `~/.config/systemd/user`. +//! +//! Each count comes with at most [`NAMES_MAX`] names: a process's `comm` (15 characters, never +//! a command line), a port number, a timer's unit or crontab name. Never a command line, an +//! argument or an environment, so no secret travels. +//! +//! System work never counts, by the documented allowlists [`SYSTEM_PROCESSES`], +//! [`SYSTEM_PROCESS_PREFIXES`], [`SYSTEM_TIMERS`] and [`SYSTEM_CRON_D`]: kernel threads, systemd +//! and its daemons, sshd, cron and atd themselves, journald, dbus, this agent (cmux-host), the +//! terminal daemon (cmux-tui and its terminal hosts), acpmux, the machine health timer, the +//! image's desktop (Xvnc, websockify, openbox, ...) and the distribution's housekeeping timers. +//! +//! A probe that cannot read `/proc` is unknown: the report then drops `activity-workload`, and +//! the Cloud never pauses a machine whose workload it does not know. + +use serde_json::{Value, json}; +use std::collections::{BTreeMap, BTreeSet}; + +/// How often the agent probes. +pub const PROBE_INTERVAL_MS: u64 = 30_000; +/// A process above this share of one CPU between two probes is busy. +pub const CPU_BUSY_PERCENT: u64 = 2; +/// At most this many names per list. +pub const NAMES_MAX: usize = 8; + +/// System processes by `comm` (exact): never busy, their listeners never count. +pub const SYSTEM_PROCESSES: &[&str] = &[ + // init and systemd's own daemons (comm is at most 15 characters). + "systemd", + "(sd-pam)", + "init", + "systemd-journal", + "systemd-logind", + "systemd-udevd", + "systemd-network", + "systemd-resolve", + "systemd-timesyn", + "systemd-oomd", + "systemd-userdbd", + "systemd-userwor", + // Remote access, scheduling, logging, the bus. + "sshd", + "sshd-session", + "cron", + "crond", + "atd", + "rsyslogd", + "dbus-daemon", + "dbus-broker", + "dbus-broker-lau", + "agetty", + "polkitd", + "chronyd", + "irqbalance", + "multipathd", + "unattended-upgr", + "snapd", + "tailscaled", + "containerd", + "dockerd", + // cmux on the machine: this agent, the CLI, acpmux and its idle daemon, machine health. + "cmux-host", + "cmux", + "acpmux", + "cmux-machine-he", + // The image's desktop (Cloud computer use). + "Xtigervnc", + "Xvnc", + "openbox", + "tint2", + "vncconfig", + "websockify", + "at-spi-bus-laun", + "at-spi2-registr", + "cmux-desktop-bo", +]; + +/// System processes by `comm` prefix: the terminal daemon and its terminal hosts. +pub const SYSTEM_PROCESS_PREFIXES: &[&str] = &["cmux-tui", "__terminal"]; + +/// Timer units that are the system's: the distribution's housekeeping (the same list +/// [`crate::config::HOUSEKEEPING_TIMERS`] stops at park, plus common others) and cmux's own. +pub const SYSTEM_TIMERS: &[&str] = &[ + "apt-daily.timer", + "apt-daily-upgrade.timer", + "dpkg-db-backup.timer", + "e2scrub_all.timer", + "fstrim.timer", + "logrotate.timer", + "man-db.timer", + "motd-news.timer", + "systemd-tmpfiles-clean.timer", + "fwupd-refresh.timer", + "ua-timer.timer", + "update-notifier-download.timer", + "update-notifier-motd.timer", + "snapd.snap-repair.timer", + "sysstat-collect.timer", + "sysstat-summary.timer", + "phpsessionclean.timer", + "certbot.timer", + "apport-autoreport.timer", + "plocate-updatedb.timer", + "mlocate.timer", + "anacron.timer", + "cmux-machine-health.timer", + "cmux-self-update-guard.timer", +]; + +/// `/etc/cron.d` entries that are the distribution's. +pub const SYSTEM_CRON_D: &[&str] = &[ + ".placeholder", + "e2scrub_all", + "popularity-contest", + "sysstat", + "php", + "anacron", + "certbot", + "mdadm", +]; + +/// Whether `comm` is a system process. +pub fn is_system_process(comm: &str) -> bool { + SYSTEM_PROCESSES.contains(&comm) || SYSTEM_PROCESS_PREFIXES.iter().any(|p| comm.starts_with(p)) +} + +/// Whether a timer the reader found is the system's. Names come as `KIND:NAME` (`cron:alice`, +/// `cron.d:backup`, `at:a0001`, `systemd:x.timer`, `user:x.timer`). +pub fn is_system_timer(name: &str) -> bool { + match name.split_once(':') { + Some(("systemd", unit)) => SYSTEM_TIMERS.contains(&unit), + Some(("cron.d", file)) => SYSTEM_CRON_D.contains(&file), + _ => false, + } +} + +/// One process at one probe (`/proc//stat`). +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ProcStat { + pub pid: u32, + pub comm: String, + /// Start time in clock ticks after boot: with the pid, the process's identity. + pub start: u64, + /// utime + stime + cutime + cstime, in clock ticks. + pub ticks: u64, + pub kthread: bool, +} + +/// `/proc//stat`, or None when it does not read. `comm` may hold spaces and parens. +pub fn parse_stat(pid: u32, text: &str) -> Option { + let open = text.find('(')?; + let close = text.rfind(')')?; + let comm = text.get(open + 1..close)?.to_owned(); + let rest: Vec<&str> = text.get(close + 1..)?.split_whitespace().collect(); + // rest[0] is field 3 (state): field k is rest[k - 3]. + let field = |k: usize| rest.get(k - 3).and_then(|v| v.parse::().ok()); + const PF_KTHREAD: u64 = 0x0020_0000; + let flags = field(9)?; + let ticks = + [14, 15, 16, 17].iter().try_fold(0u64, |sum, k| Some(sum.saturating_add(field(*k)?)))?; + Some(ProcStat { pid, comm, start: field(22)?, ticks, kthread: flags & PF_KTHREAD != 0 }) +} + +/// The LISTEN sockets of one `/proc/net/tcp` or `/proc/net/tcp6`: (port, inode). +pub fn parse_listeners(text: &str) -> Vec<(u16, u64)> { + text.lines() + .skip(1) + .filter_map(|line| { + let f: Vec<&str> = line.split_whitespace().collect(); + if f.get(3) != Some(&"0A") { + return None; + } + let port = u16::from_str_radix(f.get(1)?.rsplit(':').next()?, 16).ok()?; + let inode = f.get(9)?.parse::().ok()?; + (inode > 0).then_some((port, inode)) + }) + .collect() +} + +/// Whether a crontab holds a job: a line that is not blank, a comment or an assignment. +pub fn crontab_has_jobs(text: &str) -> bool { + text.lines().map(str::trim).any(|line| { + !line.is_empty() + && !line.starts_with('#') + && !line.split_whitespace().next().is_some_and(|first| first.contains('=')) + }) +} + +/// What one probe read. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct Snapshot { + /// Monotonic milliseconds of the probe. + pub mono_ms: u64, + /// Clock ticks per second (`sysconf(_SC_CLK_TCK)`). + pub clk_tck: u64, + pub procs: Vec, + /// Each LISTEN socket's port and its owner's `comm`, None when no process holds it. + pub listeners: Vec<(u16, Option)>, + /// Every timer found, as `KIND:NAME` ([`is_system_timer`]). + pub timers: Vec, +} + +/// The workload as the report carries it. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct Workload { + pub busy_processes: Vec, + pub listening_ports: Vec, + pub timers: Vec, + /// When a probe last found the machine busy (Unix ms): the idle period starts no earlier. + pub last_busy_at: Option, +} + +impl Workload { + pub fn busy(&self) -> bool { + !(self.busy_processes.is_empty() + && self.listening_ports.is_empty() + && self.timers.is_empty()) + } + + /// What a new report waits for: busy or not, the ports and timers, and the last busy time + /// while quiet. Which processes use CPU while busy changes from probe to probe; it rides along + /// with the next report instead of making one every probe. + pub fn key(&self) -> (bool, &[u16], &[String], Option) { + let busy = !self.busy_processes.is_empty(); + let quiet_since = if self.busy() { None } else { self.last_busy_at }; + (busy, &self.listening_ports, &self.timers, quiet_since) + } + + /// `activity.workload` of the report: counts, at most [`NAMES_MAX`] names each. + pub fn to_json(&self) -> Value { + let first = |v: &[String]| v.iter().take(NAMES_MAX).cloned().collect::>(); + let mut out = json!({ + "busy_processes": self.busy_processes.len(), + "listening_ports": self.listening_ports.len(), + "timers": self.timers.len(), + "process_names": first(&self.busy_processes), + "ports": self.listening_ports.iter().take(NAMES_MAX).collect::>(), + "timer_names": first(&self.timers), + }); + if let (Some(at), Value::Object(map)) = (self.last_busy_at, &mut out) { + map.insert("last_busy_at".to_owned(), json!(at)); + } + out + } +} + +/// CPU ticks by process identity (pid, start time). +type Ticks = BTreeMap<(u32, u64), u64>; + +/// The probes' memory: the last sample's CPU ticks and when the machine was last busy. +#[derive(Debug, Default)] +pub struct Tracker { + prev: Option<(u64, Ticks)>, + last_busy_at: Option, +} + +impl Tracker { + /// The workload from this probe (`wall_ms`: its Unix time), or None while CPU use is not + /// known yet (the first probe) or the clock did not move. + pub fn observe(&mut self, snap: &Snapshot, wall_ms: u64) -> Option { + let ticks: Ticks = snap.procs.iter().map(|p| ((p.pid, p.start), p.ticks)).collect(); + let prev = self.prev.replace((snap.mono_ms, ticks)); + let (prev_ms, prev_ticks) = prev?; + let window_ms = snap.mono_ms.checked_sub(prev_ms).filter(|ms| *ms > 0)?; + let clk_tck = snap.clk_tck.max(1); + let mut busy = BTreeSet::new(); + for p in &snap.procs { + if p.kthread || is_system_process(&p.comm) { + continue; + } + // A process new since the last probe used all its ticks inside the window. + let used = + p.ticks.saturating_sub(prev_ticks.get(&(p.pid, p.start)).copied().unwrap_or(0)); + // used / clk_tck seconds over window_ms / 1000 seconds, above CPU_BUSY_PERCENT. + if used.saturating_mul(1000 * 100) > CPU_BUSY_PERCENT * clk_tck * window_ms { + busy.insert(p.comm.clone()); + } + } + let ports: BTreeSet = snap + .listeners + .iter() + .filter(|(_, owner)| !owner.as_deref().is_some_and(is_system_process)) + .map(|(port, _)| *port) + .collect(); + let timers: BTreeSet = + snap.timers.iter().filter(|t| !is_system_timer(t)).cloned().collect(); + let mut w = Workload { + busy_processes: busy.into_iter().collect(), + listening_ports: ports.into_iter().collect(), + timers: timers.into_iter().collect(), + last_busy_at: None, + }; + if w.busy() { + // Never earlier than a busy time already seen: a clock stepped back must not lengthen the idle period. + self.last_busy_at = Some(self.last_busy_at.map_or(wall_ms, |t| t.max(wall_ms))); + } + w.last_busy_at = self.last_busy_at; + Some(w) + } +} + +#[cfg(unix)] +pub use reader::read_snapshot; + +#[cfg(unix)] +mod reader { + use super::{ProcStat, Snapshot, crontab_has_jobs, parse_listeners, parse_stat}; + use std::collections::BTreeMap; + use std::fs; + use std::path::Path; + + /// One probe of the machine under `root` (`/` in production), or None when `/proc` does not + /// read: the workload is then unknown. + pub fn read_snapshot(root: &Path, mono_ms: u64) -> Option { + let proc_dir = root.join("proc"); + let mut pids: Vec = Vec::new(); + for entry in fs::read_dir(&proc_dir).ok()? { + // An entry that does not read could be a running process: the workload is unknown. + let entry = entry.ok()?; + if let Some(pid) = entry.file_name().to_str().and_then(|n| n.parse().ok()) { + pids.push(pid); + } + } + let mut procs: Vec = Vec::with_capacity(pids.len()); + for pid in pids { + match fs::read_to_string(proc_dir.join(format!("{pid}/stat"))) { + // A process that exited between the listing and the read is gone, not unknown. + Err(e) if gone(&e) => {} + // Any other failure could hide a running process: the workload is unknown. + Err(_) => return None, + Ok(text) => procs.push(parse_stat(pid, &text)?), + } + } + if procs.is_empty() { + return None; + } + let mut sockets: Vec<(u16, u64)> = Vec::new(); + for table in ["net/tcp", "net/tcp6"] { + match fs::read_to_string(proc_dir.join(table)) { + Ok(text) => sockets.extend(parse_listeners(&text)), + // No IPv6 table (a kernel without IPv6): nothing listens there. IPv4's is required. + Err(e) if table == "net/tcp6" && e.kind() == std::io::ErrorKind::NotFound => {} + // A table that exists but does not read could hide a listener: unknown. + Err(_) => return None, + } + } + let owners = socket_owners(&proc_dir, &procs, &sockets); + let listeners = + sockets.iter().map(|(port, inode)| (*port, owners.get(inode).cloned())).collect(); + // SAFETY: sysconf has no preconditions. + let clk_tck = unsafe { libc::sysconf(libc::_SC_CLK_TCK) }; + Some(Snapshot { + mono_ms, + clk_tck: u64::try_from(clk_tck).ok().filter(|t| *t > 0).unwrap_or(100), + procs, + listeners, + timers: timers(root), + }) + } + + /// A read failed because the process exited (its `/proc` entry is gone). + fn gone(e: &std::io::Error) -> bool { + e.kind() == std::io::ErrorKind::NotFound || e.raw_os_error() == Some(libc::ESRCH) + } + + /// The `comm` of the process holding each wanted socket inode (`/proc//fd`). + fn socket_owners( + proc_dir: &Path, + procs: &[ProcStat], + sockets: &[(u16, u64)], + ) -> BTreeMap { + let mut owners = BTreeMap::new(); + if sockets.is_empty() { + return owners; + } + let wanted: std::collections::BTreeSet = sockets.iter().map(|(_, i)| *i).collect(); + for p in procs.iter().filter(|p| !p.kthread) { + let Ok(fds) = fs::read_dir(proc_dir.join(format!("{}/fd", p.pid))) else { continue }; + for fd in fds.flatten() { + let Ok(target) = fs::read_link(fd.path()) else { continue }; + let target = target.to_string_lossy(); + let inode = target + .strip_prefix("socket:[") + .and_then(|s| s.strip_suffix(']')) + .and_then(|s| s.parse::().ok()); + if let Some(inode) = inode.filter(|i| wanted.contains(i)) { + // A socket shared by a system parent and a user child counts as the user's. + let system = + owners.get(&inode).is_some_and(|c: &String| super::is_system_process(c)); + if !owners.contains_key(&inode) || system { + owners.insert(inode, p.comm.clone()); + } + } + } + if owners.len() == wanted.len() && owners.values().all(|c| !super::is_system_process(c)) + { + break; + } + } + owners + } + + fn files(dir: &Path) -> Vec<(String, std::path::PathBuf)> { + let Ok(entries) = fs::read_dir(dir) else { return Vec::new() }; + let mut out: Vec<(String, std::path::PathBuf)> = entries + .flatten() + .filter(|e| e.file_type().is_ok_and(|t| !t.is_dir())) + .filter_map(|e| Some((e.file_name().to_str()?.to_owned(), e.path()))) + .collect(); + out.sort(); + out + } + + fn timer_units(dir: &Path, kind: &str, out: &mut Vec) { + let mut dirs = vec![dir.to_path_buf()]; + if let Ok(entries) = fs::read_dir(dir) { + dirs.extend( + entries + .flatten() + .filter(|e| e.file_name().to_string_lossy().ends_with(".wants")) + .map(|e| e.path()), + ); + } + for d in dirs { + for (name, path) in files(&d) { + // A masked unit links to /dev/null: it never runs. + let masked = fs::read_link(&path).is_ok_and(|t| t == Path::new("/dev/null")); + if name.ends_with(".timer") && !masked { + out.push(format!("{kind}:{name}")); + } + } + } + } + + /// Every timer under `root`, as `KIND:NAME`. + pub(super) fn timers(root: &Path) -> Vec { + let mut out = Vec::new(); + for spool in ["var/spool/cron/crontabs", "var/spool/cron"] { + for (name, path) in files(&root.join(spool)) { + if fs::read_to_string(&path).is_ok_and(|t| crontab_has_jobs(&t)) { + out.push(format!("cron:{name}")); + } + } + } + for (name, path) in files(&root.join("etc/cron.d")) { + if fs::read_to_string(&path).is_ok_and(|t| crontab_has_jobs(&t)) { + out.push(format!("cron.d:{name}")); + } + } + for spool in ["var/spool/cron/atjobs", "var/spool/at"] { + for (name, _) in files(&root.join(spool)) { + if !name.starts_with('.') { + out.push(format!("at:{name}")); + } + } + } + timer_units(&root.join("run/systemd/transient"), "transient", &mut out); + // `systemd-run --user --on-calendar=...`: each user manager's own transient units. + if let Ok(users) = fs::read_dir(root.join("run/user")) { + for user in users.flatten() { + timer_units(&user.path().join("systemd/transient"), "transient", &mut out); + } + } + timer_units(&root.join("etc/systemd/system"), "systemd", &mut out); + timer_units(&root.join("root/.config/systemd/user"), "user", &mut out); + if let Ok(homes) = fs::read_dir(root.join("home")) { + for home in homes.flatten() { + timer_units(&home.path().join(".config/systemd/user"), "user", &mut out); + } + } + out.sort(); + out.dedup(); + out + } +} + +#[cfg(test)] +#[path = "workload_tests.rs"] +mod tests; diff --git a/cmux-tui/crates/cmux-host/src/cloud/workload_tests.rs b/cmux-tui/crates/cmux-host/src/cloud/workload_tests.rs new file mode 100644 index 000000000000..a5f8032c55d8 --- /dev/null +++ b/cmux-tui/crates/cmux-host/src/cloud/workload_tests.rs @@ -0,0 +1,283 @@ +use super::*; + +fn proc(pid: u32, comm: &str, ticks: u64) -> ProcStat { + ProcStat { pid, comm: comm.to_owned(), start: 1_000 + u64::from(pid), ticks, kthread: false } +} + +fn snap(mono_ms: u64, procs: Vec) -> Snapshot { + Snapshot { mono_ms, clk_tck: 100, procs, listeners: Vec::new(), timers: Vec::new() } +} + +#[test] +fn stat_reads_comm_with_spaces_and_parens_ticks_and_kernel_threads() { + // pid (comm) state ppid pgrp session tty tpgid flags minflt cminflt majflt cmajflt + // utime stime cutime cstime prio nice threads itreal starttime ... + let text = "42 (a (b) c) S 1 42 42 0 -1 4194560 10 0 0 0 7 3 20 10 20 0 1 0 5555 0 0"; + let p = parse_stat(42, text).unwrap(); + assert_eq!(p.comm, "a (b) c"); + assert_eq!(p.ticks, 40, "utime + stime + reaped children's"); + assert_eq!(p.start, 5555); + assert!(!p.kthread); + let kthread = "2 (kthreadd) S 0 0 0 0 -1 2129984 0 0 0 0 0 0 0 0 20 0 1 0 1 0 0"; + assert!(parse_stat(2, kthread).unwrap().kthread); + assert!(parse_stat(3, "3 (x) S").is_none(), "a short line does not read"); +} + +#[test] +fn listeners_are_listen_state_sockets_by_port() { + let tcp = " sl local_address rem_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode\n\ + 0: 0100007F:0BB8 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 111 1 0\n\ + 1: 00000000:0016 00000000:0000 0A 00000000:00000000 00:00000000 00000000 0 0 222 1 0\n\ + 2: 0100007F:0BB8 0100007F:D431 01 00000000:00000000 00:00000000 00000000 1000 0 333 1 0\n"; + assert_eq!( + parse_listeners(tcp), + vec![(3000, 111), (22, 222)], + "an established socket is not one" + ); +} + +#[test] +fn crontab_jobs_skip_comments_blanks_and_assignments() { + assert!(!crontab_has_jobs("# m h dom mon dow command\n\nSHELL=/bin/sh\nMAILTO=\"\"\n")); + assert!(crontab_has_jobs("SHELL=/bin/sh\n*/5 * * * * /home/cmux/backup.sh\n")); + assert!(crontab_has_jobs("@reboot cd /srv && ./run\n")); +} + +#[test] +fn the_first_probe_is_unknown_and_cpu_above_two_percent_is_busy() { + let mut t = Tracker::default(); + assert!( + t.observe(&snap(0, vec![proc(10, "node", 0)]), 1).is_none(), + "CPU use needs two probes" + ); + // 30 s at 100 ticks/s: 2% is 60 ticks. 61 is busy, 60 is not. + let w = t.observe(&snap(30_000, vec![proc(10, "node", 61), proc(11, "cargo", 0)]), 2).unwrap(); + assert_eq!(w.busy_processes, ["node"]); + assert_eq!(w.last_busy_at, Some(2)); + let w = t.observe(&snap(60_000, vec![proc(10, "node", 121)]), 3).unwrap(); + assert!(!w.busy(), "60 ticks in 30 s is 2%, not above it"); + assert_eq!(w.last_busy_at, Some(2), "the idle period starts at the last busy probe"); +} + +#[test] +fn a_clock_stepped_back_never_moves_the_last_busy_time_earlier() { + let mut t = Tracker::default(); + t.observe(&snap(0, vec![proc(10, "node", 0)]), 1); + t.observe(&snap(30_000, vec![proc(10, "node", 100)]), 5_000); + let w = t.observe(&snap(60_000, vec![proc(10, "node", 200)]), 4_000).unwrap(); + assert_eq!(w.last_busy_at, Some(5_000)); +} + +#[test] +fn a_process_new_since_the_last_probe_counts_all_its_ticks_and_a_reused_pid_is_new() { + let mut t = Tracker::default(); + t.observe(&snap(0, vec![proc(10, "sh", 500)]), 1); + // pid 10 again, but a different process (another start time) with 100 ticks of its own. + let reused = ProcStat { start: 9_999, ..proc(10, "rustc", 100) }; + let w = t.observe(&snap(30_000, vec![reused, proc(12, "cc1", 100)]), 2).unwrap(); + assert_eq!(w.busy_processes, ["cc1", "rustc"]); +} + +#[test] +fn system_processes_and_kernel_threads_never_count() { + let mut t = Tracker::default(); + let busy = |pid, comm: &str| proc(pid, comm, 10_000); + t.observe(&snap(0, vec![]), 1); + let mut kworker = busy(5, "kworker/0:1"); + kworker.kthread = true; + let procs = vec![ + busy(1, "systemd"), + busy(2, "systemd-journal"), + busy(3, "sshd"), + busy(4, "cmux-host"), + kworker, + busy(6, "acpmux"), + busy(7, "cmux-tui"), + busy(8, "__terminal-host"), + busy(9, "cron"), + busy(10, "Xtigervnc"), + ]; + let w = t.observe(&snap(30_000, procs), 2).unwrap(); + assert!(!w.busy(), "{w:?}"); + assert_eq!(w.last_busy_at, None); +} + +#[test] +fn a_user_listener_or_an_unowned_one_counts_a_system_one_does_not() { + let mut t = Tracker::default(); + let mut s = snap(0, vec![]); + t.observe(&s, 1); + s.mono_ms = 30_000; + s.listeners = vec![ + (22, Some("sshd".into())), + (6080, Some("websockify".into())), + (3000, Some("node".into())), + (5432, None), + (3000, Some("node".into())), + ]; + let w = t.observe(&s, 2).unwrap(); + assert_eq!(w.listening_ports, [3000, 5432]); + assert!(w.busy()); +} + +#[test] +fn user_timers_count_and_the_systems_do_not() { + let mut t = Tracker::default(); + let mut s = snap(0, vec![]); + t.observe(&s, 1); + s.mono_ms = 30_000; + s.timers = [ + "systemd:apt-daily.timer", + "systemd:cmux-machine-health.timer", + "cron.d:e2scrub_all", + "cron:cmux", + "cron.d:nightly-backup", + "at:a00001019abc", + "transient:run-u42.timer", + "user:sync.timer", + "systemd:deploy.timer", + ] + .map(String::from) + .to_vec(); + let w = t.observe(&s, 2).unwrap(); + assert_eq!( + w.timers, + [ + "at:a00001019abc", + "cron.d:nightly-backup", + "cron:cmux", + "systemd:deploy.timer", + "transient:run-u42.timer", + "user:sync.timer" + ] + ); +} + +#[test] +fn the_report_block_has_counts_and_at_most_eight_names() { + let w = Workload { + busy_processes: (0..10).map(|i| format!("p{i}")).collect(), + listening_ports: vec![3000], + timers: vec![], + last_busy_at: Some(7), + }; + let j = w.to_json(); + assert_eq!(j["busy_processes"], 10); + assert_eq!(j["process_names"].as_array().unwrap().len(), NAMES_MAX); + assert_eq!(j["listening_ports"], 1); + assert_eq!(j["ports"], json!([3000])); + assert_eq!(j["timers"], 0); + assert_eq!(j["last_busy_at"], 7); +} + +#[test] +fn the_key_ignores_which_processes_while_busy_and_holds_the_quiet_time() { + let a = Workload { + busy_processes: vec!["node".into()], + last_busy_at: Some(1), + ..Workload::default() + }; + let b = Workload { + busy_processes: vec!["cargo".into()], + last_busy_at: Some(2), + ..Workload::default() + }; + assert_eq!(a.key(), b.key(), "busy either way: no new report"); + let quiet = Workload { last_busy_at: Some(2), ..Workload::default() }; + assert_ne!(quiet.key(), b.key()); + let later = Workload { last_busy_at: Some(3), ..Workload::default() }; + assert_ne!(quiet.key(), later.key(), "a new quiet period is a new report"); +} + +#[cfg(unix)] +#[test] +fn the_reader_finds_processes_listener_owners_and_timers_under_a_root() { + use std::fs; + use std::os::unix::fs::symlink; + let root = tempfile::tempdir().unwrap(); + let r = root.path(); + let stat = |pid: u32, comm: &str| { + let dir = r.join(format!("proc/{pid}/fd")); + fs::create_dir_all(&dir).unwrap(); + fs::write( + r.join(format!("proc/{pid}/stat")), + format!("{pid} ({comm}) S 1 1 1 0 -1 4194560 0 0 0 0 5 5 0 0 20 0 1 0 77 0 0"), + ) + .unwrap(); + dir + }; + stat(1, "systemd"); + symlink("socket:[222]", stat(100, "sshd").join("3")).unwrap(); + symlink("socket:[111]", stat(200, "node").join("19")).unwrap(); + symlink("/dev/pts/0", stat(200, "node").join("0")).unwrap(); + fs::create_dir_all(r.join("proc/net")).unwrap(); + fs::write( + r.join("proc/net/tcp"), + " sl local_address rem_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode\n\ + 0: 0100007F:0BB8 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 111 1 0\n\ + 1: 00000000:0016 00000000:0000 0A 00000000:00000000 00:00000000 00000000 0 0 222 1 0\n\ + 2: 00000000:1F90 00000000:0000 0A 00000000:00000000 00:00000000 00000000 0 0 333 1 0\n", + ) + .unwrap(); + fs::create_dir_all(r.join("var/spool/cron/crontabs")).unwrap(); + fs::write(r.join("var/spool/cron/crontabs/cmux"), "*/5 * * * * true\n").unwrap(); + fs::write(r.join("var/spool/cron/crontabs/root"), "# nothing\n").unwrap(); + fs::create_dir_all(r.join("etc/systemd/system/timers.target.wants")).unwrap(); + fs::write(r.join("etc/systemd/system/cmux-machine-health.timer"), "").unwrap(); + symlink( + "/lib/systemd/system/apt-daily.timer", + r.join("etc/systemd/system/timers.target.wants/apt-daily.timer"), + ) + .unwrap(); + symlink("/dev/null", r.join("etc/systemd/system/masked.timer")).unwrap(); + fs::create_dir_all(r.join("home/cmux/.config/systemd/user")).unwrap(); + fs::write(r.join("home/cmux/.config/systemd/user/sync.timer"), "").unwrap(); + fs::create_dir_all(r.join("run/user/1000/systemd/transient")).unwrap(); + fs::write(r.join("run/user/1000/systemd/transient/run-r9.timer"), "").unwrap(); + fs::create_dir_all(r.join("var/spool/cron/atjobs")).unwrap(); + fs::write(r.join("var/spool/cron/atjobs/.SEQ"), "1").unwrap(); + fs::write(r.join("var/spool/cron/atjobs/a0000101"), "").unwrap(); + + let s = read_snapshot(r, 5).unwrap(); + assert_eq!(s.procs.len(), 3); + let mut listeners = s.listeners.clone(); + listeners.sort(); + assert_eq!( + listeners, + [(22, Some("sshd".to_owned())), (3000, Some("node".to_owned())), (8080, None)] + ); + assert_eq!( + s.timers, + [ + "at:a0000101", + "cron:cmux", + "systemd:apt-daily.timer", + "systemd:cmux-machine-health.timer", + "transient:run-r9.timer", + "user:sync.timer" + ] + ); + let mut t = Tracker::default(); + t.observe(&s, 1); + let w = t.observe(&Snapshot { mono_ms: 30_005, ..s }, 2).unwrap(); + assert_eq!(w.listening_ports, [3000, 8080]); + assert_eq!(w.timers, ["at:a0000101", "cron:cmux", "transient:run-r9.timer", "user:sync.timer"]); + assert!(w.busy_processes.is_empty()); + + assert!(read_snapshot(&r.join("nowhere"), 5).is_none(), "no /proc: unknown"); + + // A missing IPv6 table is no listener; a table that exists but does not read is unknown. + assert!(!r.join("proc/net/tcp6").exists()); + fs::create_dir_all(r.join("proc/net/tcp6")).unwrap(); + assert!(read_snapshot(r, 5).is_none(), "an unreadable listener table: unknown"); + fs::remove_dir(r.join("proc/net/tcp6")).unwrap(); + // IPv4's table is required: without it a listener could hide. + let tcp = fs::read_to_string(r.join("proc/net/tcp")).unwrap(); + fs::rename(r.join("proc/net/tcp"), r.join("proc/net/tcp6")).unwrap(); + assert!(read_snapshot(r, 5).is_none(), "no IPv4 table: unknown"); + fs::remove_file(r.join("proc/net/tcp6")).unwrap(); + fs::write(r.join("proc/net/tcp"), tcp).unwrap(); + // A process whose stat exists but does not read is unknown, not gone. + fs::create_dir_all(r.join("proc/300/stat")).unwrap(); + assert!(read_snapshot(r, 5).is_none(), "an unreadable process: unknown"); +}