jason/socktop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(agent): GPU worker thread, async journalctl, correctness + cache fixes

Browse Source 
Lightweight:
- GPU collection moves to a dedicated worker thread that owns the gfxinfo
  handle for the process lifetime. gfxinfo's active_gpu() runs a full NVML
  init/teardown (~20ms, blocking) and we were paying it on the async
  runtime for every collect — measured at ~80% of the agent's entire
  active CPU on a GPU machine. The handle holds Rc<Nvml> (not Send), so a
  thread + mpsc/oneshot channel pair confines it; a zero-total-VRAM reply
  is treated as a dead session (driver reload) and re-probed.
- journalctl now runs via tokio::process instead of blocking one of the
  two runtime workers for the duration of the subprocess.
- TtlCell (state.rs) replaces the four hand-rolled static TTL caches; a
  cached negative result now counts as fresh, so hosts with no matching
  temp sensor or GPU stop rescanning every request. Single lock+clone on
  the GPU cache hit path (was two).

Correctness:
- Process/child CPU times are now microseconds as documented; they were
  milliseconds, rendering 1000x too small next to (correct) thread times.
- Non-Linux per-process CPU%% clamps AFTER dividing by core count; a
  4-cores-busy process on an 8-core box reported 12.5% instead of 50%.
- Journal timestamps are real RFC 3339 UTC plus an additive timestamp_us
  field (sorting is now numeric); the old strings were Debug-formatted
  SystemTime mangled by string replace.
- Partition detection uses /sys/block on Linux: whole-disk filesystems on
  names like nvme0n1 or zram1 are no longer misclassified as partitions.
  One shared parent_disk_name() replaces two inline copies.
- New sampled_at_ms on the metrics payload (additive) records when the
  snapshot was actually collected, so clients can compute exact rates
  across the agent's TTL cache.

Security/robustness:
- key.pem is created 0600 (was umask default 0644, world-readable) and
  pre-1.51 keys are tightened on startup.
- Per-PID detail/journal caches now evict (60s max age, 64 entries max);
  they previously grew without bound under PID-walking clients.
- The two per-PID ws handlers collapse into one generic helper.
- /proc/<pid>/stat parsing unified in one comm-safe module.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jasonwitty
2026-08-19 14:43:37 -07:00
parent fbc788c799
commit e4b0d9b9f6
13 changed files with 511 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files
socktop/src/app.rs
+28 -5
View File
@@ -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<u64>,
rx_hist: VecDeque<u64>,
tx_hist: VecDeque<u64>,
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::<u64>();
let tx_total = m.networks.iter().map(|n| n.transmitted).sum::<u64>();
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);
socktop/src/ui/cpu.rs
+1
View File
@@ -589,6 +589,7 @@ mod render_tests {
fn fake_metrics(cores: Vec<f32>) -> Metrics {
Metrics {
sampled_at_ms: None,
cpu_total: 0.0,
cpu_per_core: cores,
mem_total: 1024,
socktop/src/ui/gpu.rs
+1
View File
@@ -249,6 +249,7 @@ mod render_tests {
fn metrics(gpus: Option<Vec<GpuInfo>>) -> Metrics {
Metrics {
sampled_at_ms: None,
cpu_total: 0.0,
cpu_per_core: vec![],
mem_total: 1024,
socktop/src/ui/processes.rs
+1
View File
@@ -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,
socktop_agent/Cargo.toml
+4 -5
View File
@@ -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"] }
socktop_agent/src/gpu.rs
+72 -17
View File
@@ -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<Nvml>` (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<Vec<GpuMetrics>, Box<dyn std::error::Error>> {
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<Vec<GpuMetrics>> {
worker::collect().await
}
#[cfg(not(feature = "gpu"))]
pub fn collect_all_gpus() -> Result<Vec<GpuMetrics>, Box<dyn std::error::Error>> {
// GPU support not available on this platform
Ok(vec![])
pub async fn collect_all_gpus() -> Option<Vec<GpuMetrics>> {
None
}
#[cfg(feature = "gpu")]
mod worker {
use super::GpuMetrics;
use once_cell::sync::OnceCell;
use std::sync::mpsc;
type Reply = tokio::sync::oneshot::Sender<Option<Vec<GpuMetrics>>>;
static TX: OnceCell<mpsc::Sender<Reply>> = OnceCell::new();
pub async fn collect() -> Option<Vec<GpuMetrics>> {
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<Reply> {
let (tx, rx) = mpsc::channel::<Reply>();
std::thread::Builder::new()
.name("socktop-gpu".into())
.spawn(move || run(rx))
.expect("spawn gpu worker thread");
tx
}
fn run(rx: mpsc::Receiver<Reply>) {
let mut handle: Option<Box<dyn gfxinfo::Gpu>> = 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()));
}
}
}
socktop_agent/src/metrics.rs
+242 -269
View File
@@ -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::<Vec<_>>()` 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/<pid>/stat` (and per-thread `task/<tid>/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::<u64>().ok()).unwrap_or(0);
let stime = it.next().and_then(|s| s.parse::<u64>().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<Instant>,
v: Option<f32>,
}
static TEMP: OnceCell<Mutex<TempCache>> = OnceCell::new();
static TEMP: crate::state::TtlCell<Option<f32>> = crate::state::TtlCell::new();
static GPUS: crate::state::TtlCell<Option<Vec<crate::gpu::GpuMetrics>>> =
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<Mutex<Option<Instant>>> = 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<Instant>,
v: Option<Vec<crate::gpu::GpuMetrics>>,
}
static GPUC: OnceCell<Mutex<GpuCache>> = OnceCell::new();
// Static caches for unchanging data
static HOSTNAME: OnceCell<String> = OnceCell::new();
struct NetworkNameCache {
@@ -166,54 +163,6 @@ struct NetworkNameCache {
static NETWORK_CACHE: OnceCell<Mutex<NetworkNameCache>> = OnceCell::new();
static CPU_VEC: OnceCell<Mutex<Vec<f32>>> = OnceCell::new();
fn cached_temp() -> Option<f32> {
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<f32>) {
if let Some(lock) = TEMP.get()
&& let Ok(mut c) = lock.lock()
{
c.v = v;
c.at = Some(Instant::now());
}
}
fn cached_gpus() -> Option<Vec<crate::gpu::GpuMetrics>> {
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<Vec<crate::gpu::GpuMetrics>>) {
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<digits> 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<DiskInfo> {
let ttl = StdDuration::from_millis(disks_ttl_ms());
@@ -445,19 +427,7 @@ pub async fn collect_disks(state: &AppState) -> Vec<DiskInfo> {
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<DiskInfo> {
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<DiskInfo> {
// 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<u64> {
#[cfg(target_os = "linux")]
#[inline]
fn read_proc_jiffies(pid: u32) -> Option<u64> {
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::<u64>().ok()?;
let stime = it.next()?.parse::<u64>().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<u32> {
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::<u32>()
.ok()
// Post-comm field 1 is ppid.
procstat::field(&stat, 1)?.parse::<u32>().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::<u64>()
.ok()?
procstat::field(&stat, 19)?.parse::<u64>().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<crate::types::ThreadInfo> {
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<crate::types::ThreadInfo> {
.unwrap_or("Unknown")
.to_string();
// 10 fields between state and utime (ppid..cmajflt).
let utime = it.nth(10).and_then(|s| s.parse::<u64>().ok()).unwrap_or(0);
let stime = it.next().and_then(|s| s.parse::<u64>().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<JournalResponse, String> {
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<JournalResponse, String> {
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<JournalResponse, String> {
"--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<JournalResponse, String> {
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::<u64>().ok())
.unwrap_or(0);
// Convert timestamp to ISO 8601 format
let timestamp = if let Ok(ts_micros) = timestamp_str.parse::<u64>() {
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<JournalResponse, String> {
entries.push(JournalEntry {
timestamp,
timestamp_us,
priority,
message,
unit,
@@ -1493,7 +1403,7 @@ pub fn collect_journal_entries(pid: u32) -> Result<JournalResponse, String> {
}
// 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<JournalResponse, String> {
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}");
}
}
socktop_agent/src/state.rs
+50 -1
View File
@@ -74,6 +74,55 @@ pub struct AppState {
pub cache_journal_entries: Arc<Mutex<HashMap<u32, CacheEntry<crate::types::JournalResponse>>>>,
}
/// 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<T> {
inner: std::sync::Mutex<CacheEntry<T>>,
}
impl<T: Clone> Default for TtlCell<T> {
fn default() -> Self {
Self::new()
}
}
impl<T: Clone> TtlCell<T> {
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<T> {
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<T> {
pub at: Option<Instant>,
@@ -87,7 +136,7 @@ impl<T> Default for CacheEntry<T> {
}
impl<T> CacheEntry<T> {
pub fn new() -> Self {
pub const fn new() -> Self {
Self {
at: None,
value: None,
socktop_agent/src/tls.rs
+21 -1
View File
@@ -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!(
socktop_agent/src/types.rs
+7 -1
View File
@@ -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<f32>,
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<String>, // systemd unit name
socktop_agent/src/ws.rs
+78 -72
View File
@@ -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<T, Fut>(
socket: &mut WebSocket,
cache: &Mutex<HashMap<u32, crate::state::CacheEntry<T>>>,
pid: u32,
ttl: std::time::Duration,
request_name: &str,
collect: impl FnOnce() -> Fut,
) where
T: serde::Serialize + Clone,
Fut: std::future::Future<Output = Result<T, String>>,
{
{
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::<u32>()
{
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::<u32>()
{
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,
socktop_agent/tests/process_details.rs
+2 -2
View File
@@ -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",
socktop_connector/src/types.rs
+4
View File
@@ -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<u64>,
pub cpu_total: f32,
pub cpu_per_core: Vec<f32>,
pub mem_total: u64,