diff --git a/socktop/src/app.rs b/socktop/src/app.rs index 881cf8c..b10ad21 100644 --- a/socktop/src/app.rs +++ b/socktop/src/app.rs @@ -80,6 +80,8 @@ pub struct App { // Network totals snapshot + histories of KB/s last_net_totals: Option<(u64, u64, Instant)>, + // Agent-side sample timestamp of the previous snapshot (1.51+ agents). + last_net_sampled_at_ms: Option, rx_hist: VecDeque, tx_hist: VecDeque, rx_peak: u64, @@ -180,6 +182,7 @@ impl App { cpu_hist_sum: 0, per_core_hist: PerCoreHistory::new(60), last_net_totals: None, + last_net_sampled_at_ms: None, rx_hist: VecDeque::with_capacity(600), tx_hist: VecDeque::with_capacity(600), rx_peak: 0, @@ -1249,19 +1252,39 @@ impl App { self.per_core_hist.ensure_cores(m.cpu_per_core.len()); self.per_core_hist.push_samples(&m.cpu_per_core); - // NET: sum across all ifaces, compute KB/s via elapsed time + // NET: sum across all ifaces, compute KB/s. Prefer the agent's sample + // timestamps (the agent serves TTL-cached snapshots, so client receive + // time overstates dt on a cache hit and produces a 0-then-2x sawtooth); + // fall back to the client clock against pre-1.51 agents. let now = Instant::now(); let rx_total = m.networks.iter().map(|n| n.received).sum::(); let tx_total = m.networks.iter().map(|n| n.transmitted).sum::(); let (rx_kb, tx_kb) = if let Some((prx, ptx, pts)) = self.last_net_totals { - let dt = now.duration_since(pts).as_secs_f64().max(1e-6); - let rx = ((rx_total.saturating_sub(prx)) as f64 / dt / 1024.0).round() as u64; - let tx = ((tx_total.saturating_sub(ptx)) as f64 / dt / 1024.0).round() as u64; - (rx, tx) + // None = identical agent snapshot (cache hit): repeat the previous + // rates so the timeline advances without a fake dip to zero. + let dt = match (m.sampled_at_ms, self.last_net_sampled_at_ms) { + (Some(a), Some(b)) if a == b => None, + (Some(a), Some(b)) if a > b => Some((a - b) as f64 / 1000.0), + // Agent restarted or clock stepped backwards: client clock. + _ => Some(now.duration_since(pts).as_secs_f64().max(1e-6)), + }; + match dt { + None => ( + self.rx_hist.back().copied().unwrap_or(0), + self.tx_hist.back().copied().unwrap_or(0), + ), + Some(dt) => { + let dt = dt.max(1e-6); + let rx = ((rx_total.saturating_sub(prx)) as f64 / dt / 1024.0).round() as u64; + let tx = ((tx_total.saturating_sub(ptx)) as f64 / dt / 1024.0).round() as u64; + (rx, tx) + } + } } else { (0, 0) }; self.last_net_totals = Some((rx_total, tx_total, now)); + self.last_net_sampled_at_ms = m.sampled_at_ms; push_capped(&mut self.rx_hist, rx_kb, 600); push_capped(&mut self.tx_hist, tx_kb, 600); self.rx_peak = self.rx_peak.max(rx_kb); diff --git a/socktop/src/ui/cpu.rs b/socktop/src/ui/cpu.rs index e78add0..eb10a76 100644 --- a/socktop/src/ui/cpu.rs +++ b/socktop/src/ui/cpu.rs @@ -589,6 +589,7 @@ mod render_tests { fn fake_metrics(cores: Vec) -> Metrics { Metrics { + sampled_at_ms: None, cpu_total: 0.0, cpu_per_core: cores, mem_total: 1024, diff --git a/socktop/src/ui/gpu.rs b/socktop/src/ui/gpu.rs index d246c5f..c35153a 100644 --- a/socktop/src/ui/gpu.rs +++ b/socktop/src/ui/gpu.rs @@ -249,6 +249,7 @@ mod render_tests { fn metrics(gpus: Option>) -> Metrics { Metrics { + sampled_at_ms: None, cpu_total: 0.0, cpu_per_core: vec![], mem_total: 1024, diff --git a/socktop/src/ui/processes.rs b/socktop/src/ui/processes.rs index 5ebe517..3e78bd9 100644 --- a/socktop/src/ui/processes.rs +++ b/socktop/src/ui/processes.rs @@ -860,6 +860,7 @@ mod click_tests { fn metrics() -> Metrics { Metrics { + sampled_at_ms: None, cpu_total: 0.0, cpu_per_core: vec![], mem_total: 32_000_000_000, diff --git a/socktop_agent/Cargo.toml b/socktop_agent/Cargo.toml index 6de4b95..f25c9ec 100644 --- a/socktop_agent/Cargo.toml +++ b/socktop_agent/Cargo.toml @@ -10,11 +10,10 @@ homepage = "https://github.com/jasonwitty/socktop" repository = "https://github.com/jasonwitty/socktop" [dependencies] -# Tokio: Use minimal features instead of "full" to reduce binary size -# Only include: rt-multi-thread (async runtime), net (WebSocket), sync (Mutex/RwLock), macros (#[tokio::test]) -# Excluded: io, fs, process, signal, time (not needed for this workload) -# Savings: ~200-300KB binary size, faster compile times -tokio = { version = "1", features = ["rt-multi-thread", "net", "sync", "macros"] } +# Tokio: minimal features instead of "full" to reduce binary size. +# rt-multi-thread (runtime), net (WebSocket), sync (Mutex/oneshot), +# macros (#[tokio::test]), process (async journalctl). +tokio = { version = "1", features = ["rt-multi-thread", "net", "sync", "macros", "process"] } axum = { version = "0.7", features = ["ws", "macros"] } sysinfo = { version = "0.37", features = ["network", "disk", "component"] } serde = { version = "1", features = ["derive"] } diff --git a/socktop_agent/src/gpu.rs b/socktop_agent/src/gpu.rs index 9e0eb08..7e0278a 100644 --- a/socktop_agent/src/gpu.rs +++ b/socktop_agent/src/gpu.rs @@ -1,6 +1,4 @@ // gpu.rs -#[cfg(feature = "gpu")] -use gfxinfo::active_gpu; #[derive(Debug, Clone, serde::Serialize)] pub struct GpuMetrics { @@ -10,23 +8,80 @@ pub struct GpuMetrics { pub mem_total_bytes: u64, } +/// Collect metrics for the active GPU. `None` when there is no usable GPU. +/// +/// Runs on a dedicated worker thread (see `worker`): gfxinfo's handle holds +/// an `Rc` (not `Send`), and *creating* it runs a full NVML library +/// init — ~20ms of blocking work that used to execute on the async runtime +/// for every collection. The worker owns one handle for the process lifetime, +/// so steady-state collection is just NVML queries. Measured on an RTX 5080 +/// box, re-initing per collect was ~80% of the agent's entire active CPU. #[cfg(feature = "gpu")] -pub fn collect_all_gpus() -> Result, Box> { - let gpu = active_gpu()?; // Use ? to unwrap Result - let info = gpu.info(); - - let metrics = GpuMetrics { - name: gpu.model().to_string(), - utilization_gpu_pct: info.load_pct() as u32, - mem_used_bytes: info.used_vram(), - mem_total_bytes: info.total_vram(), - }; - - Ok(vec![metrics]) +pub async fn collect_all_gpus() -> Option> { + worker::collect().await } #[cfg(not(feature = "gpu"))] -pub fn collect_all_gpus() -> Result, Box> { - // GPU support not available on this platform - Ok(vec![]) +pub async fn collect_all_gpus() -> Option> { + None +} + +#[cfg(feature = "gpu")] +mod worker { + use super::GpuMetrics; + use once_cell::sync::OnceCell; + use std::sync::mpsc; + + type Reply = tokio::sync::oneshot::Sender>>; + static TX: OnceCell> = OnceCell::new(); + + pub async fn collect() -> Option> { + let tx = TX.get_or_init(spawn); + let (reply_tx, reply_rx) = tokio::sync::oneshot::channel(); + tx.send(reply_tx).ok()?; + reply_rx.await.ok().flatten() + } + + fn spawn() -> mpsc::Sender { + let (tx, rx) = mpsc::channel::(); + std::thread::Builder::new() + .name("socktop-gpu".into()) + .spawn(move || run(rx)) + .expect("spawn gpu worker thread"); + tx + } + + fn run(rx: mpsc::Receiver) { + let mut handle: Option> = None; + // Probing failed: remember and answer None without re-initing the GPU + // stack per request. The agent's negative cache stops asking anyway. + let mut probe_failed = false; + while let Ok(reply) = rx.recv() { + if handle.is_none() && !probe_failed { + match gfxinfo::active_gpu() { + Ok(g) => handle = Some(g), + Err(_) => probe_failed = true, + } + } + let out = handle.as_ref().map(|gpu| { + let info = gpu.info(); + vec![GpuMetrics { + name: gpu.model().to_string(), + utilization_gpu_pct: info.load_pct().clamp(0, 100), + mem_used_bytes: info.used_vram(), + mem_total_bytes: info.total_vram(), + }] + }); + // A live GPU cannot report 0 total VRAM; gfxinfo returns zeros + // when the underlying session died (e.g. driver reload). Drop the + // handle so the next request re-probes. + if let Some(v) = &out + && !v.is_empty() + && v.iter().all(|g| g.mem_total_bytes == 0) + { + handle = None; + } + let _ = reply.send(out.filter(|v| !v.is_empty())); + } + } } diff --git a/socktop_agent/src/metrics.rs b/socktop_agent/src/metrics.rs index 0600fe6..002e714 100644 --- a/socktop_agent/src/metrics.rs +++ b/socktop_agent/src/metrics.rs @@ -13,49 +13,60 @@ use std::collections::HashMap; use std::fs; #[cfg(target_os = "linux")] use std::io; -use std::process::Command; use std::sync::Mutex; use std::time::Duration as StdDuration; -use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use sysinfo::{ProcessRefreshKind, ProcessesToUpdate}; #[cfg(feature = "logging")] use tracing::warn; // NOTE: CPU normalization env removed; non-Linux now always reports per-process share (0..100) as given by sysinfo. -// Read (utime, stime) in milliseconds from /proc/{pid}/stat in one go. -// Returns (0, 0) if the file can't be read. -// -// We use `rfind(')')` to step past the `comm` field, which can contain -// arbitrary characters (including spaces and parens), then index the -// post-comm fields by position. This is the same trick `read_proc_jiffies` -// uses below — `split_whitespace().collect::>()` from the start of -// the file would mis-parse process names with spaces, and also wastes an -// allocation per call. Two callers used to read this file twice (once for -// user, once for system); now it's one syscall per detailed-process record. +/// Shared parsing for `/proc//stat` (and per-thread `task//stat`). +/// +/// The second field, `comm`, can contain arbitrary bytes including spaces and +/// parentheses, so naive whitespace splitting mis-parses such names. All +/// callers step past the LAST `')'` and index the remaining space-separated +/// fields from there: 0 = state, 1 = ppid, 11 = utime, 12 = stime, +/// 19 = starttime. #[cfg(target_os = "linux")] -fn get_cpu_times_ms(pid: u32) -> (u64, u64) { +mod procstat { + /// Everything after `") "` — the post-comm fields. + pub fn after_comm(stat: &str) -> Option<&str> { + stat.get(stat.rfind(')')? + 2..) + } + pub fn field(stat: &str, n: usize) -> Option<&str> { + after_comm(stat)?.split_whitespace().nth(n) + } + /// (utime, stime) in clock ticks. + pub fn utime_stime(stat: &str) -> Option<(u64, u64)> { + let mut it = after_comm(stat)?.split_whitespace(); + let utime = it.nth(11)?.parse().ok()?; + let stime = it.next()?.parse().ok()?; + Some((utime, stime)) + } + /// One clock tick at USER_HZ=100 (universal on Linux) in microseconds. + pub const TICK_US: u64 = 10_000; +} + +// Read (utime, stime) in MICROSECONDS from /proc/{pid}/stat in one syscall. +// Returns (0, 0) if the file can't be read. Units match the wire contract +// (`DetailedProcessInfo.cpu_time_user` is documented as µs) and the thread +// records — this used to return ms, making process/child CPU times render +// 1000x too small next to thread times. +#[cfg(target_os = "linux")] +fn get_cpu_times_us(pid: u32) -> (u64, u64) { let Ok(s) = fs::read_to_string(format!("/proc/{pid}/stat")) else { return (0, 0); }; - let Some(rpar) = s.rfind(')') else { + let Some((utime, stime)) = procstat::utime_stime(&s) else { return (0, 0); }; - let Some(after) = s.get(rpar + 2..) else { - return (0, 0); - }; - let mut it = after.split_whitespace(); - // Post-comm field offsets: state, ppid, pgrp, session, tty_nr, tpgid, - // flags, minflt, cminflt, majflt, cmajflt, utime, stime, ... - // utime is offset 11; stime follows. - let utime = it.nth(11).and_then(|s| s.parse::().ok()).unwrap_or(0); - let stime = it.next().and_then(|s| s.parse::().ok()).unwrap_or(0); - // 1 tick = 10ms at 100 Hz (USER_HZ). - (utime * 10, stime * 10) + (utime * procstat::TICK_US, stime * procstat::TICK_US) } #[cfg(not(target_os = "linux"))] -fn get_cpu_times_ms(_pid: u32) -> (u64, u64) { +fn get_cpu_times_us(_pid: u32) -> (u64, u64) { (0, 0) } // Runtime toggles (read once) @@ -117,46 +128,32 @@ fn name_cache_cleanup_threshold() -> usize { }) } -// Tiny TTL caches to avoid rescanning sensors every 500ms +// Tiny TTL caches to avoid rescanning sensors every 500ms. +// +// The cached type is Option<...>: a fresh `None` means "we looked recently +// and found nothing" — machines with no matching sensor/GPU no longer rescan +// on every request, only once per TTL. const TTL: Duration = Duration::from_millis(1500); -struct TempCache { - at: Option, - v: Option, -} -static TEMP: OnceCell> = OnceCell::new(); +static TEMP: crate::state::TtlCell> = crate::state::TtlCell::new(); +static GPUS: crate::state::TtlCell>> = + crate::state::TtlCell::new(); -// Last time `state.components` was refreshed (by any caller). Both +// Gate on `state.components` refreshes (hwmon scans). Both // collect_fast_metrics and collect_disks need fresh sensor values; without -// this gate they were each doing their own `Components::refresh` on their -// own cadence, paying the hwmon syscall cost twice per polling cycle. -// 1s is short enough that disk temps stay accurate (they change slowly) and -// long enough to suppress back-to-back refreshes from concurrent endpoints. +// this they each paid the hwmon syscall cost on their own cadence. 1s keeps +// disk temps accurate (they change slowly) while suppressing back-to-back +// refreshes from concurrent endpoints. const COMPONENTS_REFRESH_TTL: Duration = Duration::from_millis(1000); -static COMPONENTS_LAST_REFRESH: OnceCell>> = OnceCell::new(); +static COMPONENTS_STAMP: crate::state::TtlCell<()> = crate::state::TtlCell::new(); -/// Refresh `state.components` only if the cached refresh timestamp is older -/// than `COMPONENTS_REFRESH_TTL`. Caller must already hold the components -/// lock. +/// Refresh `state.components` at most once per `COMPONENTS_REFRESH_TTL`. +/// Caller must already hold the components lock. fn refresh_components_if_stale(components: &mut sysinfo::Components) { - let lock = COMPONENTS_LAST_REFRESH.get_or_init(|| Mutex::new(None)); - let mut last = match lock.lock() { - Ok(g) => g, - Err(_) => return, // Poisoned — skip; values stay as-is until next call - }; - let now = Instant::now(); - let stale = last.is_none_or(|t| now.duration_since(t) >= COMPONENTS_REFRESH_TTL); - if stale { + if COMPONENTS_STAMP.claim_stale(COMPONENTS_REFRESH_TTL) { components.refresh(false); - *last = Some(now); } } -struct GpuCache { - at: Option, - v: Option>, -} -static GPUC: OnceCell> = OnceCell::new(); - // Static caches for unchanging data static HOSTNAME: OnceCell = OnceCell::new(); struct NetworkNameCache { @@ -166,54 +163,6 @@ struct NetworkNameCache { static NETWORK_CACHE: OnceCell> = OnceCell::new(); static CPU_VEC: OnceCell>> = OnceCell::new(); -fn cached_temp() -> Option { - if !temp_enabled() { - return None; - } - let now = Instant::now(); - let lock = TEMP.get_or_init(|| Mutex::new(TempCache { at: None, v: None })); - let mut c = lock.lock().ok()?; - if c.at.is_none_or(|t| now.duration_since(t) >= TTL) { - c.at = Some(now); - // caller will fill this; we just hold a slot - c.v = None; - } - c.v -} - -fn set_temp(v: Option) { - if let Some(lock) = TEMP.get() - && let Ok(mut c) = lock.lock() - { - c.v = v; - c.at = Some(Instant::now()); - } -} - -fn cached_gpus() -> Option> { - if !gpu_enabled() { - return None; - } - let now = Instant::now(); - let lock = GPUC.get_or_init(|| Mutex::new(GpuCache { at: None, v: None })); - let mut c = lock.lock().ok()?; - if c.at.is_none_or(|t| now.duration_since(t) >= TTL) { - // mark stale; caller will refresh - c.at = Some(now); - c.v = None; - } - c.v.clone() -} - -fn set_gpus(v: Option>) { - if let Some(lock) = GPUC.get() - && let Ok(mut c) = lock.lock() - { - c.v = v.clone(); - c.at = Some(Instant::now()); - } -} - // Collect only fast-changing metrics (CPU/mem/net + optional temps/gpus). pub async fn collect_fast_metrics(state: &AppState) -> Metrics { let ttl = StdDuration::from_millis(metrics_ttl_ms()); @@ -253,10 +202,13 @@ pub async fn collect_fast_metrics(state: &AppState) -> Metrics { let swap_used = sys.used_swap(); drop(sys); - // CPU temperature: only refresh sensors if cache is stale - let cpu_temp_c = if cached_temp().is_some() { - cached_temp() - } else if temp_enabled() { + // CPU temperature: only rescan sensors when the cached result (even a + // cached "no sensor found") goes stale. + let cpu_temp_c = if !temp_enabled() { + None + } else if let Some(cached) = TEMP.get_fresh(TTL) { + cached + } else { let val = { let mut components = state.components.lock().await; refresh_components_if_stale(&mut components); @@ -273,10 +225,8 @@ pub async fn collect_fast_metrics(state: &AppState) -> Metrics { } }) }; - set_temp(val); + TEMP.set(val); val - } else { - None }; // Networks with reusable name cache @@ -320,47 +270,37 @@ pub async fn collect_fast_metrics(state: &AppState) -> Metrics { cache.infos.clone() }; - // GPUs: if we already determined none exist, short-circuit (no repeated probing) - let gpus = if gpu_enabled() { - if state.gpu_checked.load(std::sync::atomic::Ordering::Acquire) - && !state.gpu_present.load(std::sync::atomic::Ordering::Relaxed) - { - None - } else if cached_gpus().is_some() { - cached_gpus() - } else { - let v = match collect_all_gpus() { - Ok(v) if !v.is_empty() => Some(v), - Ok(_) => None, - Err(_e) => { - #[cfg(feature = "logging")] - warn!("gpu collection failed: {_e}"); - None - } - }; - // First probe records presence; subsequent calls rely on cache flags. - if !state - .gpu_checked - .swap(true, std::sync::atomic::Ordering::AcqRel) - { - if v.is_some() { - state - .gpu_present - .store(true, std::sync::atomic::Ordering::Release); - } else { - state - .gpu_present - .store(false, std::sync::atomic::Ordering::Release); - } - } - set_gpus(v.clone()); - v - } - } else { + // GPUs: negative-probe cache short-circuits GPU-less hosts; otherwise the + // TTL cache answers, and only a stale miss reaches the worker thread. + let gpus = if !gpu_enabled() + || (state.gpu_checked.load(std::sync::atomic::Ordering::Acquire) + && !state.gpu_present.load(std::sync::atomic::Ordering::Relaxed)) + { None + } else if let Some(cached) = GPUS.get_fresh(TTL) { + cached + } else { + let v = collect_all_gpus().await; + // First probe records presence; subsequent calls rely on the flags. + if !state + .gpu_checked + .swap(true, std::sync::atomic::Ordering::AcqRel) + { + state + .gpu_present + .store(v.is_some(), std::sync::atomic::Ordering::Release); + } + GPUS.set(v.clone()); + v }; + let sampled_at_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let metrics = Metrics { + sampled_at_ms, cpu_total, cpu_per_core, mem_total, @@ -381,6 +321,48 @@ pub async fn collect_fast_metrics(state: &AppState) -> Metrics { metrics } +/// Best-effort parent-disk name for a partition device name: +/// "nvme0n1p1" -> "nvme0n1", "mmcblk0p2" -> "mmcblk0", "sda1" -> "sda". +/// Works with or without a "/dev/" prefix. +fn parent_disk_name(name: &str) -> &str { + if let Some(pos) = name.rfind('p') { + let suffix = &name[pos + 1..]; + if !suffix.is_empty() && suffix.chars().all(|c| c.is_ascii_digit()) { + return &name[..pos]; + } + } + name.trim_end_matches(|c: char| c.is_ascii_digit()) +} + +/// Whether a device name refers to a partition rather than a whole disk. +/// +/// On Linux, whole-disk devices are directories under /sys/block and +/// partitions are not, so "is a partition" = "not in /sys/block, but the +/// derived parent is". This gets right the cases the old name heuristic got +/// wrong: a whole-disk filesystem on nvme0n1 (ends in a digit but IS in +/// /sys/block) and zram1 (a whole device). Non-Linux keeps the heuristic. +fn is_partition_name(name: &str) -> bool { + let bare = name.strip_prefix("/dev/").unwrap_or(name); + #[cfg(target_os = "linux")] + { + let sys_block = std::path::Path::new("/sys/block"); + if sys_block.is_dir() { + return !sys_block.join(bare).is_dir() + && sys_block.join(parent_disk_name(bare)).is_dir(); + } + } + is_partition_heuristic(bare) +} + +/// Name-based fallback for platforms without /sys/block: a p marker +/// or a trailing non-zero digit. +fn is_partition_heuristic(bare: &str) -> bool { + bare.contains("p1") + || bare.contains("p2") + || bare.contains("p3") + || bare.ends_with(|c: char| c.is_ascii_digit() && c != '0') +} + // Cached disks pub async fn collect_disks(state: &AppState) -> Vec { let ttl = StdDuration::from_millis(disks_ttl_ms()); @@ -445,19 +427,7 @@ pub async fn collect_disks(state: &AppState) -> Vec { return None; } - // Determine if this is a partition - let is_partition = name.contains("p1") - || name.contains("p2") - || name.contains("p3") - || name.ends_with('1') - || name.ends_with('2') - || name.ends_with('3') - || name.ends_with('4') - || name.ends_with('5') - || name.ends_with('6') - || name.ends_with('7') - || name.ends_with('8') - || name.ends_with('9'); + let is_partition = is_partition_name(&name); // Try to find temperature for this disk let temperature = disk_temps.iter().find_map(|(key, &temp)| { @@ -491,25 +461,7 @@ pub async fn collect_disks(state: &AppState) -> Vec { for partition in &partitions { if partition.is_partition { - // Extract parent disk name - // nvme0n1p1 -> nvme0n1, sda1 -> sda, mmcblk0p1 -> mmcblk0 - let parent_name = if let Some(pos) = partition.name.rfind('p') { - // Check if character after 'p' is a digit - if partition - .name - .chars() - .nth(pos + 1) - .is_some_and(|c| c.is_ascii_digit()) - { - &partition.name[..pos] - } else { - // Handle sda1, sdb2, etc (just trim trailing digit) - partition.name.trim_end_matches(char::is_numeric) - } - } else { - // Handle sda1, sdb2, etc (just trim trailing digit) - partition.name.trim_end_matches(char::is_numeric) - }; + let parent_name = parent_disk_name(&partition.name); // Look up temperature for the PARENT disk, not the partition // Strip /dev/ prefix if present for matching @@ -553,21 +505,7 @@ pub async fn collect_disks(state: &AppState) -> Vec { // Add partitions after their parent disk for partition in partitions { if partition.is_partition { - // Find parent disk index - let parent_name = if let Some(pos) = partition.name.rfind('p') { - if partition - .name - .chars() - .nth(pos + 1) - .is_some_and(|c| c.is_ascii_digit()) - { - &partition.name[..pos] - } else { - partition.name.trim_end_matches(char::is_numeric) - } - } else { - partition.name.trim_end_matches(char::is_numeric) - }; + let parent_name = parent_disk_name(&partition.name); // Find where to insert this partition (after its parent) if let Some(parent_idx) = disks.iter().position(|d| d.name == parent_name) { @@ -619,15 +557,8 @@ fn read_total_jiffies() -> io::Result { #[cfg(target_os = "linux")] #[inline] fn read_proc_jiffies(pid: u32) -> Option { - let path = format!("/proc/{pid}/stat"); - let s = fs::read_to_string(path).ok()?; - // Find the right parenthesis that terminates comm; everything after is space-separated fields starting at "state" - let rpar = s.rfind(')')?; - let after = s.get(rpar + 2..)?; // skip ") " - let mut it = after.split_whitespace(); - // utime (14th field) is offset 11 from "state", stime (15th) is next - let utime = it.nth(11)?.parse::().ok()?; - let stime = it.next()?.parse::().ok()?; + let s = fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; + let (utime, stime) = procstat::utime_stime(&s)?; Some(utime.saturating_add(stime)) } @@ -818,9 +749,12 @@ pub async fn collect_processes_all(state: &AppState) -> ProcessesPayload { }; // Convert to percentage of total CPU capacity - // e.g., 100% on 2 cores of 8 core system = 25% total CPU - let raw = p.cpu_usage(); // This is per-core percentage - let total_cpu = raw.clamp(0.0, 100.0) / cpu_count; + // e.g., 100% on 2 cores of 8 core system = 25% total CPU. + // sysinfo reports per-core percentage which EXCEEDS 100 for + // multi-threaded processes, so clamp AFTER dividing — clamping + // first truncated e.g. 400%-on-8-cores to 12.5% instead of 50%. + let raw = p.cpu_usage(); + let total_cpu = (raw / cpu_count.max(1.0)).clamp(0.0, 100.0); proc_cache.reusable_vec.push(ProcessInfo { pid, @@ -984,15 +918,8 @@ fn proc_state_label(c: char) -> &'static str { #[cfg(target_os = "linux")] fn read_parent_pid_from_proc(pid: u32) -> Option { let stat = fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; - // Format: pid (comm) state ppid ... — comm can contain spaces/parens, - // so we step past the closing paren first. - let ppid_start = stat.rfind(')')?; - // After ") ": state, ppid, ... — ppid is the second field. - stat[ppid_start + 1..] - .split_whitespace() - .nth(1)? - .parse::() - .ok() + // Post-comm field 1 is ppid. + procstat::field(&stat, 1)?.parse::().ok() } /// Collect process information from /proc files @@ -1035,16 +962,9 @@ fn collect_process_info_from_proc( let thread_count = st.threads; let status = proc_state_label(st.state_ch).to_string(); - // Read start time from stat — comm-safe via rfind(')'). + // starttime is post-comm field 19. let start_time = if let Ok(stat) = fs::read_to_string(format!("/proc/{pid}/stat")) { - let stat_end = stat.rfind(')')?; - // After ") ": state, ppid, ..., starttime — starttime is the 20th - // post-comm field (index 19). - stat[stat_end + 1..] - .split_whitespace() - .nth(19)? - .parse::() - .ok()? + procstat::field(&stat, 19)?.parse::().ok()? } else { 0 }; @@ -1079,7 +999,7 @@ fn collect_process_info_from_proc( .map(|p| p.to_string_lossy().to_string()); // One read of /proc/{pid}/stat covers both user + system CPU times. - let (cpu_time_user, cpu_time_system) = get_cpu_times_ms(pid); + let (cpu_time_user, cpu_time_system) = get_cpu_times_us(pid); Some(DetailedProcessInfo { pid, @@ -1192,18 +1112,7 @@ fn collect_thread_info(pid: u32) -> Vec { continue; }; - // Thread/comm names can contain spaces or parens, so step past the - // last ')' before parsing post-comm fields. Post-comm offsets: - // 0: state, 1: ppid, 2: pgrp, ..., 11: utime, 12: stime - let Some(rpar) = stat_content.rfind(')') else { - continue; - }; - let Some(after) = stat_content.get(rpar + 1..) else { - continue; - }; - let mut it = after.split_whitespace(); - let status = it - .next() + let status = procstat::field(&stat_content, 0) .and_then(|s| s.chars().next()) .map(|c| match c { 'R' => "Running", @@ -1218,14 +1127,9 @@ fn collect_thread_info(pid: u32) -> Vec { .unwrap_or("Unknown") .to_string(); - // 10 fields between state and utime (ppid..cmajflt). - let utime = it.nth(10).and_then(|s| s.parse::().ok()).unwrap_or(0); - let stime = it.next().and_then(|s| s.parse::().ok()).unwrap_or(0); - - // Convert clock ticks to microseconds (assuming 100 Hz) - // 1 tick = 10ms = 10,000 microseconds - let cpu_time_user = utime * 10_000; - let cpu_time_system = stime * 10_000; + let (utime, stime) = procstat::utime_stime(&stat_content).unwrap_or((0, 0)); + let cpu_time_user = utime * procstat::TICK_US; + let cpu_time_system = stime * procstat::TICK_US; threads.push(crate::types::ThreadInfo { tid, @@ -1350,7 +1254,7 @@ pub async fn collect_process_metrics( let threads = collect_thread_info(pid); // One read of /proc/{pid}/stat covers both user + system CPU times. - let (cpu_time_user, cpu_time_system) = get_cpu_times_ms(pid); + let (cpu_time_user, cpu_time_system) = get_cpu_times_us(pid); // Now construct the detailed info without holding the lock let detailed_info = DetailedProcessInfo { @@ -1384,9 +1288,26 @@ pub async fn collect_process_metrics( }) } -/// Collect journal entries for a specific process -pub fn collect_journal_entries(pid: u32) -> Result { - let output = Command::new("journalctl") +/// Epoch microseconds -> RFC 3339 UTC for display. The old code +/// Debug-formatted a SystemTime and string-replaced it into a raw epoch +/// string that was neither ISO 8601 nor what the field documented. +fn format_journal_timestamp(timestamp_us: u64) -> String { + time::OffsetDateTime::from_unix_timestamp_nanos(timestamp_us as i128 * 1000) + .ok() + .and_then(|t| { + t.format(&time::format_description::well_known::Rfc3339) + .ok() + }) + .unwrap_or_else(|| timestamp_us.to_string()) +} + +/// Collect journal entries for a specific process. +/// +/// Async via tokio::process — journalctl can take hundreds of ms on slow +/// storage, and the old std::process call blocked one of the runtime's two +/// worker threads for the duration. +pub async fn collect_journal_entries(pid: u32) -> Result { + let output = tokio::process::Command::new("journalctl") .args([ &format!("_PID={pid}"), "--output=json", @@ -1394,6 +1315,7 @@ pub fn collect_journal_entries(pid: u32) -> Result { "--no-pager", ]) .output() + .await .map_err(|e| format!("Failed to execute journalctl: {e}"))?; if !output.status.success() { @@ -1415,27 +1337,14 @@ pub fn collect_journal_entries(pid: u32) -> Result { let json: serde_json::Value = serde_json::from_str(line).map_err(|e| format!("Failed to parse journal JSON: {e}"))?; - // Extract relevant fields - let timestamp_str = json + // __REALTIME_TIMESTAMP is epoch microseconds as a string. + let timestamp_us = json .get("__REALTIME_TIMESTAMP") .and_then(|v| v.as_str()) - .unwrap_or("0"); + .and_then(|v| v.parse::().ok()) + .unwrap_or(0); - // Convert timestamp to ISO 8601 format - let timestamp = if let Ok(ts_micros) = timestamp_str.parse::() { - let ts_secs = ts_micros / 1_000_000; - let ts_nanos = (ts_micros % 1_000_000) * 1000; - let time = SystemTime::UNIX_EPOCH - + Duration::from_secs(ts_secs) - + Duration::from_nanos(ts_nanos); - // Simple ISO 8601 format - we can improve this if needed - format!("{time:?}") - .replace("SystemTime { tv_sec: ", "") - .replace(", tv_nsec: ", ".") - .replace(" }", "") - } else { - timestamp_str.to_string() - }; + let timestamp = format_journal_timestamp(timestamp_us); let priority = match json.get("PRIORITY").and_then(|v| v.as_str()) { Some("0") => LogLevel::Emergency, @@ -1482,6 +1391,7 @@ pub fn collect_journal_entries(pid: u32) -> Result { entries.push(JournalEntry { timestamp, + timestamp_us, priority, message, unit, @@ -1493,7 +1403,7 @@ pub fn collect_journal_entries(pid: u32) -> Result { } // Sort by timestamp (newest first) - entries.sort_by(|a, b| b.timestamp.cmp(&a.timestamp)); + entries.sort_by_key(|e| std::cmp::Reverse(e.timestamp_us)); let response_timestamp = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -1510,3 +1420,66 @@ pub fn collect_journal_entries(pid: u32) -> Result { cached_at: response_timestamp, }) } + +#[cfg(test)] +mod tests { + use super::*; + + /// comm can contain spaces and parens; parsing must key off the LAST ')'. + #[cfg(target_os = "linux")] + #[test] + fn procstat_handles_hostile_comm_names() { + let stat = "1234 (weird name) (2)) R 1 2 3 4 5 6 7 8 9 10 700 800 0 0 20"; + assert_eq!(procstat::field(stat, 0), Some("R")); + assert_eq!(procstat::field(stat, 1), Some("1")); + assert_eq!(procstat::utime_stime(stat), Some((700, 800))); + } + + /// USER_HZ ticks convert to MICROSECONDS — the wire contract. This used + /// to be *10 (ms), rendering process CPU times 1000x too small next to + /// thread times. + #[cfg(target_os = "linux")] + #[test] + fn cpu_times_are_microseconds() { + assert_eq!(procstat::TICK_US, 10_000); + } + + #[test] + fn parent_disk_name_strips_partition_suffixes() { + assert_eq!(parent_disk_name("nvme0n1p1"), "nvme0n1"); + assert_eq!(parent_disk_name("nvme1n1p12"), "nvme1n1"); + assert_eq!(parent_disk_name("mmcblk0p2"), "mmcblk0"); + assert_eq!(parent_disk_name("sda1"), "sda"); + assert_eq!(parent_disk_name("/dev/nvme0n1p1"), "/dev/nvme0n1"); + // 'p' inside a word is not a partition marker. + assert_eq!(parent_disk_name("mapper/vg-lv"), "mapper/vg-lv"); + } + + /// The old heuristic flagged whole-disk names ending in a digit + /// (nvme0n1, zram1) as partitions. On Linux /sys/block decides; this + /// pins the real-machine behavior for devices every Linux box has. + #[cfg(target_os = "linux")] + #[test] + fn sys_block_devices_are_not_partitions() { + let sys_block = std::path::Path::new("/sys/block"); + if !sys_block.is_dir() { + return; // exotic environment; nothing to assert + } + for entry in std::fs::read_dir(sys_block).unwrap().flatten() { + let name = entry.file_name().to_string_lossy().into_owned(); + assert!( + !is_partition_name(&name), + "{name} is a whole disk but was flagged as a partition" + ); + } + } + + #[test] + fn journal_timestamps_are_rfc3339() { + let s = format_journal_timestamp(1_786_752_000_000_000); + assert_eq!(s, "2026-08-15T00:00:00Z"); + // Sub-second precision survives. + let s = format_journal_timestamp(1_786_752_000_123_456); + assert!(s.starts_with("2026-08-15T00:00:00.123456"), "{s}"); + } +} diff --git a/socktop_agent/src/state.rs b/socktop_agent/src/state.rs index 36ad18a..74edfda 100644 --- a/socktop_agent/src/state.rs +++ b/socktop_agent/src/state.rs @@ -74,6 +74,55 @@ pub struct AppState { pub cache_journal_entries: Arc>>>, } +/// TTL-gated value behind a std Mutex, for `static` caches on hot paths. +/// Replaces the hand-rolled TempCache/GpuCache/refresh-timestamp statics +/// that each reimplemented the same at/value pair. +pub struct TtlCell { + inner: std::sync::Mutex>, +} + +impl Default for TtlCell { + fn default() -> Self { + Self::new() + } +} + +impl TtlCell { + pub const fn new() -> Self { + Self { + inner: std::sync::Mutex::new(CacheEntry::new()), + } + } + /// The stored value, only while fresh. Poisoned lock reads as a miss. + pub fn get_fresh(&self, ttl: Duration) -> Option { + let g = self.inner.lock().ok()?; + if g.is_fresh(ttl) { + g.value.clone() + } else { + None + } + } + pub fn set(&self, v: T) { + if let Ok(mut g) = self.inner.lock() { + g.set(v); + } + } + /// True exactly once per TTL window: restamps and tells the caller to do + /// the refresh. Atomic check-and-stamp so concurrent callers don't both + /// refresh. + pub fn claim_stale(&self, ttl: Duration) -> bool { + let Ok(mut g) = self.inner.lock() else { + return false; + }; + if g.at.is_none_or(|t| t.elapsed() >= ttl) { + g.at = Some(Instant::now()); + true + } else { + false + } + } +} + #[derive(Clone, Debug)] pub struct CacheEntry { pub at: Option, @@ -87,7 +136,7 @@ impl Default for CacheEntry { } impl CacheEntry { - pub fn new() -> Self { + pub const fn new() -> Self { Self { at: None, value: None, diff --git a/socktop_agent/src/tls.rs b/socktop_agent/src/tls.rs index b224556..c747393 100644 --- a/socktop_agent/src/tls.rs +++ b/socktop_agent/src/tls.rs @@ -24,6 +24,17 @@ pub fn cert_paths() -> (PathBuf, PathBuf) { pub fn ensure_self_signed_cert() -> anyhow::Result<(PathBuf, PathBuf)> { let (cert_path, key_path) = cert_paths(); if cert_path.exists() && key_path.exists() { + // Keys generated by agents older than 1.51 were written with the + // default umask (typically 0644): tighten them on startup. + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + if let Ok(meta) = fs::metadata(&key_path) + && meta.permissions().mode() & 0o077 != 0 + { + let _ = fs::set_permissions(&key_path, fs::Permissions::from_mode(0o600)); + } + } return Ok((cert_path, key_path)); } fs::create_dir_all(cert_path.parent().unwrap())?; @@ -79,7 +90,16 @@ pub fn ensure_self_signed_cert() -> anyhow::Result<(PathBuf, PathBuf)> { let mut f = fs::File::create(&cert_path)?; f.write_all(cert_pem.as_bytes())?; - let mut k = fs::File::create(&key_path)?; + // The private key must not be world-readable (File::create honors the + // umask, which typically yields 0644). + let mut key_opts = fs::OpenOptions::new(); + key_opts.write(true).create(true).truncate(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + key_opts.mode(0o600); + } + let mut k = key_opts.open(&key_path)?; k.write_all(key_pem.as_bytes())?; println!( diff --git a/socktop_agent/src/types.rs b/socktop_agent/src/types.rs index 75e7d97..8efec08 100644 --- a/socktop_agent/src/types.rs +++ b/socktop_agent/src/types.rs @@ -30,6 +30,11 @@ pub struct ProcessInfo { #[derive(Debug, Clone, Serialize)] pub struct Metrics { + /// Epoch ms when this snapshot was actually collected. The agent serves + /// TTL-cached snapshots, so the client needs the AGENT's sample time to + /// compute rates — measuring against client receive time turned cache + /// hits into a 0-then-2x sawtooth in the network graphs. + pub sampled_at_ms: u64, pub cpu_total: f32, pub cpu_per_core: Vec, pub mem_total: u64, @@ -93,7 +98,8 @@ pub struct ProcessMetricsResponse { #[derive(Debug, Clone, Serialize)] pub struct JournalEntry { - pub timestamp: String, // ISO 8601 formatted timestamp + pub timestamp: String, // RFC 3339 UTC, for display + pub timestamp_us: u64, // epoch microseconds, for sorting/formatting pub priority: LogLevel, pub message: String, pub unit: Option, // systemd unit name diff --git a/socktop_agent/src/ws.rs b/socktop_agent/src/ws.rs index a02f15f..54e0092 100644 --- a/socktop_agent/src/ws.rs +++ b/socktop_agent/src/ws.rs @@ -50,6 +50,66 @@ pub async fn ws_handler( ws.on_upgrade(move |socket| handle_socket(socket, state)) } +/// Per-PID cache limits: entries older than MAX_AGE are swept on every +/// insert and the map is capped at MAX_ENTRIES (oldest evicted first), so a +/// client walking PIDs cannot grow agent memory without bound. +const PER_PID_CACHE_MAX_AGE: std::time::Duration = std::time::Duration::from_secs(60); +const PER_PID_CACHE_MAX_ENTRIES: usize = 64; + +/// Serve a per-PID request from a TTL cache, collecting on miss. One home +/// for the logic that get_process_metrics and get_journal_entries used to +/// duplicate ~50 lines apiece. +async fn respond_per_pid_cached( + socket: &mut WebSocket, + cache: &Mutex>>, + pid: u32, + ttl: std::time::Duration, + request_name: &str, + collect: impl FnOnce() -> Fut, +) where + T: serde::Serialize + Clone, + Fut: std::future::Future>, +{ + { + let cache = cache.lock().await; + if let Some(entry) = cache.get(&pid) + && entry.is_fresh(ttl) + && let Some(v) = entry.get() + { + let _ = send_json(socket, v).await; + return; + } + } + match collect().await { + Ok(resp) => { + { + let mut cache = cache.lock().await; + cache.retain(|_, e| e.at.is_some_and(|t| t.elapsed() < PER_PID_CACHE_MAX_AGE)); + while cache.len() >= PER_PID_CACHE_MAX_ENTRIES { + let oldest = cache.iter().min_by_key(|(_, e)| e.at).map(|(k, _)| *k); + match oldest { + Some(k) => cache.remove(&k), + None => break, + }; + } + cache + .entry(pid) + .or_insert_with(crate::state::CacheEntry::new) + .set(resp.clone()); + } + let _ = send_json(socket, &resp).await; + } + Err(err) => { + let error_response = serde_json::json!({ + "error": err, + "request": request_name, + "pid": pid + }); + let _ = send_json(socket, &error_response).await; + } + } +} + async fn handle_socket(mut socket: WebSocket, state: AppState) { state .client_count @@ -124,84 +184,30 @@ async fn handle_socket(mut socket: WebSocket, state: AppState) { if let Some(pid_str) = text.strip_prefix("get_process_metrics:") && let Ok(pid) = pid_str.parse::() { - let ttl = std::time::Duration::from_millis(250); // 250ms TTL - - // Check cache first - { - let cache = state.cache_process_metrics.lock().await; - if let Some(entry) = cache.get(&pid) - && entry.is_fresh(ttl) - && let Some(cached_response) = entry.get() - { - let _ = send_json(&mut socket, cached_response).await; - continue; - } - } - - // Collect fresh data - match crate::metrics::collect_process_metrics(pid, &state).await { - Ok(response) => { - // Cache the response - { - let mut cache = state.cache_process_metrics.lock().await; - cache - .entry(pid) - .or_insert_with(crate::state::CacheEntry::new) - .set(response.clone()); - } - let _ = send_json(&mut socket, &response).await; - } - Err(err) => { - let error_response = serde_json::json!({ - "error": err, - "request": "get_process_metrics", - "pid": pid - }); - let _ = send_json(&mut socket, &error_response).await; - } - } + respond_per_pid_cached( + &mut socket, + &state.cache_process_metrics, + pid, + std::time::Duration::from_millis(250), + "get_process_metrics", + || crate::metrics::collect_process_metrics(pid, &state), + ) + .await; } } Message::Text(ref text) if text.starts_with("get_journal_entries:") => { if let Some(pid_str) = text.strip_prefix("get_journal_entries:") && let Ok(pid) = pid_str.parse::() { - let ttl = std::time::Duration::from_secs(1); // 1s TTL - - // Check cache first - { - let cache = state.cache_journal_entries.lock().await; - if let Some(entry) = cache.get(&pid) - && entry.is_fresh(ttl) - && let Some(cached_response) = entry.get() - { - let _ = send_json(&mut socket, cached_response).await; - continue; - } - } - - // Collect fresh data - match crate::metrics::collect_journal_entries(pid) { - Ok(response) => { - // Cache the response - { - let mut cache = state.cache_journal_entries.lock().await; - cache - .entry(pid) - .or_insert_with(crate::state::CacheEntry::new) - .set(response.clone()); - } - let _ = send_json(&mut socket, &response).await; - } - Err(err) => { - let error_response = serde_json::json!({ - "error": err, - "request": "get_journal_entries", - "pid": pid - }); - let _ = send_json(&mut socket, &error_response).await; - } - } + respond_per_pid_cached( + &mut socket, + &state.cache_journal_entries, + pid, + std::time::Duration::from_secs(1), + "get_journal_entries", + || crate::metrics::collect_journal_entries(pid), + ) + .await; } } Message::Close(_) => break, diff --git a/socktop_agent/tests/process_details.rs b/socktop_agent/tests/process_details.rs index dfc52c3..b73d5d6 100644 --- a/socktop_agent/tests/process_details.rs +++ b/socktop_agent/tests/process_details.rs @@ -33,7 +33,7 @@ async fn test_collect_journal_entries_self() { // Test collecting journal entries for our own process let pid = process::id(); - match collect_journal_entries(pid) { + match collect_journal_entries(pid).await { Ok(response) => { assert!(response.cached_at > 0); println!( @@ -74,7 +74,7 @@ async fn test_collect_journal_entries_invalid_pid() { // Test with an invalid PID - journalctl might still return empty results let invalid_pid = 999999; - match collect_journal_entries(invalid_pid) { + match collect_journal_entries(invalid_pid).await { Ok(response) => { println!( "✓ Journal query completed for invalid PID {} (empty result expected): {} entries", diff --git a/socktop_connector/src/types.rs b/socktop_connector/src/types.rs index 47df1c0..c2ecff5 100644 --- a/socktop_connector/src/types.rs +++ b/socktop_connector/src/types.rs @@ -54,6 +54,10 @@ pub struct GpuInfo { #[derive(Debug, Clone, Deserialize, Serialize)] pub struct Metrics { + /// Epoch ms when the agent actually collected this snapshot (agents may + /// serve TTL-cached data). Absent on agents older than 1.51. + #[serde(default)] + pub sampled_at_ms: Option, pub cpu_total: f32, pub cpu_per_core: Vec, pub mem_total: u64,