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

Major refactor, additional comments, performance improvements, idle performance improvements, access token, port specification

Browse Source 
Release highlights

Introduced split client/agent architecture with a ratatui-based TUI and a lightweight WebSocket agent.
Added adaptive (idle-aware) sampler: agent samples fast only when clients are connected; sleeps when idle.
Implemented metrics JSON caching for instant ws replies; cold-start does one-off collection.
Port configuration: --port/-p, positional PORT, or SOCKTOP_PORT env (default 3000).
Optional token auth: SOCKTOP_TOKEN on agent, ws://HOST:PORT/ws?token=VALUE in client.
Logging via tracing with RUST_LOG control.
CI workflow (fmt, clippy, build) for Linux and Windows.
Systemd unit example for always-on agent.
TUI features

CPU: overall sparkline + per-core history with trend arrows and color thresholds.
Memory/Swap gauges with humanized labels.
Disks panel with per-device usage and icons.
Network download/upload sparklines (KB/s) with peak tracking.
Top processes table (PID, name, CPU%, mem, mem%).
Header with hostname and CPU temperature indicator.
Agent changes

sysinfo 0.36.1 targeted refresh: refresh_cpu_all, refresh_memory, refresh_processes_specifics(ProcessesToUpdate::All, ProcessRefreshKind::new().with_cpu().with_memory(), true).
WebSocket handler: client counting with wake notifications, cold-start handling, proper Response returns.
Sampler uses MissedTickBehavior::Skip to avoid catch-up bursts.
Docs

README updates: running instructions, port configuration, optional token auth, platform notes, example JSON.
Added socktop-agent.service systemd unit.
Platform notes

Linux (AMD/Intel) supported; tested on AMD, targeting Intel next.
Raspberry Pi supported (availability of temps varies by model).
Windows builds/run; CPU temperature may be unavailable (shows N/A).
Known/next

Roadmap includes configurable refresh interval, TUI filtering/sorting, TLS/WSS, and export to file.
Add Context...
README.md
This commit is contained in:
jasonwitty
2025-08-08 12:41:32 -07:00
parent 1c2415bc1b
commit 100434fc3c
26 changed files with 1371 additions and 713 deletions
Download Patch File Download Diff File Expand all files Collapse all files
socktop_agent/src/main.rs
+103 -184
View File
@@ -1,217 +1,136 @@
use axum::{
extract::{
ws::{Message, WebSocket, WebSocketUpgrade},
State,
},
response::IntoResponse,
routing::get,
Router,
};
use futures_util::stream::StreamExt;
use serde::Serialize;
use std::{collections::HashMap, net::SocketAddr, sync::Arc};
//! socktop agent entrypoint: sets up sysinfo handles, launches a sampler,
//! and serves a WebSocket endpoint at /ws.
mod metrics;
mod sampler;
mod state;
mod ws;
mod types;
use axum::{routing::get, Router};
use std::{collections::HashMap, net::SocketAddr, sync::Arc, time::Duration, sync::atomic::AtomicUsize};
use sysinfo::{
Components, CpuRefreshKind, Disks, MemoryRefreshKind, Networks, ProcessRefreshKind, RefreshKind,
System,
};
use tokio::sync::Mutex;
use tokio::sync::{Mutex, RwLock, Notify};
use tracing_subscriber::EnvFilter;
// ---------- Data types sent to the client ----------
#[derive(Debug, Serialize, Clone)]
struct ProcessInfo {
pid: u32,
name: String,
cpu_usage: f32,
mem_bytes: u64,
}
#[derive(Debug, Serialize, Clone)]
struct DiskInfo {
name: String,
total: u64,
available: u64,
}
#[derive(Debug, Serialize, Clone)]
struct NetworkInfo {
name: String,
// cumulative totals since the agent started (client should diff to get rates)
received: u64,
transmitted: u64,
}
#[derive(Debug, Serialize, Clone)]
struct Metrics {
cpu_total: f32,
cpu_per_core: Vec<f32>,
mem_total: u64,
mem_used: u64,
swap_total: u64,
swap_used: u64,
process_count: usize,
hostname: String,
cpu_temp_c: Option<f32>,
disks: Vec<DiskInfo>,
networks: Vec<NetworkInfo>,
top_processes: Vec<ProcessInfo>,
}
// ---------- Shared state ----------
type SharedSystem = Arc<Mutex<System>>;
type SharedNetworks = Arc<Mutex<Networks>>;
type SharedTotals = Arc<Mutex<HashMap<String, (u64, u64)>>>; // iface -> (rx_total, tx_total)
#[derive(Clone)]
struct AppState {
sys: SharedSystem,
nets: SharedNetworks,
net_totals: SharedTotals,
}
use state::{AppState, SharedTotals};
use sampler::spawn_sampler;
use ws::ws_handler;
#[tokio::main]
async fn main() {
// sysinfo 0.36: build specifics
// Init logging; configure with RUST_LOG (e.g., RUST_LOG=info).
tracing_subscriber::fmt()
.with_env_filter(EnvFilter::from_default_env())
.with_target(false)
.compact()
.init();
// sysinfo build specifics (scopes what refresh_all() will touch internally)
let refresh_kind = RefreshKind::nothing()
.with_cpu(CpuRefreshKind::everything())
.with_memory(MemoryRefreshKind::everything())
.with_processes(ProcessRefreshKind::everything());
// Initialize sysinfo handles once and keep them alive
let mut sys = System::new_with_specifics(refresh_kind);
sys.refresh_all();
// Keep Networks alive across requests so received()/transmitted() deltas work
let mut nets = Networks::new();
nets.refresh(true);
let shared = Arc::new(Mutex::new(sys));
let shared_nets = Arc::new(Mutex::new(nets));
let net_totals: SharedTotals = Arc::new(Mutex::new(HashMap::new()));
let mut components = Components::new();
components.refresh(true);
let app = Router::new()
.route("/ws", get(ws_handler))
.with_state(AppState {
sys: shared,
nets: shared_nets,
net_totals,
});
let mut disks = Disks::new();
disks.refresh(true);
let addr = SocketAddr::from(([0, 0, 0, 0], 3000));
// Shared state across requests
let state = AppState {
sys: Arc::new(Mutex::new(sys)),
nets: Arc::new(Mutex::new(nets)),
net_totals: Arc::new(Mutex::new(HashMap::<String, (u64, u64)>::new())) as SharedTotals,
components: Arc::new(Mutex::new(components)),
disks: Arc::new(Mutex::new(disks)),
last_json: Arc::new(RwLock::new(String::new())),
// new: adaptive sampling controls
client_count: Arc::new(AtomicUsize::new(0)),
wake_sampler: Arc::new(Notify::new()),
auth_token: std::env::var("SOCKTOP_TOKEN").ok().filter(|s| !s.is_empty()),
};
// Start background sampler (adjust cadence as needed)
let _sampler = spawn_sampler(state.clone(), Duration::from_millis(500));
// Web app
let port = resolve_port();
let app = Router::new().route("/ws", get(ws_handler)).with_state(state);
let addr = SocketAddr::from(([0, 0, 0, 0], port));
//output to console
println!("Remote agent running at http://{}", addr);
println!("WebSocket endpoint: ws://{}/ws", addr);
//trace logging
tracing::info!("Remote agent running at http://{} (ws at /ws)", addr);
tracing::info!("WebSocket endpoint: ws://{}/ws", addr);
let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
axum::serve(listener, app).await.unwrap();
}
async fn ws_handler(ws: WebSocketUpgrade, State(state): State<AppState>) -> impl IntoResponse {
ws.on_upgrade(move |socket| handle_socket(socket, state))
}
// Resolve the listening port from CLI args/env with a 3000 default.
// Supports: --port <PORT>, -p <PORT>, a bare numeric positional arg, or SOCKTOP_PORT.
fn resolve_port() -> u16 {
const DEFAULT: u16 = 3000;
async fn handle_socket(mut socket: WebSocket, state: AppState) {
while let Some(Ok(msg)) = socket.next().await {
if let Message::Text(text) = msg {
if text == "get_metrics" {
let metrics = collect_metrics(&state).await;
let json = serde_json::to_string(&metrics).unwrap();
let _ = socket.send(Message::Text(json)).await;
// Env takes precedence over positional, but is overridden by explicit flags if present.
if let Ok(s) = std::env::var("SOCKTOP_PORT") {
if let Ok(p) = s.parse::<u16>() {
if p != 0 {
return p;
}
}
eprintln!("Warning: invalid SOCKTOP_PORT='{}'; using default {}", s, DEFAULT);
}
let mut args = std::env::args().skip(1);
while let Some(arg) = args.next() {
match arg.as_str() {
"--port" | "-p" => {
if let Some(v) = args.next() {
match v.parse::<u16>() {
Ok(p) if p != 0 => return p,
_ => {
eprintln!("Invalid port '{}'; using default {}", v, DEFAULT);
return DEFAULT;
}
}
} else {
eprintln!("Missing value for {} ; using default {}", arg, DEFAULT);
return DEFAULT;
}
}
"--help" | "-h" => {
println!("Usage: socktop_agent [--port <PORT>] [PORT]\n SOCKTOP_PORT=<PORT> socktop_agent");
std::process::exit(0);
}
s => {
if let Ok(p) = s.parse::<u16>() {
if p != 0 {
return p;
}
}
}
}
}
DEFAULT
}
// ---------- Metrics collection ----------
async fn collect_metrics(state: &AppState) -> Metrics {
// System (CPU/mem/proc)
let mut sys = state.sys.lock().await;
sys.refresh_all();
let hostname = System::host_name().unwrap_or_else(|| "unknown".into());
// Temps via Components (separate handle in 0.36)
let mut components = Components::new();
components.refresh(true);
let cpu_temp_c = best_cpu_temp(&components);
// Disks (separate handle in 0.36)
let mut disks_struct = Disks::new();
disks_struct.refresh(true);
// Filter anything with available == 0 (e.g., overlay)
let disks: Vec<DiskInfo> = disks_struct
.list()
.iter()
.filter(|d| d.available_space() > 0)
.map(|d| DiskInfo {
name: d.name().to_string_lossy().to_string(),
total: d.total_space(),
available: d.available_space(),
})
.collect();
// Networks: use a persistent Networks + rolling totals
let mut nets = state.nets.lock().await;
nets.refresh(true);
let mut totals = state.net_totals.lock().await;
let mut networks: Vec<NetworkInfo> = Vec::new();
for (name, data) in nets.iter() {
// sysinfo 0.36: data.received()/transmitted() are deltas since *last* refresh
let delta_rx = data.received();
let delta_tx = data.transmitted();
let entry = totals.entry(name.clone()).or_insert((0, 0));
entry.0 = entry.0.saturating_add(delta_rx);
entry.1 = entry.1.saturating_add(delta_tx);
networks.push(NetworkInfo {
name: name.clone(),
received: entry.0,
transmitted: entry.1,
});
}
// get number of cpu cores
let n_cpus = sys.cpus().len().max(1) as f32;
// Top processes: include PID and memory, top 20 by CPU
let mut top_processes: Vec<ProcessInfo> = sys
.processes()
.values()
.map(|p| ProcessInfo {
pid: p.pid().as_u32(),
name: p.name().to_string_lossy().to_string(),
cpu_usage: (p.cpu_usage() / n_cpus).min(100.0),
mem_bytes: p.memory(), // sysinfo 0.36: bytes
})
.collect();
top_processes.sort_by(|a, b| b.cpu_usage.partial_cmp(&a.cpu_usage).unwrap());
top_processes.truncate(20);
Metrics {
cpu_total: sys.global_cpu_usage(),
cpu_per_core: sys.cpus().iter().map(|c| c.cpu_usage()).collect(),
mem_total: sys.total_memory(),
mem_used: sys.used_memory(),
swap_total: sys.total_swap(),
swap_used: sys.used_swap(),
process_count: sys.processes().len(),
hostname,
cpu_temp_c,
disks,
networks,
top_processes,
}
}
fn best_cpu_temp(components: &Components) -> Option<f32> {
components
.iter()
.filter(|c| {
let label = c.label().to_lowercase();
label.contains("cpu") || label.contains("package") || label.contains("tctl") || label.contains("tdie")
})
.filter_map(|c| c.temperature())
.max_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal))
}
socktop_agent/src/metrics.rs
+128
View File
@@ -0,0 +1,128 @@
//! Metrics collection using sysinfo. Keeps sysinfo handles in AppState to
//! avoid repeated allocations and allow efficient refreshes.
use crate::state::AppState;
use crate::types::{DiskInfo, Metrics, NetworkInfo, ProcessInfo};
use sysinfo::{Components, System};
pub async fn collect_metrics(state: &AppState) -> Metrics {
// System (CPU/mem/proc)
let mut sys = state.sys.lock().await;
// Simple and safe — can be replaced by more granular refresh if desired:
// sys.refresh_cpu(); sys.refresh_memory(); sys.refresh_processes_specifics(...);
//sys.refresh_all();
//refresh all was found to use 2X CPU rather than individual refreshes
sys.refresh_cpu_all();
sys.refresh_memory();
sys.refresh_processes(sysinfo::ProcessesToUpdate::All, true);
let hostname = System::host_name().unwrap_or_else(|| "unknown".into());
// Temps via a persistent Components handle
let mut components = state.components.lock().await;
components.refresh(true);
let cpu_temp_c = best_cpu_temp(&components);
// Disks via a persistent Disks handle
let mut disks_struct = state.disks.lock().await;
disks_struct.refresh(true);
// Filter anything with available == 0 (e.g., overlay/virtual)
let disks: Vec<DiskInfo> = disks_struct
.list()
.iter()
.filter(|d| d.available_space() > 0)
.map(|d| DiskInfo {
name: d.name().to_string_lossy().to_string(),
total: d.total_space(),
available: d.available_space(),
})
.collect();
// Networks: use a persistent Networks + rolling totals
let mut nets = state.nets.lock().await;
nets.refresh(true);
let mut totals = state.net_totals.lock().await;
let mut networks: Vec<NetworkInfo> = Vec::new();
for (name, data) in nets.iter() {
// sysinfo: received()/transmitted() are deltas since last refresh
let delta_rx = data.received();
let delta_tx = data.transmitted();
let entry = totals.entry(name.clone()).or_insert((0, 0));
entry.0 = entry.0.saturating_add(delta_rx);
entry.1 = entry.1.saturating_add(delta_tx);
networks.push(NetworkInfo {
name: name.clone(),
received: entry.0,
transmitted: entry.1,
});
}
// Normalize process CPU to 0..100 across all cores
let n_cpus = sys.cpus().len().max(1) as f32;
// Build process list
let mut procs: Vec<ProcessInfo> = sys
.processes()
.values()
.map(|p| ProcessInfo {
pid: p.pid().as_u32(),
name: p.name().to_string_lossy().to_string(),
cpu_usage: (p.cpu_usage() / n_cpus).min(100.0),
mem_bytes: p.memory(),
})
.collect();
// Partial select: get the top 20 by CPU without fully sorting the vector
const TOP_N: usize = 20;
if procs.len() > TOP_N {
// nth index is TOP_N-1 (0-based)
let nth = TOP_N - 1;
procs.select_nth_unstable_by(nth, |a, b| {
b.cpu_usage
.partial_cmp(&a.cpu_usage)
.unwrap_or(std::cmp::Ordering::Equal)
});
procs.truncate(TOP_N);
// Order those 20 nicely for display
procs.sort_by(|a, b| {
b.cpu_usage
.partial_cmp(&a.cpu_usage)
.unwrap_or(std::cmp::Ordering::Equal)
});
} else {
procs.sort_by(|a, b| {
b.cpu_usage
.partial_cmp(&a.cpu_usage)
.unwrap_or(std::cmp::Ordering::Equal)
});
}
Metrics {
cpu_total: sys.global_cpu_usage(),
cpu_per_core: sys.cpus().iter().map(|c| c.cpu_usage()).collect(),
mem_total: sys.total_memory(),
mem_used: sys.used_memory(),
swap_total: sys.total_swap(),
swap_used: sys.used_swap(),
process_count: sys.processes().len(),
hostname,
cpu_temp_c,
disks,
networks,
top_processes: procs,
}
}
// Pick the hottest CPU-like sensor (labels vary by platform)
pub fn best_cpu_temp(components: &Components) -> Option<f32> {
components
.iter()
.filter(|c| {
let label = c.label().to_lowercase();
label.contains("cpu") || label.contains("package") || label.contains("tctl") || label.contains("tdie")
})
.filter_map(|c| c.temperature())
.max_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal))
}
socktop_agent/src/sampler.rs
+36
View File
@@ -0,0 +1,36 @@
//! Background sampler: periodically collects metrics and updates a JSON cache,
//! so WS replies are just a read of the cached string.
use crate::metrics::collect_metrics;
use crate::state::AppState;
//use serde_json::to_string;
use tokio::task::JoinHandle;
use tokio::time::{Duration, interval, MissedTickBehavior};
pub fn spawn_sampler(state: AppState, period: Duration) -> JoinHandle<()> {
tokio::spawn(async move {
let idle_period = Duration::from_secs(10);
loop {
let active = state.client_count.load(std::sync::atomic::Ordering::Relaxed) > 0;
let mut ticker = interval(if active { period } else { idle_period });
ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
ticker.tick().await;
if !active {
tokio::select! {
_ = ticker.tick() => {},
_ = state.wake_sampler.notified() => continue,
}
}
if let Ok(json) = async {
let m = collect_metrics(&state).await;
serde_json::to_string(&m)
}
.await
{
*state.last_json.write().await = json;
}
}
})
}
socktop_agent/src/state.rs
+30
View File
@@ -0,0 +1,30 @@
//! Shared agent state: sysinfo handles and hot JSON cache.
use std::{collections::HashMap, sync::Arc};
use std::sync::atomic::AtomicUsize;
use sysinfo::{Components, Disks, Networks, System};
use tokio::sync::{Mutex, RwLock, Notify};
pub type SharedSystem = Arc<Mutex<System>>;
pub type SharedNetworks = Arc<Mutex<Networks>>;
pub type SharedTotals = Arc<Mutex<HashMap<String, (u64, u64)>>>;
pub type SharedComponents = Arc<Mutex<Components>>;
pub type SharedDisks = Arc<Mutex<Disks>>;
#[derive(Clone)]
pub struct AppState {
// Persistent sysinfo handles
pub sys: SharedSystem,
pub nets: SharedNetworks,
pub net_totals: SharedTotals, // iface -> (rx_total, tx_total)
pub components: SharedComponents,
pub disks: SharedDisks,
// Last serialized JSON snapshot for fast WS responses
pub last_json: Arc<RwLock<String>>,
// Adaptive sampling controls
pub client_count: Arc<AtomicUsize>,
pub wake_sampler: Arc<Notify>,
pub auth_token: Option<String>,
}
socktop_agent/src/types.rs
+43
View File
@@ -0,0 +1,43 @@
//! Data types sent to the client over WebSocket.
//! Keep this module minimal and stable — it defines the wire format.
use serde::Serialize;
#[derive(Debug, Serialize, Clone)]
pub struct ProcessInfo {
pub pid: u32,
pub name: String,
pub cpu_usage: f32,
pub mem_bytes: u64,
}
#[derive(Debug, Serialize, Clone)]
pub struct DiskInfo {
pub name: String,
pub total: u64,
pub available: u64,
}
#[derive(Debug, Serialize, Clone)]
pub struct NetworkInfo {
pub name: String,
// cumulative totals since the agent started (client should diff to get rates)
pub received: u64,
pub transmitted: u64,
}
#[derive(Debug, Serialize, Clone)]
pub struct Metrics {
pub cpu_total: f32,
pub cpu_per_core: Vec<f32>,
pub mem_total: u64,
pub mem_used: u64,
pub swap_total: u64,
pub swap_used: u64,
pub process_count: usize,
pub hostname: String,
pub cpu_temp_c: Option<f32>,
pub disks: Vec<DiskInfo>,
pub networks: Vec<NetworkInfo>,
pub top_processes: Vec<ProcessInfo>,
}
socktop_agent/src/ws.rs
+66
View File
@@ -0,0 +1,66 @@
//! WebSocket upgrade and per-connection handler. Serves cached JSON quickly.
use axum::{
extract::{
ws::{Message, WebSocket, WebSocketUpgrade},
Query, State,
},
http::StatusCode,
response::{IntoResponse, Response},
};
use futures_util::stream::StreamExt;
use crate::metrics::collect_metrics;
use crate::state::AppState;
use std::collections::HashMap;
use std::sync::atomic::Ordering;
pub async fn ws_handler(
ws: WebSocketUpgrade,
State(state): State<AppState>,
Query(q): Query<HashMap<String, String>>,
) -> Response {
if let Some(expected) = state.auth_token.as_ref() {
match q.get("token") {
Some(t) if t == expected => {}
_ => return StatusCode::UNAUTHORIZED.into_response(),
}
}
ws.on_upgrade(move |socket| handle_socket(socket, state))
}
async fn handle_socket(mut socket: WebSocket, state: AppState) {
// Bump client count on connect and wake the sampler.
state.client_count.fetch_add(1, Ordering::Relaxed);
state.wake_sampler.notify_waiters();
// Ensure we decrement on disconnect (drop).
struct ClientGuard(AppState);
impl Drop for ClientGuard {
fn drop(&mut self) {
self.0.client_count.fetch_sub(1, Ordering::Relaxed);
self.0.wake_sampler.notify_waiters();
}
}
let _guard = ClientGuard(state.clone());
while let Some(Ok(msg)) = socket.next().await {
match msg {
Message::Text(text) if text == "get_metrics" => {
// Serve the cached JSON quickly; if empty (cold start), collect once.
let cached = state.last_json.read().await.clone();
if !cached.is_empty() {
let _ = socket.send(Message::Text(cached)).await;
} else {
let metrics = collect_metrics(&state).await;
if let Ok(js) = serde_json::to_string(&metrics) {
let _ = socket.send(Message::Text(js)).await;
}
}
}
Message::Close(_) => break,
_ => {}
}
}
}