Compare commits

..

106 Commits

Author SHA1 Message Date
jason 4e378b882a add license file
CI / build (ubuntu-latest) (push) Has been cancelled
CI / build (windows-latest) (push) Has been cancelled
2025-09-03 23:10:48 -07:00
jason 0c5a1d7553 update readme 2025-08-30 01:59:25 -07:00
jason 0bd709d2a7 slipsteam note for rpi users on kernel version 2025-08-29 11:28:59 -07:00
jason 31f5f9ce76 re: Accessibility: Cross-compile guide 2025-08-29 11:23:41 -07:00
jason df2308e6e9 code optimizations to reduce cpu usage of agent on all platforms and additional unit test. 2025-08-28 16:03:05 -07:00
jason 7592709a43 clamp then divide by cores for more accurate statistics 2025-08-28 13:11:48 -07:00
jason 61fe1cc38e socktop_agent: bump version to 1.40.65 2025-08-28 12:27:48 -07:00
jason eed346abb6 scripts: add publish_socktop_agent.sh job 2025-08-28 12:07:03 -07:00
jason ab3bb33711 socktop_agent: bump version to 1.40.64 2025-08-28 12:03:45 -07:00
jason 7caf2f4bfb remove ununused var 2025-08-27 17:15:35 -07:00
jason b249c7ba99 Update metrics.rs 2025-08-27 16:55:12 -07:00
jason f0858525e8 fix for macos defect where processes less than .01% were being filtered 2025-08-27 16:55:09 -07:00
jason 2fe005ed90 ProcessesToUpdate::All enum 2025-08-27 16:20:55 -07:00
jason ca6a5cbdfa use ProcessesToUpdate::All ENUM 2025-08-27 16:20:41 -07:00
jason 56301d61fd fixes for non linux compilation issues. 2025-08-27 16:11:17 -07:00
jason 55e5c708fe MACOS / NON LINUX metrics optimizations. 2025-08-27 16:00:29 -07:00
jason 2d17cf1598 additional optimizations for macos 2025-08-27 15:05:38 -07:00
jason 353c08c35e increment version and macos performance 2025-08-26 12:14:32 -07:00
jason f13ea45360 increment version 2025-08-26 10:57:01 -07:00
jason 8ce00a5dad non linux optimizations for macbook 2025-08-26 10:14:14 -07:00
jason f37b8d9ff4 chore(agent): fix clippy unused mut on non-linux process list 2025-08-26 00:22:29 -07:00
jason 322981ada7 cargo fmt and version bump 2025-08-26 00:20:52 -07:00
jason 3394beab67 chore: make pre-commit resilient when cargo absent 2025-08-26 00:18:10 -07:00
jason c9ebea92f5 perf(agent,non-linux): enable CPU core normalization by default; tighter scaling threshold 2025-08-25 23:50:26 -07:00
jason e2dc5e8ac9 perf(agent,non-linux): add per-process CPU scaling heuristic to reduce overreporting 2025-08-25 23:47:31 -07:00
jason beddba0072 perf(agent,non-linux): reduce process collection overhead; configurable CPU sample delay 2025-08-25 23:44:28 -07:00
jason cacc4cba9f cargo fmt 2025-08-25 23:24:50 -07:00
jason 66270c16b7 perf(agent): windows/mac cpu collection tweak; add optional normalization; silence linux dead_code 2025-08-25 22:59:07 -07:00
jason 00d5777d05 Merge branch 'master' of https://github.com/jasonwitty/socktop 2025-08-25 22:35:02 -07:00
jason f62b5274d2 optimize non linux metrics collection 2025-08-25 22:35:01 -07:00
jason bbbe35111a Update README.md 2025-08-25 01:33:40 -07:00
jason a4356b5ece update readme with animated demo 2025-08-24 20:35:56 -07:00
jason b6e656738b chore(release): bump to 1.40.0
CI / build (ubuntu-latest) (push) Has been cancelled
CI / build (windows-latest) (push) Has been cancelled
2025-08-24 18:56:40 -07:00
jason f83cb07d57 Release candidate 1.4
increment version, support version flag on socktop
2025-08-24 18:03:45 -07:00
jason 7697c7dc2b docs: add per-crate README.md and link via Cargo.toml readme field 2025-08-24 17:56:22 -07:00
jason 1043fffc8d Merge pull request #5 from jasonwitty/feature/housekeeping
Feature/housekeeping
2025-08-24 17:46:26 -07:00
jason ce59dd9dfe chore(agent): fix clippy needless_return for non-linux process collection
CI / build (ubuntu-latest) (push) Has been cancelled
CI / build (windows-latest) (push) Has been cancelled
2025-08-24 12:52:56 -07:00
jason 8d48fa4c3b increment version 2025-08-24 12:47:49 -07:00
jason 51e702368e cargo fmt 2025-08-24 12:40:35 -07:00
jason 85f9a44e46 perf(agent): add hostname + TTL caches (metrics/disks/processes) and reuse sys for processes 2025-08-24 12:38:32 -07:00
jason b2468a5936 refactor(agent): remove unused background sampler infrastructure (request-driven only) 2025-08-24 12:29:23 -07:00
jason 8de5943f34 test(agent): move inline port parsing test to tests/port_parse.rs 2025-08-24 12:15:32 -07:00
jason e624751f56 chore: align sysinfo to 0.37 across workspace 2025-08-24 12:11:38 -07:00
jason 8bd1af7a27 chore: remove unused deps (thiserror, chrono, futures, nvml-wrapper, tungstenite, bytes, prost-types) 2025-08-24 12:08:52 -07:00
jason 5c32d15156 first run messaging.
display welcome message on first run when a profile file is not created and remote server not specified.
2025-08-24 12:01:17 -07:00
jason 471d547b5d cargo fmt and increment version 2025-08-23 02:18:09 -07:00
jason d3aff590bc client: fully disable hostname verification by custom ServerCertVerifier unless --verify-hostname used 2025-08-23 02:15:45 -07:00
jason 47910725a8 increment version 2025-08-22 22:48:41 -07:00
jason a8e3f4ef26 docs: describe --verify-hostname flag and default relaxed SAN behavior 2025-08-22 22:39:18 -07:00
jason fab1e5a104 client: default skip hostname verification; add --verify-hostname to enable 2025-08-22 22:39:06 -07:00
jason d0455611d5 cargo fmt 2025-08-22 14:09:02 -07:00
jason 4c45b85c98 fix san and increment version 2025-08-22 14:07:41 -07:00
jason d9fdc31e8f docs: document SOCKTOP_AGENT_EXTRA_SANS for additional certificate SANs 2025-08-22 13:48:07 -07:00
jason dc9aa4c026 agent: support extra SANs via SOCKTOP_AGENT_EXTRA_SANS env var 2025-08-22 13:47:51 -07:00
jason c2e91bd20c docs: document TLS cert expiry and manual regeneration procedure 2025-08-22 12:57:48 -07:00
jason 25229d6b03 cargo fmt 2025-08-22 12:49:12 -07:00
jason 290e2a8fb2 fix for expired certificate 2025-08-22 12:46:01 -07:00
jason 30d263c71e agent: dynamic self-signed cert validity (~397d from now) to avoid immediate expiry 2025-08-22 12:43:48 -07:00
jason 9b177f3206 cargo fmt 2025-08-22 11:55:41 -07:00
jason 8a6ae3fcd7 increment version 2025-08-22 11:53:27 -07:00
jason 5b8ec7efc1 agent: add --version / -V flag 2025-08-22 11:52:51 -07:00
jason 155c420a1a cargo fmt all check ins 2025-08-22 11:10:11 -07:00
jason d3fa55e572 chore: ignore .vscode and remove from repo 2025-08-22 11:09:33 -07:00
jason faf2861b29 cargo fmt 2025-08-22 10:53:47 -07:00
jason 59432ab1d3 agent: fix rcgen usage for self-signed cert generation 2025-08-22 10:48:01 -07:00
jason d1c8a64418 agent: replace openssl self-signed cert generation with rcgen (pure Rust) 2025-08-22 10:46:29 -07:00
jason 8def4b2d06 Publish: include proto in each crate and fix build.rs paths 2025-08-22 10:44:56 -07:00
jason a42ca71a9f update version prior to cargo publish 2025-08-22 10:41:17 -07:00
jason 9f675fa804 Merge pull request #4 from jasonwitty/feature/connection-profiles
Feature/connection profiles
2025-08-22 09:31:28 -07:00
jason 3ac03c07ba enable GPU polling only when GPU is present 2025-08-22 09:27:05 -07:00
jason e53d0ab98d Add TLS / Token, polling interval indicators.
CI / build (ubuntu-latest) (push) Has been cancelled
CI / build (windows-latest) (push) Has been cancelled
2025-08-21 17:38:26 -07:00
jason 2ca51adc61 tui: refine header icons (crossed TLS when disabled, spacing fix) 2025-08-21 17:28:21 -07:00
jason 67ecf36883 feat(tui): header shows TLS/token status and polling intervals 2025-08-21 17:24:41 -07:00
jason 9a35306340 cargo fmt 2025-08-21 16:19:49 -07:00
jason a4bb6f170a feat(client): configurable metrics/process intervals with profile persistence; docs updated 2025-08-21 16:18:41 -07:00
jason 384953d5d5 Merge pull request #3 from jasonwitty/feature/connection-profiles
Feature/connection profiles
2025-08-21 16:17:49 -07:00
jason f9114426cc add unit tests for profile creation and update readme 2025-08-21 14:42:15 -07:00
jason 8ee2a03a2c chore(client): clean up demo mode integration and add stop log line 2025-08-21 13:55:02 -07:00
jason 0275b1871d cargo fmt 2025-08-21 13:49:36 -07:00
jason 9491dc50a8 feat(client): demo mode (--demo or select demo) auto-spawns local agent on 3231 2025-08-21 13:47:28 -07:00
jason e7eb3e6557 cargo fmt 2025-08-21 13:18:36 -07:00
jason a596acfb72 chore(client): refactor profile overwrite logic to satisfy clippy 2025-08-21 13:17:53 -07:00
jason b727e54589 feat(client): prompt for URL/CA when specifying a new profile name 2025-08-21 12:56:11 -07:00
jason 2af08c455a fix(client): correct profile overwrite prompt logic (only save on confirm or --save) 2025-08-21 12:48:53 -07:00
jason d049846564 docs: add connection profiles section to README 2025-08-21 12:41:46 -07:00
jason 97308f9d15 feat(client): connection profiles (--profile/-P, --save) with JSON persistence 2025-08-21 12:39:21 -07:00
jason 4cef273e57 Merge pull request #2 from jasonwitty/feature/protobuf-processes
Feature/protobuf processes
2025-08-21 11:50:27 -07:00
jason 660474a6ce ci cleanup
CI / build (ubuntu-latest) (push) Has been cancelled
CI / build (windows-latest) (push) Has been cancelled
2025-08-20 20:36:49 -07:00
jason 93dd14967d try/fix windows again ! 2025-08-20 16:32:51 -07:00
jason 923a3872fe add logging to help debug windows problems 2025-08-20 15:47:28 -07:00
jason 5f10e34341 windows try/fix 2025-08-20 15:20:07 -07:00
jason b80d322650 cargo fmt 2025-08-20 11:29:22 -07:00
jason fff386f9d5 fixing windows build problems. i hate windows !
Agent:
Added GET /healthz that returns 200 immediately.
File: main.rs (router now includes /healthz).
CI workflow:
Start agent from target/release on both OSes.
Set SOCKTOP_ENABLE_SSL=0 explicitly.
Ubuntu: wait on curl http://127.0.0.1:3000/healthz (60s), log tail and ss/netstat on failure.
Windows: wait on Invoke-WebRequest to /healthz (60s), capture stdout/stderr, print netstat on failure.
File: .github/workflows/ci.yml.
2025-08-20 11:26:09 -07:00
jason 93f4e1feea fix windows build after ssl feature and optimize build 2025-08-20 10:24:24 -07:00
jason 97255b42fb fix windows build 2025-08-20 00:14:21 -07:00
jason 554a2c349f protobuff Process list
BREAKING: Process list over WS is now Protocol Buffers; client required.
Agent: returns all processes (no server-side top-k); large payloads gzip-compressed.
Client: decodes protobuf (gz/raw), moves sorting/pagination to TUI.
Build: add prost/prost-build with vendored protoc; enable thin LTO, panic=abort, strip symbols.
Cleanup: cfg-gate Linux-only code; fix Clippy across platforms; tests updated (ws probe TLS CA).
2025-08-19 23:24:36 -07:00
jason 10501168c5 clippy fixes. 2025-08-19 15:52:30 -07:00
jason d346c61c28 Merge pull request #1 from jasonwitty/feature/wss-selfsigned
SSL Support
2025-08-19 15:33:50 -07:00
jason 7652095109 cargo fmt
CI / build (ubuntu-latest) (push) Has been cancelled
CI / build (windows-latest) (push) Has been cancelled
2025-08-19 15:33:11 -07:00
jason 6b58ac67f6 Merge branch 'master' into feature/wss-selfsigned 2025-08-19 15:31:10 -07:00
jason 3ad1d52fe2 fix windows build.
CI / build (ubuntu-latest) (push) Has been cancelled
CI / build (windows-latest) (push) Has been cancelled
Gate Linux-specific imports and fields to avoid Windows dead-code/unused warnings.
Ensure Linux-only proc CPU tracker is not referenced on non-Linux builds.
2025-08-16 17:42:18 -07:00
jason 2e8cc24e81 remove dead code and increment version 2025-08-16 17:31:32 -07:00
jason 36e73fd9ed cargo fmt 2025-08-16 01:25:03 -07:00
jason 3d14e4a370 SSL Support
Add WSS/TLS (self‑signed) with client cert pinning; auto ws→wss on --tls-ca/-t; add -p/-t flags; harden TLS test; fix clippy; update README.

feat: WSS/TLS support (self‑signed + pinning), auto ws→wss when CA provided, new -p/-t flags; tests + clippy cleanup; docs updated.

Add TLS: self‑signed certs on agent, client pin via --tls-ca/-t (auto‑upgrade to wss), CLI/tests/README updates, clippy fixes.

12 files changed
Cargo.toml
README.md
Cargo.tomlsocktop_agent
main.rssocktop_agent/src
tls.rssocktop_agent/src
cli_args.rssocktop_agent/tests
Add Context...
README.md
2025-08-16 01:23:20 -07:00
jason c6b8c9c905 patch for macbook compatibility issues. 2025-08-15 19:21:34 -07:00
jason f980b6ace9 Update README.md 2025-08-13 20:17:36 -07:00
42 changed files with 3859 additions and 499 deletions
+39
View File
@@ -0,0 +1,39 @@
#!/usr/bin/env bash
# This repository uses a custom hooks directory (.githooks). To enable this pre-commit hook run:
# git config core.hooksPath .githooks
# Ensure this file is executable: chmod +x .githooks/pre-commit
set -euo pipefail
echo "[pre-commit] Running cargo fmt --all" >&2
if ! command -v cargo >/dev/null 2>&1; then
# Try loading rustup environment (common install path)
if [ -f "$HOME/.cargo/env" ]; then
# shellcheck source=/dev/null
. "$HOME/.cargo/env"
fi
fi
if ! command -v cargo >/dev/null 2>&1; then
echo "[pre-commit] cargo not found in PATH; skipping fmt (install Rust or adjust PATH)." >&2
exit 0
fi
cargo fmt --all
# Stage any Rust files that were reformatted
changed=$(git diff --name-only --diff-filter=M | grep -E '\\.rs$' || true)
if [ -n "$changed" ]; then
echo "$changed" | xargs git add
echo "[pre-commit] Added formatted files" >&2
fi
# Fail if further diffs remain (shouldn't happen normally)
unfmt=$(git diff --name-only --diff-filter=M | grep -E '\\.rs$' || true)
if [ -n "$unfmt" ]; then
echo "[pre-commit] Some Rust files still differ after formatting:" >&2
echo "$unfmt" >&2
exit 1
fi
exit 0
+81 -13
View File
@@ -5,16 +5,16 @@ on:
jobs:
build:
runs-on: ${{ matrix.os }}
strategy:
matrix:
os: [ubuntu-latest, windows-latest]
runs-on: ${{ matrix.os }}
steps:
- uses: actions/checkout@v4
- uses: dtolnay/rust-toolchain@stable
with:
components: clippy, rustfmt
- name: Install system dependencies
- name: Install system dependencies (Linux)
if: matrix.os == 'ubuntu-latest'
run: sudo apt-get update && sudo apt-get install -y libdrm-dev libdrm-amdgpu1
- name: Cargo fmt
@@ -23,22 +23,89 @@ jobs:
run: cargo clippy --all-targets --all-features -- -D warnings
- name: Build (release)
run: cargo build --release --workspace
- name: "Linux: start agent and run WS probe"
if: matrix.os == 'ubuntu-latest'
shell: bash
run: |
set -euo pipefail
RUST_LOG=info SOCKTOP_ENABLE_SSL=0 SOCKTOP_AGENT_GPU=0 SOCKTOP_AGENT_TEMP=0 ./target/release/socktop_agent -p 3000 > agent.log 2>&1 &
AGENT_PID=$!
for i in {1..60}; do
if curl -fsS http://127.0.0.1:3000/healthz >/dev/null; then break; fi
sleep 1
done
if ! curl -fsS http://127.0.0.1:3000/healthz >/dev/null; then
echo "--- agent.log (tail) ---"
tail -n 200 agent.log || true
(command -v ss >/dev/null && ss -ltnp || netstat -ltnp) || true
kill $AGENT_PID || true
exit 1
fi
SOCKTOP_WS=ws://127.0.0.1:3000/ws cargo test -p socktop --test ws_probe -- --nocapture
kill $AGENT_PID || true
- name: "Windows: start agent and run WS probe"
if: matrix.os == 'windows-latest'
shell: pwsh
run: |
$env:SOCKTOP_ENABLE_SSL = "0"
$env:SOCKTOP_AGENT_GPU = "0"
$env:SOCKTOP_AGENT_TEMP = "0"
$out = Join-Path $PWD "agent.out.txt"
$err = Join-Path $PWD "agent.err.txt"
$p = Start-Process -FilePath "${PWD}\target\release\socktop_agent.exe" -ArgumentList "-p 3000" -RedirectStandardOutput $out -RedirectStandardError $err -PassThru -NoNewWindow
$ready = $false
for ($i = 0; $i -lt 60; $i++) {
$pinfo = New-Object System.Diagnostics.ProcessStartInfo
$pinfo.FileName = "curl.exe"
$pinfo.Arguments = "-fsS http://127.0.0.1:3000/healthz"
$pinfo.RedirectStandardOutput = $true
$pinfo.RedirectStandardError = $true
$pinfo.UseShellExecute = $false
$proc = [System.Diagnostics.Process]::Start($pinfo)
$proc.WaitForExit()
if ($proc.ExitCode -eq 0) { $ready = $true; break }
Start-Sleep -Seconds 1
}
if (-not $ready) {
Write-Warning "TCP connect to (127.0.0.1 : 3000) failed"
if (Test-Path $out) { Write-Host "--- agent.out (full) ---"; Get-Content $out }
if (Test-Path $err) { Write-Host "--- agent.err (full) ---"; Get-Content $err }
Write-Host "--- netstat ---"
netstat -ano | Select-String ":3000" | ForEach-Object { $_.Line }
if ($p -and !$p.HasExited) { Stop-Process -Id $p.Id -Force -ErrorAction SilentlyContinue }
throw "agent did not become ready"
}
$env:SOCKTOP_WS = "ws://127.0.0.1:3000/ws"
try {
cargo test -p socktop --test ws_probe -- --nocapture
} finally {
if ($p -and !$p.HasExited) { Stop-Process -Id $p.Id -Force -ErrorAction SilentlyContinue }
}
- name: Smoke test (client --help)
run: cargo run -p socktop -- --help
- name: Package artifacts
- name: Package artifacts (Linux)
if: matrix.os == 'ubuntu-latest'
shell: bash
run: |
set -e
mkdir dist
if [[ "${{ matrix.os }}" == "windows-latest" ]]; then
cp target/release/socktop.exe dist/
cp target/release/socktop_agent.exe dist/
7z a socktop-${{ matrix.os }}.zip dist/*
else
cp target/release/socktop dist/
cp target/release/socktop_agent dist/
tar czf socktop-${{ matrix.os }}.tar.gz -C dist .
fi
mkdir -p dist
cp target/release/socktop dist/
cp target/release/socktop_agent dist/
tar czf socktop-${{ matrix.os }}.tar.gz -C dist .
- name: Package artifacts (Windows)
if: matrix.os == 'windows-latest'
shell: pwsh
run: |
New-Item -ItemType Directory -Force -Path dist | Out-Null
Copy-Item target\release\socktop.exe dist\
Copy-Item target\release\socktop_agent.exe dist\
Compress-Archive -Path dist\* -DestinationPath socktop-${{ matrix.os }}.zip -Force
- name: Upload build artifacts (ephemeral)
uses: actions/upload-artifact@v4
with:
@@ -46,6 +113,7 @@ jobs:
path: |
*.tar.gz
*.zip
- name: Upload to rolling GitHub Release (main only)
if: github.ref == 'refs/heads/main' && github.event_name == 'push'
uses: softprops/action-gh-release@v2
+1
View File
@@ -1 +1,2 @@
/target
.vscode/
-83
View File
@@ -1,83 +0,0 @@
{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"type": "lldb",
"request": "launch",
"name": "Debug executable 'socktop'",
"cargo": {
"args": [
"build",
"--bin=socktop",
"--package=socktop"
],
"filter": {
"name": "socktop",
"kind": "bin"
}
},
"args": ["ws://127.0.0.1:3000/ws"],
"cwd": "${workspaceFolder}"
},
{
"type": "lldb",
"request": "launch",
"name": "Debug unit tests in executable 'socktop'",
"cargo": {
"args": [
"test",
"--no-run",
"--bin=socktop",
"--package=socktop"
],
"filter": {
"name": "socktop",
"kind": "bin"
}
},
"args": [],
"cwd": "${workspaceFolder}"
},
{
"type": "lldb",
"request": "launch",
"name": "Debug executable 'socktop_agent'",
"cargo": {
"args": [
"build",
"--bin=socktop_agent",
"--package=socktop_agent"
],
"filter": {
"name": "socktop_agent",
"kind": "bin"
}
},
"args": [],
"cwd": "${workspaceFolder}"
},
{
"type": "lldb",
"request": "launch",
"name": "Debug unit tests in executable 'socktop_agent'",
"cargo": {
"args": [
"test",
"--no-run",
"--bin=socktop_agent",
"--package=socktop_agent"
],
"filter": {
"name": "socktop_agent",
"kind": "bin"
}
},
"args": [],
"cwd": "${workspaceFolder}"
}
]
}
Generated
+914 -67
View File
File diff suppressed because it is too large Load Diff
+15 -8
View File
@@ -8,29 +8,36 @@ members = [
[workspace.dependencies]
# async + streams
tokio = { version = "1", features = ["full"] }
futures = "0.3"
futures-util = "0.3"
anyhow = "1.0"
# websocket
tokio-tungstenite = "0.24"
tungstenite = "0.24"
tokio-tungstenite = { version = "0.24", features = ["__rustls-tls", "connect"] }
url = "2.5"
# JSON + error handling
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
thiserror = "1.0"
# system stats
sysinfo = "0.32"
# system stats (align across crates)
sysinfo = "0.37"
# CLI UI
ratatui = "0.28"
crossterm = "0.27"
# date/time
chrono = { version = "0.4", features = ["serde"] }
# web server (remote-agent)
axum = { version = "0.7", features = ["ws"] }
# protobuf
prost = "0.13"
dirs-next = "2"
[profile.release]
# Favor smaller, simpler binaries with good runtime perf
lto = "thin"
codegen-units = 1
panic = "abort"
opt-level = 3
strip = "symbols"
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Witty One Off
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+216 -9
View File
@@ -5,13 +5,14 @@ socktop is a remote system monitor with a rich TUI, inspired by top/btop, talkin
- Linux agent: near-zero CPU when idle (request-driven, no always-on sampler)
- TUI: smooth graphs, sortable process table, scrollbars, readable colors
![socktop screenshot](./docs/14900ks_arch_alacritty_gpu_active_v2.jpg)
<img src="./docs/socktop_demo.apng" width="100%">
---
## Features
- Remote monitoring via WebSocket (JSON over WS)
- Optional WSS (TLS): agent autogenerates a selfsigned cert on first run; client pins the cert via --tls-ca/-t
- TUI built with ratatui
- CPU
- Overall sparkline + per-core mini bars
@@ -50,7 +51,7 @@ exec bash # or: exec zsh / exec fish
Windows (for the brave): install from https://rustup.rs with the MSVC toolchain. Yes, youll need Visual Studio Build Tools. You chose Windows — enjoy the ride.
### Raspberry Pi (required)
### Raspberry Pi / Ubuntu / PopOS (required)
Install GPU support with apt command below
@@ -59,15 +60,17 @@ sudo apt-get update
sudo apt-get install libdrm-dev libdrm-amdgpu1
```
_Additional note for Raspberry Pi users. Please update your system to use the newest kernel available through app, kernel version 6.6+ will use considerably less overall CPU to run the agent. For example on a rpi4 the kernel < 6.6 the agent will consume .8 cpu but on the same hardware on > 6.6 the agent will consume only .2 cpu. (these numbers indicate continuous polling at web socket endpoints, when not in use the usage is 0)_
---
## Architecture
Two components:
1) Agent (remote): small Rust WS server using sysinfo + /proc. It collects on demand when the client asks (fast metrics ~500 ms, processes ~2 s, disks ~5 s). No background loop when nobody is connected.
1) Agent (remote): small Rust WS server using sysinfo + /proc. It collects metrics only when the client requests them over the WebSocket (request-driven). No background sampling loop.
2) Client (local): TUI that connects to ws://HOST:PORT/ws and renders updates.
2) Client (local): TUI that connects to ws://HOST:PORT/ws (or wss://HOST:PORT/ws when TLS is enabled) and renders updates.
---
@@ -93,7 +96,24 @@ cargo build --release
./target/release/socktop ws://REMOTE_HOST:3000/ws
```
Tip: Add ?token=... if you enable auth (see Security).
### Cross-compiling for Raspberry Pi
For Raspberry Pi and other ARM devices, you can cross-compile the agent from a more powerful machine:
- [Cross-compilation guide](./docs/cross-compiling.md) - Instructions for cross-compiling from Linux, macOS, or Windows hosts
### Quick demo (no agent setup)
Spin up a temporary local agent on port 3231 and connect automatically:
```bash
socktop --demo
```
Or just run `socktop` with no arguments and pick the builtin `demo` entry from the interactive profile list (if you have saved profiles, `demo` is appended). The demo agent:
- Runs locally (`ws://127.0.0.1:3231/ws`)
- Stops automatically (you'll see "Stopped demo agent on port 3231") when you quit the TUI or press Ctrl-C
---
@@ -114,7 +134,8 @@ Notes:
- After installing Rust via rustup, reload your shell (e.g., exec bash) so cargo is on PATH.
- Windows: you can also grab prebuilt EXEs from GitHub Actions artifacts if rustup scares you. It shouldnt. Be brave.
Option B: System-wide agent (Linux)
System-wide agent (Linux)
```bash
# If you installed with cargo, binaries are in ~/.cargo/bin
sudo install -o root -g root -m 0755 "$HOME/.cargo/bin/socktop_agent" /usr/local/bin/socktop_agent
@@ -125,6 +146,36 @@ sudo systemctl daemon-reload
sudo systemctl enable --now socktop-agent
```
```bash
# Enable SSL
# Stop service
sudo systemctl stop socktop-agent
# Edit service to append SSL option and port
sudo micro /etc/systemd/system/socktop-agent.service
--
ExecStart=/usr/local/bin/socktop_agent --enableSSL --port 8443
--
# Reload
sudo systemctl daemon-reload
# Restart
sudo systemctl start socktop-agent
# check logs for certificate location
sudo journalctl -u socktop-agent -f
--
Aug 22 22:25:26 rpi-master socktop_agent[2913998]: socktop_agent: generated self-signed TLS certificate at /var/lib/socktop/.config/socktop_agent/tls/cert.pem
--
```
---
## Usage
@@ -135,6 +186,8 @@ Agent (server):
socktop_agent --port 3000
# or env: SOCKTOP_PORT=3000 socktop_agent
# optional auth: SOCKTOP_TOKEN=changeme socktop_agent
# enable TLS (selfsigned cert, default port 8443; you can also use -p):
socktop_agent --enableSSL --port 8443
```
Client (TUI):
@@ -143,6 +196,13 @@ Client (TUI):
socktop ws://HOST:3000/ws
# with token:
socktop "ws://HOST:3000/ws?token=changeme"
# TLS with pinned server certificate (recommended over the internet):
socktop --tls-ca /path/to/cert.pem wss://HOST:8443/ws
# (By default hostname/SAN verification is skipped for ease on home networks. To enforce it add --verify-hostname)
socktop --verify-hostname --tls-ca /path/to/cert.pem wss://HOST:8443/ws
# shorthand:
socktop -t /path/to/cert.pem wss://HOST:8443/ws
# Note: providing --tls-ca/-t automatically upgrades ws:// to wss:// if you forget
```
Intervals (client-driven):
@@ -154,6 +214,96 @@ The agent stays idle unless queried. When queried, it collects just whats nee
---
## Connection Profiles (Named)
You can save frequently used connection settings (URL + optional TLS CA path) under a short name and reuse them later.
Config file location:
- Linux (XDG): `$XDG_CONFIG_HOME/socktop/profiles.json`
- Fallback (when XDG not set): `~/.config/socktop/profiles.json`
### Creating a profile
First time you specify a new `--profile/-P` name together with a URL (and optional `--tls-ca`), it is saved automatically:
```bash
socktop --profile prod ws://prod-host:3000/ws
# With TLS pinning:
socktop --profile prod-tls --tls-ca /path/to/cert.pem wss://prod-host:8443/ws
You can also set custom intervals (milliseconds):
```bash
socktop --profile prod --metrics-interval-ms 750 --processes-interval-ms 3000 ws://prod-host:3000/ws
```
```
If a profile already exists you will be prompted before overwriting:
```
$ socktop --profile prod ws://new-host:3000/ws
Overwrite existing profile 'prod'? [y/N]: y
```
To overwrite without an interactive prompt pass `--save`:
```bash
socktop --profile prod --save ws://new-host:3000/ws
```
### Using a saved profile
Just pass the profile name (no URL needed):
```bash
socktop --profile prod
socktop -P prod-tls # short flag
```
The stored URL (and TLS CA path, if any) plus any saved intervals will be used. TLS auto-upgrade still applies if a CA path is stored alongside a ws:// URL.
### Interactive selection (no args)
If you run `socktop` with no arguments and at least one profile exists, you will be shown a numbered list to pick from:
```
$ socktop
Select profile:
1. prod
2. prod-tls
Enter number (or blank to abort): 2
```
Choosing a number starts the TUI with that profile. A builtin `demo` option is always appended; selecting it launches a local agent on port 3231 (no TLS) and connects to `ws://127.0.0.1:3231/ws`. Pressing Enter on blank aborts without connecting.
### JSON format
An example `profiles.json` (prettyprinted):
```json
{
"profiles": {
"prod": { "url": "ws://prod-host:3000/ws" },
"prod-tls": {
"url": "wss://prod-host:8443/ws",
"tls_ca": "/home/user/certs/prod-cert.pem",
"metrics_interval_ms": 500,
"processes_interval_ms": 2000
}
},
"version": 0
}
```
Notes:
- The `tls_ca` path is stored as given; if you move or rotate the certificate update the profile by re-running with `--profile NAME --save`.
- Deleting a profile: edit the JSON file and remove the entry (TUI does not yet have an in-app delete command).
- Profiles are client-side convenience only; they do not affect the agent.
- Intervals: `metrics_interval_ms` controls the fast metrics poll (default 500 ms). `processes_interval_ms` controls process list polling (default 2000 ms). Values below 100 ms (metrics) or 200 ms (processes) are clamped.
---
## Updating
Update the agent (systemd):
@@ -188,6 +338,25 @@ Tip: If only the binary changed, restart is enough. If the unit file changed, ru
- Flag: --port 8080 or -p 8080
- Positional: socktop_agent 8080
- Env: SOCKTOP_PORT=8080
- TLS (selfsigned):
- Enable: --enableSSL
- Default TLS port: 8443 (override with --port/-p)
- Certificate/Key location (created on first TLS run):
- Linux (XDG): $XDG_CONFIG_HOME/socktop_agent/tls/{cert.pem,key.pem} (defaults to ~/.config)
- The agent prints these paths on creation.
- You can set XDG_CONFIG_HOME before first run to control where certs are written.
- Additional SANs: set `SOCKTOP_AGENT_EXTRA_SANS` (commaseparated) before first TLS start to include extra IPs/DNS names in the cert. Example:
```bash
SOCKTOP_AGENT_EXTRA_SANS="192.168.1.101,myhost.internal" socktop_agent --enableSSL
```
This prevents client errors like `NotValidForName` when connecting via an IP not present in the default cert SAN list.
- Expiry / rotation: the generated cert is valid for ~397 days from creation. If the agent fails to start with an "ExpiredCertificate" error (or your client reports expiry), simply delete the existing cert and key:
```bash
rm ~/.config/socktop_agent/tls/cert.pem ~/.config/socktop_agent/tls/key.pem
# (adjust path if XDG_CONFIG_HOME is set or different user)
systemctl restart socktop-agent # if running under systemd
```
On next TLS start the agent will generate a fresh pair. Only distribute the new cert.pem to clients (never the key).
- Auth token (optional): SOCKTOP_TOKEN=changeme
- Disable GPU metrics: SOCKTOP_AGENT_GPU=0
- Disable CPU temperature: SOCKTOP_AGENT_TEMP=0
@@ -250,6 +419,28 @@ Client:
socktop "ws://HOST:3000/ws?token=changeme"
```
### TLS / WSS
For encrypted connections, enable TLS on the agent and pin the server certificate on the client.
Server (generates selfsigned cert and key on first run):
```bash
socktop_agent --enableSSL --port 8443
```
Client (trust/pin the server cert; copy cert.pem from the agent):
```bash
socktop --tls-ca /path/to/agent/cert.pem wss://HOST:8443/ws
```
Notes:
- Do not copy the private key off the server; only the cert.pem is needed by clients.
- When --tls-ca/-t is supplied, the client autoupgrades ws:// to wss:// to avoid protocol mismatch.
- Hostname (SAN) verification is DISABLED by default (the cert is still pinned). Use `--verify-hostname` to enable strict SAN checking.
- You can run multiple clients with different cert paths by passing --tls-ca per invocation.
---
## Using tmux to monitor multiple hosts
@@ -319,9 +510,22 @@ Tips:
cargo fmt
cargo clippy --all-targets --all-features
cargo run -p socktop -- ws://127.0.0.1:3000/ws
cargo run -p socktop_agent -- --port 3000
# TLS (dev): first run will create certs under ~/.config/socktop_agent/tls/
cargo run -p socktop_agent -- --enableSSL --port 8443
```
### Auto-format on commit
A sample pre-commit hook that runs `cargo fmt --all` is provided in `.githooks/pre-commit`.
Enable it (one-time):
```bash
git config core.hooksPath .githooks
chmod +x .githooks/pre-commit
```
Every commit will then format Rust sources and restage them automatically.
---
## Roadmap
@@ -329,10 +533,13 @@ cargo run -p socktop_agent -- --port 3000
- [x] Agent authentication (token)
- [x] Hide per-thread entries; only show processes
- [x] Sort top processes in the TUI
- [ ] Configurable refresh intervals (client)
- [x] Configurable refresh intervals (client)
- [ ] Export metrics to file
- [ ] TLS / WSS support
- [x] TLS / WSS support (selfsigned server cert + client pinning)
- [x] Split processes/disks to separate WS calls with independent cadences (already logical on client; formalize API)
- [ ] Outage notifications and reconnect.
- [ ] Per process detailed statistics pane
- [ ] cleanup of Disks section, properly display physical disks / partitions, remove duplicate entries
---
+204
View File
@@ -0,0 +1,204 @@
# Cross-Compiling socktop_agent for Raspberry Pi
This guide explains how to cross-compile the socktop_agent on various host systems and deploy it to a Raspberry Pi. Cross-compiling is particularly useful for older or resource-constrained Pi models where native compilation might be slow.
## Cross-Compilation Host Setup
Choose your host operating system:
- [Debian/Ubuntu](#debianubuntu-based-systems)
- [Arch Linux](#arch-linux-based-systems)
- [macOS](#macos)
- [Windows](#windows)
## Debian/Ubuntu Based Systems
### Prerequisites
Install the cross-compilation toolchain for your target Raspberry Pi architecture:
```bash
# For 64-bit Raspberry Pi (aarch64)
sudo apt update
sudo apt install gcc-aarch64-linux-gnu libc6-dev-arm64-cross libdrm-dev:arm64
# For 32-bit Raspberry Pi (armv7)
sudo apt update
sudo apt install gcc-arm-linux-gnueabihf libc6-dev-armhf-cross libdrm-dev:armhf
```
### Setup Rust Cross-Compilation Targets
```bash
# For 64-bit Raspberry Pi
rustup target add aarch64-unknown-linux-gnu
# For 32-bit Raspberry Pi
rustup target add armv7-unknown-linux-gnueabihf
```
### Configure Cargo for Cross-Compilation
Create or edit `~/.cargo/config.toml`:
```toml
[target.aarch64-unknown-linux-gnu]
linker = "aarch64-linux-gnu-gcc"
[target.armv7-unknown-linux-gnueabihf]
linker = "arm-linux-gnueabihf-gcc"
```
## Arch Linux Based Systems
### Prerequisites
Install the cross-compilation toolchain using pacman and AUR:
```bash
# Install base dependencies
sudo pacman -S base-devel
# For 64-bit Raspberry Pi (aarch64)
sudo pacman -S aarch64-linux-gnu-gcc
# Install libdrm for aarch64 using an AUR helper (e.g., yay, paru)
yay -S aarch64-linux-gnu-libdrm
# For 32-bit Raspberry Pi (armv7)
sudo pacman -S arm-linux-gnueabihf-gcc
# Install libdrm for armv7 using an AUR helper
yay -S arm-linux-gnueabihf-libdrm
```
### Setup Rust Cross-Compilation Targets
```bash
# For 64-bit Raspberry Pi
rustup target add aarch64-unknown-linux-gnu
# For 32-bit Raspberry Pi
rustup target add armv7-unknown-linux-gnueabihf
```
### Configure Cargo for Cross-Compilation
Create or edit `~/.cargo/config.toml`:
```toml
[target.aarch64-unknown-linux-gnu]
linker = "aarch64-linux-gnu-gcc"
[target.armv7-unknown-linux-gnueabihf]
linker = "arm-linux-gnueabihf-gcc"
```
## macOS
The recommended approach for cross-compiling from macOS is to use Docker:
```bash
# Install Docker
brew install --cask docker
# Pull a cross-compilation Docker image
docker pull messense/rust-musl-cross:armv7-musleabihf # For 32-bit Pi
docker pull messense/rust-musl-cross:aarch64-musl # For 64-bit Pi
```
### Using Docker for Cross-Compilation
```bash
# Navigate to your socktop project directory
cd path/to/socktop
# For 64-bit Raspberry Pi
docker run --rm -it -v "$(pwd)":/home/rust/src messense/rust-musl-cross:aarch64-musl cargo build --release --target aarch64-unknown-linux-musl -p socktop_agent
# For 32-bit Raspberry Pi
docker run --rm -it -v "$(pwd)":/home/rust/src messense/rust-musl-cross:armv7-musleabihf cargo build --release --target armv7-unknown-linux-musleabihf -p socktop_agent
```
The compiled binaries will be available in your local target directory.
## Windows
The recommended approach for Windows is to use Windows Subsystem for Linux (WSL2):
1. Install WSL2 with a Debian/Ubuntu distribution by following the [official Microsoft documentation](https://docs.microsoft.com/en-us/windows/wsl/install).
2. Once WSL2 is set up with a Debian/Ubuntu distribution, open your WSL terminal and follow the [Debian/Ubuntu instructions](#debianubuntu-based-systems) above.
## Cross-Compile the Agent
After setting up your environment, build the socktop_agent for your target Raspberry Pi:
```bash
# For 64-bit Raspberry Pi
cargo build --release --target aarch64-unknown-linux-gnu -p socktop_agent
# For 32-bit Raspberry Pi
cargo build --release --target armv7-unknown-linux-gnueabihf -p socktop_agent
```
## Transfer the Binary to Your Raspberry Pi
Use SCP to transfer the compiled binary to your Raspberry Pi:
```bash
# For 64-bit Raspberry Pi
scp target/aarch64-unknown-linux-gnu/release/socktop_agent pi@raspberry-pi-ip:~/
# For 32-bit Raspberry Pi
scp target/armv7-unknown-linux-gnueabihf/release/socktop_agent pi@raspberry-pi-ip:~/
```
Replace `raspberry-pi-ip` with your Raspberry Pi's IP address and `pi` with your username.
## Install Dependencies on the Raspberry Pi
SSH into your Raspberry Pi and install the required dependencies:
```bash
ssh pi@raspberry-pi-ip
# For Raspberry Pi OS (Debian-based)
sudo apt update
sudo apt install libdrm-dev libdrm-amdgpu1
# For Arch Linux ARM
sudo pacman -Syu
sudo pacman -S libdrm
```
## Make the Binary Executable and Install
```bash
chmod +x ~/socktop_agent
# Optional: Install system-wide
sudo install -o root -g root -m 0755 ~/socktop_agent /usr/local/bin/socktop_agent
# Optional: Set up as a systemd service
sudo install -o root -g root -m 0644 ~/socktop-agent.service /etc/systemd/system/socktop-agent.service
sudo systemctl daemon-reload
sudo systemctl enable --now socktop-agent
```
## Troubleshooting
If you encounter issues with the cross-compiled binary:
1. **Incorrect Architecture**: Ensure you've chosen the correct target for your Raspberry Pi model:
- For Raspberry Pi 2: use `armv7-unknown-linux-gnueabihf`
- For Raspberry Pi 3/4/5 in 64-bit mode: use `aarch64-unknown-linux-gnu`
- For Raspberry Pi 3/4/5 in 32-bit mode: use `armv7-unknown-linux-gnueabihf`
2. **Dependency Issues**: Check for missing libraries:
```bash
ldd ~/socktop_agent
```
3. **Run with Backtrace**: Get detailed error information:
```bash
RUST_BACKTRACE=1 ~/socktop_agent
```
Binary file not shown.

After

Width:  |  Height:  |  Size: 47 MiB

+8
View File
@@ -0,0 +1,8 @@
socktop_agent: TLS enabled. Listening on wss://0.0.0.0:8433/ws
socktop_agent: TLS enabled. Listening on wss://0.0.0.0:8433/ws
socktop_agent: TLS enabled. Listening on wss://0.0.0.0:8433/ws
Error: Address already in use (os error 98)
socktop_agent: TLS enabled. Listening on wss://0.0.0.0:8433/ws
Error: Address already in use (os error 98)
socktop_agent: TLS enabled. Listening on wss://0.0.0.0:8443/ws
socktop_agent: TLS enabled. Listening on wss://0.0.0.0:8443/ws
+15
View File
@@ -0,0 +1,15 @@
syntax = "proto3";
package socktop;
// All running processes. Sorting is done client-side.
message Processes {
uint64 process_count = 1; // total processes in the system
repeated Process rows = 2; // all processes
}
message Process {
uint32 pid = 1;
string name = 2;
float cpu_usage = 3; // 0..100
uint64 mem_bytes = 4; // RSS bytes
}
+3
View File
@@ -0,0 +1,3 @@
[toolchain]
channel = "stable"
components = ["clippy", "rustfmt"]
+47
View File
@@ -0,0 +1,47 @@
#!/usr/bin/env bash
set -euo pipefail
# Cross-check Windows build from Linux using the GNU (MinGW) toolchain.
# - Ensures target `x86_64-pc-windows-gnu` is installed
# - Verifies MinGW cross-compiler is available (x86_64-w64-mingw32-gcc)
# - Runs cargo clippy with warnings-as-errors for the Windows target
# - Builds release binaries for the Windows target
echo "[socktop] Windows cross-check: clippy + build (GNU target)"
have() { command -v "$1" >/dev/null 2>&1; }
if ! have rustup; then
echo "error: rustup not found. Install Rust via rustup first (see README)." >&2
exit 1
fi
if ! rustup target list --installed | grep -q '^x86_64-pc-windows-gnu$'; then
echo "+ rustup target add x86_64-pc-windows-gnu"
rustup target add x86_64-pc-windows-gnu
fi
if ! have x86_64-w64-mingw32-gcc; then
echo "error: Missing MinGW cross-compiler (x86_64-w64-mingw32-gcc)." >&2
if have pacman; then
echo "Arch Linux: sudo pacman -S --needed mingw-w64-gcc" >&2
elif have apt-get; then
echo "Debian/Ubuntu: sudo apt-get install -y mingw-w64" >&2
elif have dnf; then
echo "Fedora: sudo dnf install -y mingw64-gcc" >&2
else
echo "Install the mingw-w64 toolchain for your distro, then re-run." >&2
fi
exit 1
fi
CARGO_FLAGS=(--workspace --all-targets --all-features --target x86_64-pc-windows-gnu)
echo "+ cargo clippy ${CARGO_FLAGS[*]} -- -D warnings"
cargo clippy "${CARGO_FLAGS[@]}" -- -D warnings
echo "+ cargo build --release ${CARGO_FLAGS[*]}"
cargo build --release "${CARGO_FLAGS[@]}"
echo "✅ Windows clippy and build completed successfully."
+43
View File
@@ -0,0 +1,43 @@
#!/usr/bin/env bash
set -euo pipefail
# Publish job: "publish new socktop agent version"
# Usage: ./scripts/publish_socktop_agent.sh <new_version>
if [[ ${1:-} == "" ]]; then
echo "Usage: $0 <new_version>" >&2
exit 1
fi
NEW_VERSION="$1"
ROOT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd)
CRATE_DIR="$ROOT_DIR/socktop_agent"
echo "==> Formatting socktop_agent"
(cd "$ROOT_DIR" && cargo fmt -p socktop_agent)
echo "==> Running tests for socktop_agent"
(cd "$ROOT_DIR" && cargo test -p socktop_agent)
echo "==> Running clippy (warnings as errors) for socktop_agent"
(cd "$ROOT_DIR" && cargo clippy -p socktop_agent -- -D warnings)
echo "==> Building release for socktop_agent"
(cd "$ROOT_DIR" && cargo build -p socktop_agent --release)
echo "==> Bumping version to $NEW_VERSION in socktop_agent/Cargo.toml"
sed -i.bak -E "s/^version = \"[0-9]+\.[0-9]+\.[0-9]+\"/version = \"$NEW_VERSION\"/" "$CRATE_DIR/Cargo.toml"
rm -f "$CRATE_DIR/Cargo.toml.bak"
echo "==> Committing version bump"
(cd "$ROOT_DIR" && git add -A && git commit -m "socktop_agent: bump version to $NEW_VERSION")
CURRENT_BRANCH=$(cd "$ROOT_DIR" && git rev-parse --abbrev-ref HEAD)
echo "==> Pushing to origin $CURRENT_BRANCH"
(cd "$ROOT_DIR" && git push origin "$CURRENT_BRANCH")
echo "==> Publishing socktop_agent $NEW_VERSION to crates.io"
(cd "$ROOT_DIR" && cargo publish -p socktop_agent)
echo "==> Done: socktop_agent $NEW_VERSION published"
+15 -4
View File
@@ -1,22 +1,33 @@
[package]
name = "socktop"
version = "0.1.1"
version = "1.40.0"
authors = ["Jason Witty <jasonpwitty+socktop@proton.me>"]
description = "Remote system monitor over WebSocket, TUI like top"
edition = "2021"
license = "MIT"
readme = "README.md"
[dependencies]
tokio = { workspace = true }
tokio-tungstenite = { workspace = true }
futures = { workspace = true }
futures-util = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
url = { workspace = true }
ratatui = { workspace = true }
crossterm = { workspace = true }
chrono = { workspace = true }
anyhow = { workspace = true }
flate2 = { version = "1", default-features = false, features = ["rust_backend"] }
tungstenite = "0.27.0"
dirs-next = { workspace = true }
sysinfo = { workspace = true }
rustls = "0.23"
rustls-pemfile = "2.1"
prost = { workspace = true }
[dev-dependencies]
assert_cmd = "2.0"
tempfile = "3"
[build-dependencies]
prost-build = "0.13"
protoc-bin-vendored = "3"
+26
View File
@@ -0,0 +1,26 @@
# socktop (client)
Minimal TUI client for the socktop remote monitoring agent.
Features:
- Connects to a socktop_agent over WebSocket / secure WebSocket
- Displays CPU, memory, swap, disks, network, processes, (optional) GPU metrics
- Selfsigned TLS cert pinning via --tls-ca
- Profile management with saved intervals
- Low CPU usage (request-driven updates)
Quick start:
```
cargo install socktop
socktop ws://HOST:3000/ws
```
With TLS (copy agent cert first):
```
socktop --tls-ca cert.pem wss://HOST:8443/ws
```
Demo mode (spawns a local agent automatically on first run prompt):
```
socktop --demo
```
Full documentation, screenshots, and advanced usage:
https://github.com/jasonwitty/socktop
+14
View File
@@ -0,0 +1,14 @@
fn main() {
// Vendored protoc for reproducible builds (works on crates.io build machines)
let protoc = protoc_bin_vendored::protoc_bin_path().expect("protoc");
std::env::set_var("PROTOC", &protoc);
// Tell Cargo when to re-run
println!("cargo:rerun-if-changed=proto/processes.proto");
let mut cfg = prost_build::Config::new();
cfg.out_dir(std::env::var("OUT_DIR").unwrap());
// Use in-crate relative path so `cargo package` includes the file
cfg.compile_protos(&["proto/processes.proto"], &["proto"]) // paths relative to CARGO_MANIFEST_DIR
.expect("compile protos");
}
+15
View File
@@ -0,0 +1,15 @@
syntax = "proto3";
package socktop;
// All running processes. Sorting is done client-side.
message Processes {
uint64 process_count = 1; // total processes in the system
repeated Process rows = 2; // all processes
}
message Process {
uint32 pid = 1;
string name = 2;
float cpu_usage = 3; // 0..100
uint64 mem_bytes = 4; // RSS bytes
}
+49 -4
View File
@@ -63,6 +63,13 @@ pub struct App {
last_disks_poll: Instant,
procs_interval: Duration,
disks_interval: Duration,
metrics_interval: Duration,
// For reconnects
ws_url: String,
// Security / status flags
pub is_tls: bool,
pub has_token: bool,
}
impl App {
@@ -91,12 +98,38 @@ impl App {
.unwrap_or_else(Instant::now),
procs_interval: Duration::from_secs(2),
disks_interval: Duration::from_secs(5),
metrics_interval: Duration::from_millis(500),
ws_url: String::new(),
is_tls: false,
has_token: false,
}
}
pub async fn run(&mut self, url: &str) -> Result<(), Box<dyn std::error::Error>> {
pub fn with_intervals(mut self, metrics_ms: Option<u64>, procs_ms: Option<u64>) -> Self {
if let Some(m) = metrics_ms {
self.metrics_interval = Duration::from_millis(m.max(100));
}
if let Some(p) = procs_ms {
self.procs_interval = Duration::from_millis(p.max(200));
}
self
}
pub fn with_status(mut self, is_tls: bool, has_token: bool) -> Self {
self.is_tls = is_tls;
self.has_token = has_token;
self
}
pub async fn run(
&mut self,
url: &str,
tls_ca: Option<&str>,
) -> Result<(), Box<dyn std::error::Error>> {
// Connect to agent
let mut ws = connect(url).await?;
//let mut ws = connect(url, tls_ca).await?;
self.ws_url = url.to_string();
let mut ws = connect(url, tls_ca).await?;
// Terminal setup
enable_raw_mode()?;
@@ -274,7 +307,7 @@ impl App {
terminal.draw(|f| self.draw(f))?;
// Tick rate
sleep(Duration::from_millis(500)).await;
sleep(self.metrics_interval).await;
}
Ok(())
@@ -341,7 +374,15 @@ impl App {
.split(area);
// Header
draw_header(f, rows[0], self.last_metrics.as_ref());
draw_header(
f,
rows[0],
self.last_metrics.as_ref(),
self.is_tls,
self.has_token,
self.metrics_interval,
self.procs_interval,
);
// Top row: left CPU avg, right Per-core (full top-right)
let top_lr = ratatui::layout::Layout::default()
@@ -461,6 +502,10 @@ impl Default for App {
.unwrap_or_else(Instant::now),
procs_interval: Duration::from_secs(2),
disks_interval: Duration::from_secs(5),
metrics_interval: Duration::from_millis(500),
ws_url: String::new(),
is_tls: false,
has_token: false,
}
}
}
+4
View File
@@ -0,0 +1,4 @@
//! Library surface for integration tests and reuse.
pub mod types;
pub mod ws;
+412 -12
View File
@@ -2,29 +2,429 @@
mod app;
mod history;
mod profiles;
mod types;
mod ui;
mod ws;
use app::App;
use profiles::{load_profiles, save_profiles, ProfileEntry, ProfileRequest, ResolveProfile};
use std::env;
use std::io::{self, Write};
pub(crate) struct ParsedArgs {
url: Option<String>,
tls_ca: Option<String>,
profile: Option<String>,
save: bool,
demo: bool,
dry_run: bool, // hidden test helper: skip connecting
metrics_interval_ms: Option<u64>,
processes_interval_ms: Option<u64>,
verify_hostname: bool,
}
pub(crate) fn parse_args<I: IntoIterator<Item = String>>(args: I) -> Result<ParsedArgs, String> {
let mut it = args.into_iter();
let prog = it.next().unwrap_or_else(|| "socktop".into());
let mut url: Option<String> = None;
let mut tls_ca: Option<String> = None;
let mut profile: Option<String> = None;
let mut save = false;
let mut demo = false;
let mut dry_run = false;
let mut metrics_interval_ms: Option<u64> = None;
let mut processes_interval_ms: Option<u64> = None;
let mut verify_hostname = false;
while let Some(arg) = it.next() {
match arg.as_str() {
"-h" | "--help" => {
return Err(format!("Usage: {prog} [--tls-ca CERT_PEM|-t CERT_PEM] [--verify-hostname] [--profile NAME|-P NAME] [--save] [--demo] [--metrics-interval-ms N] [--processes-interval-ms N] [ws://HOST:PORT/ws]\n"));
}
"--tls-ca" | "-t" => {
tls_ca = it.next();
}
"--verify-hostname" => {
// opt-in hostname (SAN) verification
// default behavior is to skip it for easier home network usage
// (still pins the provided certificate)
verify_hostname = true;
}
"--profile" | "-P" => {
profile = it.next();
}
"--save" => {
save = true;
}
"--demo" => {
demo = true;
}
"--dry-run" => {
// intentionally undocumented
dry_run = true;
}
"--metrics-interval-ms" => {
metrics_interval_ms = it.next().and_then(|v| v.parse().ok());
}
"--processes-interval-ms" => {
processes_interval_ms = it.next().and_then(|v| v.parse().ok());
}
_ if arg.starts_with("--tls-ca=") => {
if let Some((_, v)) = arg.split_once('=') {
if !v.is_empty() {
tls_ca = Some(v.to_string());
}
}
}
_ if arg.starts_with("--profile=") => {
if let Some((_, v)) = arg.split_once('=') {
if !v.is_empty() {
profile = Some(v.to_string());
}
}
}
_ if arg.starts_with("--metrics-interval-ms=") => {
if let Some((_, v)) = arg.split_once('=') {
metrics_interval_ms = v.parse().ok();
}
}
_ if arg.starts_with("--processes-interval-ms=") => {
if let Some((_, v)) = arg.split_once('=') {
processes_interval_ms = v.parse().ok();
}
}
_ => {
if url.is_none() {
url = Some(arg);
} else {
return Err(format!("Unexpected argument. Usage: {prog} [--tls-ca CERT_PEM|-t CERT_PEM] [--verify-hostname] [--profile NAME|-P NAME] [--save] [--demo] [ws://HOST:PORT/ws]"));
}
}
}
}
Ok(ParsedArgs {
url,
tls_ca,
profile,
save,
demo,
dry_run,
metrics_interval_ms,
processes_interval_ms,
verify_hostname,
})
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let mut args = env::args();
let prog = args.next().unwrap_or_else(|| "socktop".into());
let url = match args.next() {
Some(flag) if flag == "-h" || flag == "--help" => {
println!("Usage: {prog} ws://HOST:PORT/ws");
let parsed = match parse_args(env::args()) {
Ok(v) => v,
Err(msg) => {
eprintln!("{msg}");
return Ok(());
}
Some(url) => url,
None => {
eprintln!("Usage: {prog} ws://HOST:PORT/ws");
std::process::exit(1);
}
};
let mut app = App::new();
app.run(&url).await
//support version flag (print and exit)
if env::args().any(|a| a == "--version" || a == "-V") {
println!("socktop {}", env!("CARGO_PKG_VERSION"));
return Ok(());
}
if parsed.demo || matches!(parsed.profile.as_deref(), Some("demo")) {
return run_demo_mode(parsed.tls_ca.as_deref()).await;
}
if parsed.verify_hostname {
// Set env var consumed by ws::connect logic
std::env::set_var("SOCKTOP_VERIFY_NAME", "1");
}
let profiles_file = load_profiles();
let req = ProfileRequest {
profile_name: parsed.profile.clone(),
url: parsed.url.clone(),
tls_ca: parsed.tls_ca.clone(),
};
let resolved = req.resolve(&profiles_file);
let mut profiles_mut = profiles_file.clone();
let (url, tls_ca, metrics_interval_ms, processes_interval_ms): (
String,
Option<String>,
Option<u64>,
Option<u64>,
) = match resolved {
ResolveProfile::Direct(u, t) => {
if let Some(name) = parsed.profile.as_ref() {
let existing = profiles_mut.profiles.get(name);
match existing {
None => {
let (mi, pi) = gather_intervals(
parsed.metrics_interval_ms,
parsed.processes_interval_ms,
)?;
profiles_mut.profiles.insert(
name.clone(),
ProfileEntry {
url: u.clone(),
tls_ca: t.clone(),
metrics_interval_ms: mi,
processes_interval_ms: pi,
},
);
let _ = save_profiles(&profiles_mut);
(u, t, mi, pi)
}
Some(entry) => {
let changed = entry.url != u || entry.tls_ca != t;
if changed {
let overwrite = if parsed.save {
true
} else {
prompt_yes_no(&format!(
"Overwrite existing profile '{name}'? [y/N]: "
))
};
if overwrite {
let (mi, pi) = gather_intervals(
parsed.metrics_interval_ms,
parsed.processes_interval_ms,
)?;
profiles_mut.profiles.insert(
name.clone(),
ProfileEntry {
url: u.clone(),
tls_ca: t.clone(),
metrics_interval_ms: mi,
processes_interval_ms: pi,
},
);
let _ = save_profiles(&profiles_mut);
(u, t, mi, pi)
} else {
(u, t, entry.metrics_interval_ms, entry.processes_interval_ms)
}
} else {
(u, t, entry.metrics_interval_ms, entry.processes_interval_ms)
}
}
}
} else {
(
u,
t,
parsed.metrics_interval_ms,
parsed.processes_interval_ms,
)
}
}
ResolveProfile::Loaded(u, t) => {
let entry = profiles_mut
.profiles
.get(parsed.profile.as_ref().unwrap())
.unwrap();
(u, t, entry.metrics_interval_ms, entry.processes_interval_ms)
}
ResolveProfile::PromptSelect(mut names) => {
if !names.iter().any(|n: &String| n == "demo") {
names.push("demo".into());
}
eprintln!("Select profile:");
for (i, n) in names.iter().enumerate() {
eprintln!(" {}. {}", i + 1, n);
}
eprint!("Enter number (or blank to abort): ");
let _ = io::stderr().flush();
let mut line = String::new();
if io::stdin().read_line(&mut line).is_ok() {
if let Ok(idx) = line.trim().parse::<usize>() {
if idx >= 1 && idx <= names.len() {
let name = &names[idx - 1];
if name == "demo" {
return run_demo_mode(parsed.tls_ca.as_deref()).await;
}
if let Some(entry) = profiles_mut.profiles.get(name) {
(
entry.url.clone(),
entry.tls_ca.clone(),
entry.metrics_interval_ms,
entry.processes_interval_ms,
)
} else {
return Ok(());
}
} else {
return Ok(());
}
} else {
return Ok(());
}
} else {
return Ok(());
}
}
ResolveProfile::PromptCreate(name) => {
eprintln!("Profile '{name}' does not exist yet.");
let url = prompt_string("Enter URL (ws://HOST:PORT/ws or wss://...): ")?;
if url.trim().is_empty() {
return Ok(());
}
let ca = prompt_string("Enter TLS CA path (or leave blank): ")?;
let ca_opt = if ca.trim().is_empty() {
None
} else {
Some(ca.trim().to_string())
};
let (mi, pi) =
gather_intervals(parsed.metrics_interval_ms, parsed.processes_interval_ms)?;
profiles_mut.profiles.insert(
name.clone(),
ProfileEntry {
url: url.trim().to_string(),
tls_ca: ca_opt.clone(),
metrics_interval_ms: mi,
processes_interval_ms: pi,
},
);
let _ = save_profiles(&profiles_mut);
(url.trim().to_string(), ca_opt, mi, pi)
}
ResolveProfile::None => {
//eprintln!("No URL provided and no profiles to select.");
//first run, no args, no profiles: show welcome message and offer demo mode
if profiles_mut.profiles.is_empty() && parsed.url.is_none() {
eprintln!("Welcome to socktop!");
eprintln!("It looks like this is your first time running the application.");
eprintln!("You can connect to a socktop_agent instance to monitor system metrics and processes.");
eprintln!("If you don't have an agent running, you can try the demo mode.");
if prompt_yes_no("Would you like to start the demo mode now? [Y/n]: ") {
return run_demo_mode(parsed.tls_ca.as_deref()).await;
} else {
eprintln!("Aborting. You can run 'socktop --help' for usage information.");
return Ok(());
}
}
return Err("No URL provided and no profiles to select.".into());
}
};
let is_tls = url.starts_with("wss://");
let has_token = url.contains("token=");
let mut app = App::new()
.with_intervals(metrics_interval_ms, processes_interval_ms)
.with_status(is_tls, has_token);
if parsed.dry_run {
return Ok(());
}
app.run(&url, tls_ca.as_deref()).await
}
fn prompt_yes_no(prompt: &str) -> bool {
eprint!("{prompt}");
let _ = io::stderr().flush();
let mut line = String::new();
if io::stdin().read_line(&mut line).is_ok() {
matches!(line.trim().to_ascii_lowercase().as_str(), "y" | "yes")
} else {
false
}
}
fn prompt_string(prompt: &str) -> io::Result<String> {
eprint!("{prompt}");
let _ = io::stderr().flush();
let mut line = String::new();
io::stdin().read_line(&mut line)?;
Ok(line)
}
fn gather_intervals(
arg_metrics: Option<u64>,
arg_procs: Option<u64>,
) -> Result<(Option<u64>, Option<u64>), Box<dyn std::error::Error>> {
let default_metrics = 500u64;
let default_procs = 2000u64;
let metrics = match arg_metrics {
Some(v) => Some(v),
None => {
let inp = prompt_string(&format!(
"Metrics interval ms (default {default_metrics}, Enter for default): "
))?;
let t = inp.trim();
if t.is_empty() {
Some(default_metrics)
} else {
Some(t.parse()?)
}
}
};
let procs = match arg_procs {
Some(v) => Some(v),
None => {
let inp = prompt_string(&format!(
"Processes interval ms (default {default_procs}, Enter for default): "
))?;
let t = inp.trim();
if t.is_empty() {
Some(default_procs)
} else {
Some(t.parse()?)
}
}
};
Ok((metrics, procs))
}
// Demo mode implementation
async fn run_demo_mode(_tls_ca: Option<&str>) -> Result<(), Box<dyn std::error::Error>> {
let port = 3231;
let url = format!("ws://127.0.0.1:{port}/ws");
let child = spawn_demo_agent(port)?;
let mut app = App::new();
tokio::select! { res=app.run(&url,None)=>{ drop(child); res } _=tokio::signal::ctrl_c()=>{ drop(child); Ok(()) } }
}
struct DemoGuard {
port: u16,
child: std::sync::Arc<std::sync::Mutex<Option<std::process::Child>>>,
}
impl Drop for DemoGuard {
fn drop(&mut self) {
if let Some(mut ch) = self.child.lock().unwrap().take() {
let _ = ch.kill();
}
eprintln!("Stopped demo agent on port {}", self.port);
}
}
fn spawn_demo_agent(port: u16) -> Result<DemoGuard, Box<dyn std::error::Error>> {
let candidate = find_agent_executable();
let mut cmd = std::process::Command::new(candidate);
cmd.arg("--port").arg(port.to_string());
cmd.env("SOCKTOP_ENABLE_SSL", "0");
//JW: do not disable GPU and TEMP in demo mode
//cmd.env("SOCKTOP_AGENT_GPU", "0");
//cmd.env("SOCKTOP_AGENT_TEMP", "0");
let child = cmd.spawn()?;
std::thread::sleep(std::time::Duration::from_millis(300));
Ok(DemoGuard {
port,
child: std::sync::Arc::new(std::sync::Mutex::new(Some(child))),
})
}
fn find_agent_executable() -> std::path::PathBuf {
if let Ok(exe) = std::env::current_exe() {
if let Some(parent) = exe.parent() {
#[cfg(windows)]
let name = "socktop_agent.exe";
#[cfg(not(windows))]
let name = "socktop_agent";
let candidate = parent.join(name);
if candidate.exists() {
return candidate;
}
}
}
std::path::PathBuf::from("socktop_agent")
}
+102
View File
@@ -0,0 +1,102 @@
//! Connection profiles: load/save simple JSON mapping of profile name -> { url, tls_ca }
//! Stored under XDG config dir: $XDG_CONFIG_HOME/socktop/profiles.json (fallback ~/.config/socktop/profiles.json)
use serde::{Deserialize, Serialize};
use std::{collections::BTreeMap, fs, path::PathBuf};
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct ProfileEntry {
pub url: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub tls_ca: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub metrics_interval_ms: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub processes_interval_ms: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct ProfilesFile {
#[serde(default)]
pub profiles: BTreeMap<String, ProfileEntry>,
#[serde(default)]
pub version: u32,
}
pub fn config_dir() -> PathBuf {
if let Some(xdg) = std::env::var_os("XDG_CONFIG_HOME") {
PathBuf::from(xdg).join("socktop")
} else {
dirs_next::config_dir()
.unwrap_or_else(|| PathBuf::from("."))
.join("socktop")
}
}
pub fn profiles_path() -> PathBuf {
config_dir().join("profiles.json")
}
pub fn load_profiles() -> ProfilesFile {
let path = profiles_path();
match fs::read_to_string(&path) {
Ok(s) => serde_json::from_str(&s).unwrap_or_default(),
Err(_) => ProfilesFile::default(),
}
}
pub fn save_profiles(p: &ProfilesFile) -> std::io::Result<()> {
let path = profiles_path();
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
let data = serde_json::to_vec_pretty(p).expect("serialize profiles");
fs::write(path, data)
}
pub enum ResolveProfile {
/// Use the provided runtime inputs (not persisted). (url, tls_ca)
Direct(String, Option<String>),
/// Loaded from existing profile entry (url, tls_ca)
Loaded(String, Option<String>),
/// Should prompt user to select among profile names
PromptSelect(Vec<String>),
/// Should prompt user to create a new profile (name)
PromptCreate(String),
/// No profile could be resolved (e.g., missing arguments)
None,
}
pub struct ProfileRequest {
pub profile_name: Option<String>,
pub url: Option<String>,
pub tls_ca: Option<String>,
}
impl ProfileRequest {
pub fn resolve(self, pf: &ProfilesFile) -> ResolveProfile {
// Case: only profile name given -> try load
if self.url.is_none() && self.profile_name.is_some() {
let name = self.profile_name.unwrap();
if let Some(entry) = pf.profiles.get(&name) {
return ResolveProfile::Loaded(entry.url.clone(), entry.tls_ca.clone());
} else {
return ResolveProfile::PromptCreate(name);
}
}
// Both provided -> direct (maybe later saved by caller)
if let Some(u) = self.url {
return ResolveProfile::Direct(u, self.tls_ca);
}
// Nothing provided -> maybe prompt select if profiles exist
if self.url.is_none() && self.profile_name.is_none() {
if pf.profiles.is_empty() {
ResolveProfile::None
} else {
ResolveProfile::PromptSelect(pf.profiles.keys().cloned().collect())
}
} else {
ResolveProfile::None
}
}
}
+27 -7
View File
@@ -5,9 +5,18 @@ use ratatui::{
layout::Rect,
widgets::{Block, Borders},
};
use std::time::Duration;
pub fn draw_header(f: &mut ratatui::Frame<'_>, area: Rect, m: Option<&Metrics>) {
let title = if let Some(mm) = m {
pub fn draw_header(
f: &mut ratatui::Frame<'_>,
area: Rect,
m: Option<&Metrics>,
is_tls: bool,
has_token: bool,
metrics_interval: Duration,
procs_interval: Duration,
) {
let base = if let Some(mm) = m {
let temp = mm
.cpu_temp_c
.map(|t| {
@@ -21,12 +30,23 @@ pub fn draw_header(f: &mut ratatui::Frame<'_>, area: Rect, m: Option<&Metrics>)
format!("CPU Temp: {t:.1}°C {icon}")
})
.unwrap_or_else(|| "CPU Temp: N/A".into());
format!(
"socktop — host: {} | {} (press 'q' to quit)",
mm.hostname, temp
)
format!("socktop — host: {} | {}", mm.hostname, temp)
} else {
"socktop — connecting... (press 'q' to quit)".into()
"socktop — connecting...".into()
};
// TLS indicator: lock vs lock with cross (using ✗). Keep explicit label for clarity.
let tls_txt = if is_tls { "🔒 TLS" } else { "🔒✗ TLS" };
// Token indicator
let tok_txt = if has_token { "🔑 token" } else { "" };
let mi = metrics_interval.as_millis();
let pi = procs_interval.as_millis();
let intervals = format!("{mi}ms metrics | {pi}ms procs");
let mut parts = vec![base, tls_txt.into()];
if !tok_txt.is_empty() {
parts.push(tok_txt.into());
}
parts.push(intervals);
parts.push("(q to quit)".into());
let title = parts.join(" | ");
f.render_widget(Block::default().title(title).borders(Borders::BOTTOM), area);
}
+137 -83
View File
@@ -2,18 +2,109 @@
use flate2::bufread::GzDecoder;
use futures_util::{SinkExt, StreamExt};
use prost::Message as _;
use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
use rustls::{ClientConfig, RootCertStore};
use rustls::{DigitallySignedStruct, SignatureScheme};
use rustls_pemfile::Item;
use std::io::Read;
use std::{fs::File, io::BufReader, sync::Arc};
use tokio::net::TcpStream;
use tokio::time::{interval, Duration};
use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
use tokio_tungstenite::{
connect_async, connect_async_tls_with_config, tungstenite::client::IntoClientRequest,
tungstenite::Message, Connector, MaybeTlsStream, WebSocketStream,
};
use url::Url;
use crate::types::{DiskInfo, Metrics, ProcessesPayload};
use crate::types::{DiskInfo, Metrics, ProcessInfo, ProcessesPayload};
mod pb {
// generated by build.rs
include!(concat!(env!("OUT_DIR"), "/socktop.rs"));
}
pub type WsStream = WebSocketStream<MaybeTlsStream<TcpStream>>;
// Connect to the agent and return the WS stream
pub async fn connect(url: &str) -> Result<WsStream, Box<dyn std::error::Error>> {
let (ws, _) = connect_async(url).await?;
pub async fn connect(
url: &str,
tls_ca: Option<&str>,
) -> Result<WsStream, Box<dyn std::error::Error>> {
let mut u = Url::parse(url)?;
if let Some(ca_path) = tls_ca {
if u.scheme() == "ws" {
let _ = u.set_scheme("wss");
}
return connect_with_ca(u.as_str(), ca_path).await;
}
let (ws, _) = connect_async(u.as_str()).await?;
Ok(ws)
}
async fn connect_with_ca(url: &str, ca_path: &str) -> Result<WsStream, Box<dyn std::error::Error>> {
let mut root = RootCertStore::empty();
let mut reader = BufReader::new(File::open(ca_path)?);
let mut der_certs = Vec::new();
while let Ok(Some(item)) = rustls_pemfile::read_one(&mut reader) {
if let Item::X509Certificate(der) = item {
der_certs.push(der);
}
}
root.add_parsable_certificates(der_certs);
let mut cfg = ClientConfig::builder()
.with_root_certificates(root)
.with_no_client_auth();
let req = url.into_client_request()?;
let verify_domain = std::env::var("SOCKTOP_VERIFY_NAME").ok().as_deref() == Some("1");
if !verify_domain {
#[derive(Debug)]
struct NoVerify;
impl ServerCertVerifier for NoVerify {
fn verify_server_cert(
&self,
_end_entity: &CertificateDer<'_>,
_intermediates: &[CertificateDer<'_>],
_server_name: &ServerName,
_ocsp_response: &[u8],
_now: UnixTime,
) -> Result<ServerCertVerified, rustls::Error> {
Ok(ServerCertVerified::assertion())
}
fn verify_tls12_signature(
&self,
_message: &[u8],
_cert: &CertificateDer<'_>,
_dss: &DigitallySignedStruct,
) -> Result<HandshakeSignatureValid, rustls::Error> {
Ok(HandshakeSignatureValid::assertion())
}
fn verify_tls13_signature(
&self,
_message: &[u8],
_cert: &CertificateDer<'_>,
_dss: &DigitallySignedStruct,
) -> Result<HandshakeSignatureValid, rustls::Error> {
Ok(HandshakeSignatureValid::assertion())
}
fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
// Provide common schemes; not strictly needed for skipping but keeps API happy
vec![
SignatureScheme::ECDSA_NISTP256_SHA256,
SignatureScheme::ED25519,
SignatureScheme::RSA_PSS_SHA256,
]
}
}
cfg.dangerous().set_certificate_verifier(Arc::new(NoVerify));
eprintln!("socktop: hostname verification disabled (default). Use --verify-hostname to enable strict SAN checking.");
}
let cfg = Arc::new(cfg);
let (ws, _) =
connect_async_tls_with_config(req, None, verify_domain, Some(Connector::Rustls(cfg)))
.await?;
Ok(ws)
}
@@ -39,6 +130,16 @@ fn gunzip_to_string(bytes: &[u8]) -> Option<String> {
Some(out)
}
fn gunzip_to_vec(bytes: &[u8]) -> Option<Vec<u8>> {
let mut dec = GzDecoder::new(bytes);
let mut out = Vec::new();
dec.read_to_end(&mut out).ok()?;
Some(out)
}
fn is_gzip(bytes: &[u8]) -> bool {
bytes.len() >= 2 && bytes[0] == 0x1f && bytes[1] == 0x8b
}
// Suppress dead_code until these are wired into the app
#[allow(dead_code)]
pub enum Payload {
@@ -47,23 +148,6 @@ pub enum Payload {
Processes(ProcessesPayload),
}
#[allow(dead_code)]
fn parse_any_payload(json: &str) -> Result<Payload, serde_json::Error> {
if let Ok(m) = serde_json::from_str::<Metrics>(json) {
return Ok(Payload::Metrics(m));
}
if let Ok(d) = serde_json::from_str::<Vec<DiskInfo>>(json) {
return Ok(Payload::Disks(d));
}
if let Ok(p) = serde_json::from_str::<ProcessesPayload>(json) {
return Ok(Payload::Processes(p));
}
Err(serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"unknown payload",
)))
}
// Send a "get_disks" request and await a JSON Vec<DiskInfo>
pub async fn request_disks(ws: &mut WsStream) -> Option<Vec<DiskInfo>> {
if ws.send(Message::Text("get_disks".into())).await.is_err() {
@@ -78,7 +162,7 @@ pub async fn request_disks(ws: &mut WsStream) -> Option<Vec<DiskInfo>> {
}
}
// Send a "get_processes" request and await a JSON ProcessesPayload
// Send a "get_processes" request and await a ProcessesPayload decoded from protobuf (binary, may be gzipped)
pub async fn request_processes(ws: &mut WsStream) -> Option<ProcessesPayload> {
if ws
.send(Message::Text("get_processes".into()))
@@ -89,68 +173,38 @@ pub async fn request_processes(ws: &mut WsStream) -> Option<ProcessesPayload> {
}
match ws.next().await {
Some(Ok(Message::Binary(b))) => {
gunzip_to_string(&b).and_then(|s| serde_json::from_str::<ProcessesPayload>(&s).ok())
let gz = is_gzip(&b);
let data = if gz { gunzip_to_vec(&b)? } else { b };
match pb::Processes::decode(data.as_slice()) {
Ok(pb) => {
let rows: Vec<ProcessInfo> = pb
.rows
.into_iter()
.map(|p: pb::Process| ProcessInfo {
pid: p.pid,
name: p.name,
cpu_usage: p.cpu_usage,
mem_bytes: p.mem_bytes,
})
.collect();
Some(ProcessesPayload {
process_count: pb.process_count as usize,
top_processes: rows,
})
}
Err(e) => {
if std::env::var("SOCKTOP_DEBUG").ok().as_deref() == Some("1") {
eprintln!("protobuf decode failed: {e}");
}
// Fallback: maybe it's JSON (bytes already decompressed if gz)
match String::from_utf8(data) {
Ok(s) => serde_json::from_str::<ProcessesPayload>(&s).ok(),
Err(_) => None,
}
}
}
}
Some(Ok(Message::Text(json))) => serde_json::from_str::<ProcessesPayload>(&json).ok(),
_ => None,
}
}
#[allow(dead_code)]
pub async fn start_ws_polling(mut ws: WsStream) {
let mut t_fast = interval(Duration::from_millis(500));
let mut t_procs = interval(Duration::from_secs(2));
let mut t_disks = interval(Duration::from_secs(5));
let _ = ws.send(Message::Text("get_metrics".into())).await;
let _ = ws.send(Message::Text("get_processes".into())).await;
let _ = ws.send(Message::Text("get_disks".into())).await;
loop {
tokio::select! {
_ = t_fast.tick() => {
let _ = ws.send(Message::Text("get_metrics".into())).await;
}
_ = t_procs.tick() => {
let _ = ws.send(Message::Text("get_processes".into())).await;
}
_ = t_disks.tick() => {
let _ = ws.send(Message::Text("get_disks".into())).await;
}
maybe = ws.next() => {
let Some(result) = maybe else { break; };
let Ok(msg) = result else { break; };
match msg {
Message::Binary(b) => {
if let Some(json) = gunzip_to_string(&b) {
if let Ok(payload) = parse_any_payload(&json) {
match payload {
Payload::Metrics(_m) => {
// update your app state with fast metrics
}
Payload::Disks(_d) => {
// update your app state with disks
}
Payload::Processes(_p) => {
// update your app state with processes
}
}
}
}
}
Message::Text(s) => {
if let Ok(payload) = parse_any_payload(&s) {
match payload {
Payload::Metrics(_m) => {}
Payload::Disks(_d) => {}
Payload::Processes(_p) => {}
}
}
}
Message::Close(_) => break,
_ => {}
}
}
}
}
}
+75
View File
@@ -0,0 +1,75 @@
//! CLI arg parsing tests for socktop (client)
use std::process::Command;
// We test the parsing by invoking the binary with --help and ensuring the help mentions short and long flags.
// Also directly test the parse_args function via a tiny helper in a doctest-like fashion using a small
// reimplementation here kept in sync with main (compile-time test).
#[test]
fn test_help_mentions_short_and_long_flags() {
let output = Command::new(env!("CARGO_BIN_EXE_socktop"))
.arg("--help")
.output()
.expect("run socktop --help");
let text = format!(
"{}{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
assert!(
text.contains("--tls-ca")
&& text.contains("-t")
&& text.contains("--profile")
&& text.contains("-P"),
"help text missing expected flags (--tls-ca/-t, --profile/-P)\n{text}"
);
}
#[test]
fn test_tlc_ca_arg_long_and_short_parsed() {
// Use --help combined with flags to avoid network and still exercise arg acceptance
let exe = env!("CARGO_BIN_EXE_socktop");
// Long form with help
let out = Command::new(exe)
.args(["--tls-ca", "/tmp/cert.pem", "--help"])
.output()
.expect("run socktop");
assert!(
out.status.success(),
"socktop --tls-ca … --help did not succeed"
);
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
assert!(text.contains("Usage:"));
// Short form with help
let out2 = Command::new(exe)
.args(["-t", "/tmp/cert.pem", "--help"])
.output()
.expect("run socktop");
assert!(out2.status.success(), "socktop -t … --help did not succeed");
let text2 = format!(
"{}{}",
String::from_utf8_lossy(&out2.stdout),
String::from_utf8_lossy(&out2.stderr)
);
assert!(text2.contains("Usage:"));
// Profile flags with help (should not error)
let out3 = Command::new(exe)
.args(["--profile", "dev", "--help"])
.output()
.expect("run socktop");
assert!(
out3.status.success(),
"socktop --profile dev --help did not succeed"
);
let text3 = format!(
"{}{}",
String::from_utf8_lossy(&out3.stdout),
String::from_utf8_lossy(&out3.stderr)
);
assert!(text3.contains("Usage:"));
}
+118
View File
@@ -0,0 +1,118 @@
//! Tests for profile load/save and resolution logic (non-interactive paths only)
use std::fs;
use std::sync::Mutex;
// Global lock to serialize tests that mutate process-wide environment variables.
static ENV_LOCK: Mutex<()> = Mutex::new(());
#[allow(dead_code)] // touch crate
fn touch() {
let _ = socktop::types::Metrics {
cpu_total: 0.0,
cpu_per_core: vec![],
mem_total: 0,
mem_used: 0,
swap_total: 0,
swap_used: 0,
process_count: None,
hostname: String::new(),
cpu_temp_c: None,
disks: vec![],
networks: vec![],
top_processes: vec![],
gpus: None,
};
}
// We re-import internal modules by copying minimal logic here because profiles.rs isn't public.
// Instead of exposing internals, we simulate profile saving through CLI invocations.
use std::process::Command;
fn run_socktop(args: &[&str]) -> (bool, String) {
let exe = env!("CARGO_BIN_EXE_socktop");
let output = Command::new(exe).args(args).output().expect("run socktop");
let ok = output.status.success();
let text = format!(
"{}{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
(ok, text)
}
fn config_dir() -> std::path::PathBuf {
if let Some(xdg) = std::env::var_os("XDG_CONFIG_HOME") {
std::path::PathBuf::from(xdg).join("socktop")
} else {
dirs_next::config_dir()
.unwrap_or_else(|| std::path::PathBuf::from("."))
.join("socktop")
}
}
fn profiles_path() -> std::path::PathBuf {
config_dir().join("profiles.json")
}
#[test]
fn test_profile_created_on_first_use() {
let _guard = ENV_LOCK.lock().unwrap();
// Isolate config in a temp dir
let td = tempfile::tempdir().unwrap();
std::env::set_var("XDG_CONFIG_HOME", td.path());
// Ensure directory exists fresh
std::fs::create_dir_all(td.path().join("socktop")).unwrap();
let _ = fs::remove_file(profiles_path());
// Provide profile + url => should create profiles.json
let (_ok, _out) = run_socktop(&["--profile", "unittest", "ws://example:1/ws", "--dry-run"]);
// We pass --help to exit early after parsing (no network attempt)
let data = fs::read_to_string(profiles_path()).expect("profiles.json created");
assert!(
data.contains("unittest"),
"profiles.json missing profile entry: {data}"
);
}
#[test]
fn test_profile_overwrite_only_when_changed() {
let _guard = ENV_LOCK.lock().unwrap();
let td = tempfile::tempdir().unwrap();
std::env::set_var("XDG_CONFIG_HOME", td.path());
std::fs::create_dir_all(td.path().join("socktop")).unwrap();
let _ = fs::remove_file(profiles_path());
// Initial create
let (_ok, _out) = run_socktop(&["--profile", "prod", "ws://one/ws", "--dry-run"]); // create
let first = fs::read_to_string(profiles_path()).unwrap();
// Re-run identical (should not duplicate or corrupt)
let (_ok2, _out2) = run_socktop(&["--profile", "prod", "ws://one/ws", "--dry-run"]); // identical
let second = fs::read_to_string(profiles_path()).unwrap();
assert_eq!(
first, second,
"Profile file changed despite identical input"
);
// Overwrite with different URL using --save (no prompt path)
let (_ok3, _out3) = run_socktop(&["--profile", "prod", "--save", "ws://two/ws", "--dry-run"]);
let third = fs::read_to_string(profiles_path()).unwrap();
assert!(third.contains("two"), "Updated URL not written: {third}");
}
#[test]
fn test_profile_tls_ca_persisted() {
let _guard = ENV_LOCK.lock().unwrap();
let td = tempfile::tempdir().unwrap();
std::env::set_var("XDG_CONFIG_HOME", td.path());
std::fs::create_dir_all(td.path().join("socktop")).unwrap();
let _ = fs::remove_file(profiles_path());
let (_ok, _out) = run_socktop(&[
"--profile",
"secureX",
"--tls-ca",
"/tmp/cert.pem",
"wss://host/ws",
"--dry-run",
]);
let data = fs::read_to_string(profiles_path()).unwrap();
assert!(data.contains("secureX"));
assert!(data.contains("cert.pem"));
}
+29
View File
@@ -0,0 +1,29 @@
use socktop::ws::{connect, request_metrics, request_processes};
// Integration probe: only runs when SOCKTOP_WS is set to an agent WebSocket URL.
// Example: SOCKTOP_WS=ws://127.0.0.1:3000/ws cargo test -p socktop --test ws_probe -- --nocapture
#[tokio::test]
async fn probe_ws_endpoints() {
// Gate the test to avoid CI failures when no agent is running.
let url = match std::env::var("SOCKTOP_WS") {
Ok(v) if !v.is_empty() => v,
_ => {
eprintln!(
"skipping ws_probe: set SOCKTOP_WS=ws://host:port/ws to run this integration test"
);
return;
}
};
// Optional pinned CA for WSS/self-signed setups
let tls_ca = std::env::var("SOCKTOP_TLS_CA").ok();
let mut ws = connect(&url, tls_ca.as_deref()).await.expect("connect ws");
// Should get fast metrics quickly
let m = request_metrics(&mut ws).await;
assert!(m.is_some(), "expected Metrics payload within timeout");
// Processes may be gzipped and a bit slower, but should arrive
let p = request_processes(&mut ws).await;
assert!(p.is_some(), "expected Processes payload within timeout");
}
+21 -5
View File
@@ -1,10 +1,11 @@
[package]
name = "socktop_agent"
version = "0.1.1"
version = "1.40.67"
authors = ["Jason Witty <jasonpwitty+socktop@proton.me>"]
description = "Remote system monitor over WebSocket, TUI like top"
edition = "2021"
license = "MIT"
readme = "README.md"
[dependencies]
tokio = { version = "1", features = ["full"] }
@@ -13,11 +14,26 @@ sysinfo = { version = "0.37", features = ["network", "disk", "component"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
flate2 = { version = "1", default-features = false, features = ["rust_backend"] }
futures = "0.3"
futures-util = "0.3.31"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
nvml-wrapper = "0.10"
# nvml-wrapper removed (unused; GPU metrics via gfxinfo only now)
gfxinfo = "0.1.2"
tungstenite = "0.27.0"
once_cell = "1.19"
once_cell = "1.19"
axum-server = { version = "0.6", features = ["tls-rustls"] }
rustls = "0.23"
rustls-pemfile = "2.1"
rcgen = "0.13" # pure-Rust self-signed cert generation (replaces openssl vendored build)
anyhow = "1"
hostname = "0.3"
prost = { workspace = true }
time = { version = "0.3", default-features = false, features = ["formatting", "macros", "parsing" ] }
[build-dependencies]
prost-build = "0.13"
tonic-build = { version = "0.12", default-features = false, optional = true }
protoc-bin-vendored = "3"
[dev-dependencies]
assert_cmd = "2.0"
tempfile = "3.10"
tokio-tungstenite = "0.21"
+392
View File
@@ -0,0 +1,392 @@
# socktop_agent (server)
Lightweight ondemand metrics WebSocket server for the socktop TUI.
Highlights:
- Collects system metrics only when requested (keeps idle CPU <1%)
- Optional TLS (selfsigned cert autogenerated & pinned by client)
- JSON for fast metrics / disks; protobuf (optionally gzipped) for processes
- Accurate perprocess CPU% on Linux via /proc jiffies delta
- Optional GPU & temperature metrics (disable via env vars)
- Simple token auth (?token=...) support
Run (no TLS):
```
cargo install socktop_agent
socktop_agent --port 3000
```
Enable TLS:
```
SOCKTOP_ENABLE_SSL=1 socktop_agent --port 8443
# cert/key stored under $XDG_DATA_HOME/socktop_agent/tls
```
Environment toggles:
- SOCKTOP_AGENT_GPU=0 (disable GPU collection)
- SOCKTOP_AGENT_TEMP=0 (disable temperature)
- SOCKTOP_TOKEN=secret (require token param from client)
- SOCKTOP_AGENT_METRICS_TTL_MS=250 (cache fast metrics window)
- SOCKTOP_AGENT_PROCESSES_TTL_MS=1000
- SOCKTOP_AGENT_DISKS_TTL_MS=1000
Systemd unit example & full docs:
https://github.com/jasonwitty/socktop
## WebSocket API Integration Guide
The socktop_agent exposes a WebSocket API that can be directly integrated with your own applications. This allows you to build custom monitoring dashboards or analysis tools using the agent's metrics.
### WebSocket Endpoint
```
ws://HOST:PORT/ws # Without TLS
wss://HOST:PORT/ws # With TLS
```
With authentication token (if configured):
```
ws://HOST:PORT/ws?token=YOUR_TOKEN
wss://HOST:PORT/ws?token=YOUR_TOKEN
```
### Communication Protocol
All communication uses JSON format for requests and responses, except for the process list which uses Protocol Buffers (protobuf) format with optional gzip compression.
#### Request Types
Send a JSON message with a `type` field to request specific metrics:
```json
{"type": "metrics"} // Request fast-changing metrics (CPU, memory, network)
{"type": "disks"} // Request disk information
{"type": "processes"} // Request process list (returns protobuf)
```
#### Response Formats
1. **Fast Metrics** (JSON):
```json
{
"cpu_total": 12.4,
"cpu_per_core": [11.2, 15.7],
"mem_total": 33554432,
"mem_used": 18321408,
"swap_total": 0,
"swap_used": 0,
"hostname": "myserver",
"cpu_temp_c": 42.5,
"networks": [{"name":"eth0","received":12345678,"transmitted":87654321}],
"gpus": [{"name":"nvidia-0","usage":56.7,"memory_total":8589934592,"memory_used":1073741824,"temp_c":65.0}]
}
```
2. **Disks** (JSON):
```json
[
{"name":"nvme0n1p2","total":512000000000,"available":320000000000},
{"name":"sda1","total":1000000000000,"available":750000000000}
]
```
3. **Processes** (Protocol Buffers):
Processes are returned in Protocol Buffers format, optionally gzip-compressed for large process lists. The protobuf schema is:
```protobuf
syntax = "proto3";
message Process {
uint32 pid = 1;
string name = 2;
float cpu_usage = 3;
uint64 mem_bytes = 4;
}
message ProcessList {
uint32 process_count = 1;
repeated Process processes = 2;
}
```
### Example Integration (JavaScript/Node.js)
```javascript
const WebSocket = require('ws');
// Connect to the agent
const ws = new WebSocket('ws://localhost:3000/ws');
ws.on('open', function open() {
console.log('Connected to socktop_agent');
// Request metrics immediately on connection
ws.send(JSON.stringify({type: 'metrics'}));
// Set up regular polling
setInterval(() => {
ws.send(JSON.stringify({type: 'metrics'}));
}, 1000);
// Request processes every 3 seconds
setInterval(() => {
ws.send(JSON.stringify({type: 'processes'}));
}, 3000);
});
ws.on('message', function incoming(data) {
// Check if the response is JSON or binary (protobuf)
try {
const jsonData = JSON.parse(data);
console.log('Received JSON data:', jsonData);
} catch (e) {
console.log('Received binary data (protobuf), length:', data.length);
// Process binary protobuf data with a library like protobufjs
}
});
ws.on('close', function close() {
console.log('Disconnected from socktop_agent');
});
```
### Example Integration (Python)
```python
import json
import asyncio
import websockets
async def monitor_system():
uri = "ws://localhost:3000/ws"
async with websockets.connect(uri) as websocket:
print("Connected to socktop_agent")
# Request initial metrics
await websocket.send(json.dumps({"type": "metrics"}))
# Set up regular polling
while True:
# Request metrics
await websocket.send(json.dumps({"type": "metrics"}))
# Receive and process response
response = await websocket.recv()
# Check if response is JSON or binary (protobuf)
try:
data = json.loads(response)
print(f"CPU: {data['cpu_total']}%, Memory: {data['mem_used']/data['mem_total']*100:.1f}%")
except json.JSONDecodeError:
print(f"Received binary data, length: {len(response)}")
# Process binary protobuf data with a library like protobuf
# Wait before next poll
await asyncio.sleep(1)
asyncio.run(monitor_system())
```
### Notes for Integration
1. **Error Handling**: The WebSocket connection may close unexpectedly; implement reconnection logic in your client.
2. **Rate Limiting**: Avoid excessive polling that could impact the system being monitored. Recommended intervals:
- Metrics: 500ms or slower
- Processes: 2000ms or slower
- Disks: 5000ms or slower
3. **Authentication**: If the agent is configured with a token, always include it in the WebSocket URL.
4. **Protocol Buffers Handling**: For processing the binary process list data, use a Protocol Buffers library for your language and the schema provided in the `proto/processes.proto` file.
5. **Compression**: Process lists may be gzip-compressed. Check if the response starts with the gzip magic bytes (`0x1f, 0x8b`) and decompress if necessary.
## LLM Integration Guide
If you're using an LLM to generate code for integrating with socktop_agent, this section provides structured information to help the model understand the API better.
### API Schema
```yaml
# WebSocket API Schema for socktop_agent
endpoint: ws://HOST:PORT/ws or wss://HOST:PORT/ws (with TLS)
authentication:
type: query parameter
parameter: token
example: ws://HOST:PORT/ws?token=YOUR_TOKEN
requests:
- type: metrics
format: JSON
example: {"type": "metrics"}
description: Fast-changing metrics (CPU, memory, network)
- type: disks
format: JSON
example: {"type": "disks"}
description: Disk information
- type: processes
format: JSON
example: {"type": "processes"}
description: Process list (returns protobuf)
responses:
- request_type: metrics
format: JSON
schema:
cpu_total: float # percentage of total CPU usage
cpu_per_core: [float] # array of per-core CPU usage percentages
mem_total: uint64 # total memory in bytes
mem_used: uint64 # used memory in bytes
swap_total: uint64 # total swap in bytes
swap_used: uint64 # used swap in bytes
hostname: string # system hostname
cpu_temp_c: float? # CPU temperature in Celsius (optional)
networks: [
{
name: string # network interface name
received: uint64 # total bytes received
transmitted: uint64 # total bytes transmitted
}
]
gpus: [
{
name: string # GPU device name
usage: float # GPU usage percentage
memory_total: uint64 # total GPU memory in bytes
memory_used: uint64 # used GPU memory in bytes
temp_c: float # GPU temperature in Celsius
}
]?
- request_type: disks
format: JSON
schema:
[
{
name: string # disk name
total: uint64 # total space in bytes
available: uint64 # available space in bytes
}
]
- request_type: processes
format: Protocol Buffers (optionally gzip-compressed)
schema: See protobuf definition below
```
### Protobuf Schema (processes.proto)
```protobuf
syntax = "proto3";
message Process {
uint32 pid = 1;
string name = 2;
float cpu_usage = 3;
uint64 mem_bytes = 4;
}
message ProcessList {
uint32 process_count = 1;
repeated Process processes = 2;
}
```
### Step-by-Step Integration Pseudocode
```
1. Establish WebSocket connection to ws://HOST:PORT/ws
- Add token if required: ws://HOST:PORT/ws?token=YOUR_TOKEN
2. For regular metrics updates:
- Send: {"type": "metrics"}
- Parse JSON response
- Extract CPU, memory, network info
3. For disk information:
- Send: {"type": "disks"}
- Parse JSON response
- Extract disk usage data
4. For process list:
- Send: {"type": "processes"}
- Check if response is binary
- If starts with 0x1f, 0x8b bytes:
- Decompress using gzip
- Parse binary data using protobuf schema
- Extract process information
5. Implement reconnection logic:
- On connection close/error
- Use exponential backoff
6. Respect rate limits:
- metrics: ≥ 500ms interval
- disks: ≥ 5000ms interval
- processes: ≥ 2000ms interval
```
### Common Implementation Patterns
**Pattern 1: Periodic Polling**
```javascript
// Set up separate timers for different metric types
const metricsInterval = setInterval(() => ws.send(JSON.stringify({type: 'metrics'})), 500);
const disksInterval = setInterval(() => ws.send(JSON.stringify({type: 'disks'})), 5000);
const processesInterval = setInterval(() => ws.send(JSON.stringify({type: 'processes'})), 2000);
// Clean up on disconnect
ws.on('close', () => {
clearInterval(metricsInterval);
clearInterval(disksInterval);
clearInterval(processesInterval);
});
```
**Pattern 2: Processing Binary Protobuf Data**
```javascript
// Using protobufjs
const root = protobuf.loadSync('processes.proto');
const ProcessList = root.lookupType('ProcessList');
ws.on('message', function(data) {
if (typeof data !== 'string') {
// Check for gzip compression
if (data[0] === 0x1f && data[1] === 0x8b) {
data = gunzipSync(data); // Use appropriate decompression library
}
// Decode protobuf
const processes = ProcessList.decode(new Uint8Array(data));
console.log(`Total processes: ${processes.process_count}`);
processes.processes.forEach(p => {
console.log(`PID: ${p.pid}, Name: ${p.name}, CPU: ${p.cpu_usage}%`);
});
}
});
```
**Pattern 3: Reconnection Logic**
```javascript
function connect() {
const ws = new WebSocket('ws://localhost:3000/ws');
ws.on('open', () => {
console.log('Connected');
// Start polling
});
ws.on('close', () => {
console.log('Connection lost, reconnecting...');
setTimeout(connect, 1000); // Reconnect after 1 second
});
// Handle other events...
}
connect();
```
+14
View File
@@ -0,0 +1,14 @@
fn main() {
// Vendored protoc for reproducible builds
let protoc = protoc_bin_vendored::protoc_bin_path().expect("protoc");
std::env::set_var("PROTOC", &protoc);
println!("cargo:rerun-if-changed=proto/processes.proto");
// Compile protobuf definitions for processes
let mut cfg = prost_build::Config::new();
cfg.out_dir(std::env::var("OUT_DIR").unwrap());
// Use local path (ensures file is inside published crate tarball)
cfg.compile_protos(&["proto/processes.proto"], &["proto"]) // relative to CARGO_MANIFEST_DIR
.expect("compile protos");
}
+15
View File
@@ -0,0 +1,15 @@
syntax = "proto3";
package socktop;
// All running processes. Sorting is done client-side.
message Processes {
uint64 process_count = 1; // total processes in the system
repeated Process rows = 2; // all processes
}
message Process {
uint32 pid = 1;
string name = 2;
float cpu_usage = 3; // 0..100
uint64 mem_bytes = 4; // RSS bytes
}
+69 -77
View File
@@ -1,99 +1,91 @@
//! socktop agent entrypoint: sets up sysinfo handles, launches a sampler,
//! and serves a WebSocket endpoint at /ws.
//! socktop agent entrypoint: sets up sysinfo handles and serves a WebSocket endpoint at /ws.
mod gpu;
mod metrics;
mod sampler;
mod proto;
// sampler module removed (metrics now purely request-driven)
mod state;
mod types;
mod ws;
use axum::{routing::get, Router};
use axum::{http::StatusCode, routing::get, Router};
use std::net::SocketAddr;
use std::str::FromStr;
mod tls;
use crate::sampler::{spawn_disks_sampler, spawn_process_sampler, spawn_sampler};
use state::AppState;
use ws::ws_handler;
fn arg_flag(name: &str) -> bool {
std::env::args().any(|a| a == name)
}
fn arg_value(name: &str) -> Option<String> {
let mut it = std::env::args();
while let Some(a) = it.next() {
if a == name {
return it.next();
}
}
None
}
#[tokio::main]
async fn main() {
async fn main() -> anyhow::Result<()> {
tracing_subscriber::fmt::init();
// Version flag (print and exit). Keep before heavy initialization.
if arg_flag("--version") || arg_flag("-V") {
println!("socktop_agent {}", env!("CARGO_PKG_VERSION"));
return Ok(());
}
let state = AppState::new();
// Start background sampler (adjust cadence as needed)
// 500ms fast metrics
let _h_fast = spawn_sampler(state.clone(), std::time::Duration::from_millis(500));
// 2s processes (top 50)
let _h_procs = spawn_process_sampler(state.clone(), std::time::Duration::from_secs(2), 50);
// 5s disks
let _h_disks = spawn_disks_sampler(state.clone(), std::time::Duration::from_secs(5));
// No background samplers: metrics collected on-demand per websocket request.
// Web app
let port = resolve_port();
// Web app: route /ws to the websocket handler
async fn healthz() -> StatusCode {
println!("/healthz request");
StatusCode::OK
}
let app = Router::new()
.route("/ws", get(ws_handler))
.with_state(state);
.route("/ws", get(ws::ws_handler))
.route("/healthz", get(healthz))
.with_state(state.clone());
let enable_ssl =
arg_flag("--enableSSL") || std::env::var("SOCKTOP_ENABLE_SSL").ok().as_deref() == Some("1");
if enable_ssl {
// Port can be overridden by --port or SOCKTOP_PORT; default to 8443 when SSL
let port = arg_value("--port")
.or_else(|| arg_value("-p"))
.or_else(|| std::env::var("SOCKTOP_PORT").ok())
.and_then(|s| s.parse::<u16>().ok())
.unwrap_or(8443);
let (cert_path, key_path) = tls::ensure_self_signed_cert()?;
let cfg = axum_server::tls_rustls::RustlsConfig::from_pem_file(cert_path, key_path).await?;
let addr = SocketAddr::from_str(&format!("0.0.0.0:{port}"))?;
println!("socktop_agent: TLS enabled. Listening on wss://{addr}/ws");
axum_server::bind_rustls(addr, cfg)
.serve(app.into_make_service())
.await?;
return Ok(());
}
// Non-TLS HTTP/WS path
let port = arg_value("--port")
.or_else(|| arg_value("-p"))
.or_else(|| std::env::var("SOCKTOP_PORT").ok())
.and_then(|s| s.parse::<u16>().ok())
.unwrap_or(3000);
let addr = SocketAddr::from(([0, 0, 0, 0], port));
//output to console
println!("Remote agent running at http://{addr}");
println!("WebSocket endpoint: ws://{addr}/ws");
//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();
println!("socktop_agent: Listening on ws://{addr}/ws");
axum_server::bind(addr)
.serve(app.into_make_service())
.await?;
Ok(())
}
// 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;
// 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='{s}'; using default {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 '{v}'; using default {DEFAULT}");
return DEFAULT;
}
}
} else {
eprintln!("Missing value for {arg} ; using default {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
}
// Unit tests for CLI parsing moved to `tests/port_parse.rs`.
+280 -56
View File
@@ -4,14 +4,19 @@ use crate::gpu::collect_all_gpus;
use crate::state::AppState;
use crate::types::{DiskInfo, Metrics, NetworkInfo, ProcessInfo, ProcessesPayload};
use once_cell::sync::OnceCell;
#[cfg(target_os = "linux")]
use std::collections::HashMap;
#[cfg(target_os = "linux")]
use std::fs;
#[cfg(target_os = "linux")]
use std::io;
use std::sync::Mutex;
use std::time::Duration as StdDuration;
use std::time::{Duration, Instant};
use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System};
use sysinfo::{ProcessRefreshKind, ProcessesToUpdate};
use tracing::warn;
// NOTE: CPU normalization env removed; non-Linux now always reports per-process share (0..100) as given by sysinfo.
// Runtime toggles (read once)
fn gpu_enabled() -> bool {
static ON: OnceCell<bool> = OnceCell::new();
@@ -44,6 +49,15 @@ struct GpuCache {
}
static GPUC: OnceCell<Mutex<GpuCache>> = OnceCell::new();
// Static caches for unchanging data
static HOSTNAME: OnceCell<String> = OnceCell::new();
struct NetworkNameCache {
names: Vec<String>,
infos: Vec<NetworkInfo>,
}
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;
@@ -94,6 +108,20 @@ fn set_gpus(v: Option<Vec<crate::gpu::GpuMetrics>>) {
// Collect only fast-changing metrics (CPU/mem/net + optional temps/gpus).
pub async fn collect_fast_metrics(state: &AppState) -> Metrics {
// TTL (ms) overridable via env, default 250ms
let ttl_ms: u64 = std::env::var("SOCKTOP_AGENT_METRICS_TTL_MS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(250);
let ttl = StdDuration::from_millis(ttl_ms);
{
let cache = state.cache_metrics.lock().await;
if cache.is_fresh(ttl) {
if let Some(c) = cache.get() {
return c.clone();
}
}
}
let mut sys = state.sys.lock().await;
if let Err(e) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
sys.refresh_cpu_usage();
@@ -102,9 +130,19 @@ pub async fn collect_fast_metrics(state: &AppState) -> Metrics {
warn!("sysinfo selective refresh panicked: {e:?}");
}
let hostname = System::host_name().unwrap_or_else(|| "unknown".to_string());
// Get or initialize hostname once
let hostname = HOSTNAME.get_or_init(|| state.hostname.clone()).clone();
// Reuse CPU vector to avoid allocation
let cpu_total = sys.global_cpu_usage();
let cpu_per_core: Vec<f32> = sys.cpus().iter().map(|c| c.cpu_usage()).collect();
let cpu_per_core = {
let vec_lock = CPU_VEC.get_or_init(|| Mutex::new(Vec::with_capacity(32)));
let mut vec = vec_lock.lock().unwrap();
vec.clear();
vec.extend(sys.cpus().iter().map(|c| c.cpu_usage()));
vec.clone() // Still need to clone but the allocation is reused
};
let mem_total = sys.total_memory();
let mem_used = mem_total.saturating_sub(sys.available_memory());
let swap_total = sys.total_swap();
@@ -137,38 +175,80 @@ pub async fn collect_fast_metrics(state: &AppState) -> Metrics {
None
};
// Networks
let networks: Vec<NetworkInfo> = {
// Networks with reusable name cache
let networks = {
let mut nets = state.networks.lock().await;
nets.refresh(false);
nets.iter()
.map(|(name, data)| NetworkInfo {
name: name.to_string(),
// Get or initialize network cache
let cache = NETWORK_CACHE.get_or_init(|| {
Mutex::new(NetworkNameCache {
names: Vec::new(),
infos: Vec::with_capacity(4), // Most systems have few network interfaces
})
});
let mut cache = cache.lock().unwrap();
// Collect current network names
let current_names: Vec<_> = nets.keys().map(|name| name.to_string()).collect();
// Update cached network names if they changed
if cache.names != current_names {
cache.names = current_names;
}
// Reuse NetworkInfo objects
cache.infos.clear();
for (name, data) in nets.iter() {
cache.infos.push(NetworkInfo {
name: name.to_string(), // We'll still clone but avoid Vec reallocation
received: data.total_received(),
transmitted: data.total_transmitted(),
})
.collect()
});
}
cache.infos.clone()
};
// GPUs: refresh only when cache is stale
let gpus = if cached_gpus().is_some() {
cached_gpus()
} else if gpu_enabled() {
let v = match collect_all_gpus() {
Ok(v) if !v.is_empty() => Some(v),
Ok(_) => None,
Err(e) => {
warn!("gpu collection failed: {e}");
None
// 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) => {
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
set_gpus(v.clone());
v
}
} else {
None
};
Metrics {
let metrics = Metrics {
cpu_total,
cpu_per_core,
mem_total,
@@ -181,23 +261,48 @@ pub async fn collect_fast_metrics(state: &AppState) -> Metrics {
networks,
top_processes: Vec::new(),
gpus,
};
{
let mut cache = state.cache_metrics.lock().await;
cache.set(metrics.clone());
}
metrics
}
// Cached disks
pub async fn collect_disks(state: &AppState) -> Vec<DiskInfo> {
let ttl_ms: u64 = std::env::var("SOCKTOP_AGENT_DISKS_TTL_MS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(1_000);
let ttl = StdDuration::from_millis(ttl_ms);
{
let cache = state.cache_disks.lock().await;
if cache.is_fresh(ttl) {
if let Some(v) = cache.get() {
return v.clone();
}
}
}
let mut disks_list = state.disks.lock().await;
disks_list.refresh(false); // don't drop missing disks
disks_list
let disks: Vec<DiskInfo> = disks_list
.iter()
.map(|d| DiskInfo {
name: d.name().to_string_lossy().into_owned(),
total: d.total_space(),
available: d.available_space(),
})
.collect()
.collect();
{
let mut cache = state.cache_disks.lock().await;
cache.set(disks.clone());
}
disks
}
// Linux-only helpers and implementation using /proc deltas for accurate CPU%.
#[cfg(target_os = "linux")]
#[inline]
fn read_total_jiffies() -> io::Result<u64> {
// /proc/stat first line: "cpu user nice system idle iowait irq softirq steal ..."
@@ -216,6 +321,7 @@ fn read_total_jiffies() -> io::Result<u64> {
Err(io::Error::other("no cpu line"))
}
#[cfg(target_os = "linux")]
#[inline]
fn read_proc_jiffies(pid: u32) -> Option<u64> {
let path = format!("/proc/{pid}/stat");
@@ -230,12 +336,26 @@ fn read_proc_jiffies(pid: u32) -> Option<u64> {
Some(utime.saturating_add(stime))
}
// Replace the body of collect_processes_top_k to use /proc deltas.
// This makes CPU% = (delta_proc / delta_total) * 100 over the 2s interval.
pub async fn collect_processes_top_k(state: &AppState, k: usize) -> ProcessesPayload {
// Fresh view to avoid lingering entries and select "no tasks" (no per-thread rows).
// Only processes, no per-thread entries.
let mut sys = System::new();
/// Collect all processes (Linux): compute CPU% via /proc jiffies delta; sorting moved to client.
#[cfg(target_os = "linux")]
pub async fn collect_processes_all(state: &AppState) -> ProcessesPayload {
let ttl_ms: u64 = std::env::var("SOCKTOP_AGENT_PROCESSES_TTL_MS")
.ok()
.and_then(|v| v.parse().ok())
// Higher default (1500ms) on non-Linux only; keep 1500 here for Linux correctness (more frequent updates).
.unwrap_or(1_500);
let ttl = StdDuration::from_millis(ttl_ms);
{
let cache = state.cache_processes.lock().await;
if cache.is_fresh(ttl) {
if let Some(c) = cache.get() {
return c.clone();
}
}
}
// Reuse shared System to avoid reallocation; refresh processes fully.
let mut sys_guard = state.sys.lock().await;
let sys = &mut *sys_guard;
sys.refresh_processes_specifics(
ProcessesToUpdate::All,
false,
@@ -256,12 +376,20 @@ pub async fn collect_processes_top_k(state: &AppState, k: usize) -> ProcessesPay
// Compute deltas vs last sample
let (last_total, mut last_map) = {
let mut t = state.proc_cpu.lock().await;
let lt = t.last_total;
let lm = std::mem::take(&mut t.last_per_pid);
t.last_total = total_now;
t.last_per_pid = current.clone();
(lt, lm)
#[cfg(target_os = "linux")]
{
let mut t = state.proc_cpu.lock().await;
let lt = t.last_total;
let lm = std::mem::take(&mut t.last_per_pid);
t.last_total = total_now;
t.last_per_pid = current.clone();
(lt, lm)
}
#[cfg(not(target_os = "linux"))]
{
let _: u64 = total_now; // silence unused warning
(0u64, HashMap::new())
}
};
// On first run or if total delta is tiny, report zeros
@@ -278,7 +406,7 @@ pub async fn collect_processes_top_k(state: &AppState, k: usize) -> ProcessesPay
.collect();
return ProcessesPayload {
process_count: total_count,
top_processes: top_k_sorted(procs, k),
top_processes: procs,
};
}
@@ -302,26 +430,122 @@ pub async fn collect_processes_top_k(state: &AppState, k: usize) -> ProcessesPay
})
.collect();
ProcessesPayload {
let payload = ProcessesPayload {
process_count: total_count,
top_processes: top_k_sorted(procs, k),
top_processes: procs,
};
{
let mut cache = state.cache_processes.lock().await;
cache.set(payload.clone());
}
payload
}
// Small helper to select and sort top-k by cpu
fn top_k_sorted(mut v: Vec<ProcessInfo>, k: usize) -> Vec<ProcessInfo> {
if v.len() > k {
v.select_nth_unstable_by(k, |a, b| {
b.cpu_usage
.partial_cmp(&a.cpu_usage)
.unwrap_or(std::cmp::Ordering::Equal)
});
v.truncate(k);
/// Collect all processes (non-Linux): optimized for reduced allocations and selective updates.
#[cfg(not(target_os = "linux"))]
pub async fn collect_processes_all(state: &AppState) -> ProcessesPayload {
// Serve from cache if fresh
{
let cache = state.cache_processes.lock().await;
if cache.is_fresh(StdDuration::from_millis(2_000)) {
// Use fixed TTL for cache check
if let Some(c) = cache.get() {
return c.clone();
}
}
}
v.sort_by(|a, b| {
b.cpu_usage
.partial_cmp(&a.cpu_usage)
.unwrap_or(std::cmp::Ordering::Equal)
});
v
// Single efficient refresh with optimized CPU collection
let (total_count, procs) = {
let mut sys = state.sys.lock().await;
let kind = ProcessRefreshKind::nothing().with_memory();
// Optimize refresh strategy based on system load
//if load > 5.0 {
//JW too complicated. simplify to remove strange behavior
// For active systems, get accurate CPU metrics
sys.refresh_processes_specifics(ProcessesToUpdate::All, false, kind.with_cpu());
// } else {
// // For idle systems, just get basic process info
// sys.refresh_processes_specifics(ProcessesToUpdate::All, false, kind);
// sys.refresh_cpu_usage();
// }
let total_count = sys.processes().len();
let cpu_count = sys.cpus().len() as f32;
// Reuse allocations via process cache
let mut proc_cache = state.proc_cache.lock().await;
proc_cache.reusable_vec.clear();
// Collect all processes, will sort by CPU later
for p in sys.processes().values() {
let pid = p.pid().as_u32();
// Reuse cached name if available
let name = if let Some(cached) = proc_cache.names.get(&pid) {
cached.clone()
} else {
let new_name = p.name().to_string_lossy().into_owned();
proc_cache.names.insert(pid, new_name.clone());
new_name
};
// 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;
proc_cache.reusable_vec.push(ProcessInfo {
pid,
name,
cpu_usage: total_cpu,
mem_bytes: p.memory(),
});
}
//JW no need to sort here; client does the sorting
// // Sort by CPU usage
// proc_cache.reusable_vec.sort_by(|a, b| {
// b.cpu_usage
// .partial_cmp(&a.cpu_usage)
// .unwrap_or(std::cmp::Ordering::Equal)
// });
// Clean up old process names cache when it grows too large
let cache_cleanup_threshold = std::env::var("SOCKTOP_AGENT_NAME_CACHE_CLEANUP_THRESHOLD")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(1000); // Default: most modern systems have 400-700 processes
if total_count > proc_cache.names.len() + cache_cleanup_threshold {
let now = std::time::Instant::now();
proc_cache
.names
.retain(|pid, _| sys.processes().contains_key(&sysinfo::Pid::from_u32(*pid)));
tracing::debug!(
"Cleaned up {} stale process names in {}ms",
proc_cache.names.capacity() - proc_cache.names.len(),
now.elapsed().as_millis()
);
}
// Get all processes, take ownership of the vec (will be replaced with empty vec)
(total_count, std::mem::take(&mut proc_cache.reusable_vec))
};
let payload = ProcessesPayload {
process_count: total_count,
top_processes: procs,
};
{
let mut cache = state.cache_processes.lock().await;
cache.set(payload.clone());
}
payload
}
+4 -31
View File
@@ -1,32 +1,5 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Metrics {
pub ts_unix_ms: i64,
pub host: String,
pub uptime_secs: u64,
pub cpu_overall: f32,
pub cpu_per_core: Vec<f32>,
pub load_avg: (f64, f64, f64),
pub mem_total_mb: u64,
pub mem_used_mb: u64,
pub swap_total_mb: u64,
pub swap_used_mb: u64,
pub net_aggregate: NetTotals,
pub top_processes: Vec<Proc>,
// Generated protobuf modules live under OUT_DIR; include them here.
// This module will expose socktop::Processes and socktop::Process types.
pub mod pb {
include!(concat!(env!("OUT_DIR"), "/socktop.rs"));
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NetTotals {
pub rx_bytes: u64,
pub tx_bytes: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Proc {
pub pid: i32,
pub name: String,
pub cpu: f32,
pub mem_mb: u64,
pub status: String,
}
-34
View File
@@ -1,34 +0,0 @@
//! Background sampler: periodically collects metrics and updates precompressed caches,
//! so WS replies just read and send cached bytes.
use crate::state::AppState;
use tokio::task::JoinHandle;
use tokio::time::{sleep, Duration};
// 500ms: fast path (cpu/mem/net/temp/gpu)
pub fn spawn_sampler(_state: AppState, _period: Duration) -> JoinHandle<()> {
tokio::spawn(async move {
// no-op background sampler (request-driven collection elsewhere)
loop {
sleep(Duration::from_secs(3600)).await;
}
})
}
// 2s: processes top-k
pub fn spawn_process_sampler(_state: AppState, _period: Duration, _top_k: usize) -> JoinHandle<()> {
tokio::spawn(async move {
loop {
sleep(Duration::from_secs(3600)).await;
}
})
}
// 5s: disks
pub fn spawn_disks_sampler(_state: AppState, _period: Duration) -> JoinHandle<()> {
tokio::spawn(async move {
loop {
sleep(Duration::from_secs(3600)).await;
}
})
}
+68 -2
View File
@@ -1,8 +1,9 @@
//! Shared agent state: sysinfo handles and hot JSON cache.
use std::collections::HashMap;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::{AtomicBool, AtomicUsize};
use std::sync::Arc;
use std::time::{Duration, Instant};
use sysinfo::{Components, Disks, Networks, System};
use tokio::sync::Mutex;
@@ -11,26 +12,82 @@ pub type SharedComponents = Arc<Mutex<Components>>;
pub type SharedDisks = Arc<Mutex<Disks>>;
pub type SharedNetworks = Arc<Mutex<Networks>>;
#[cfg(target_os = "linux")]
#[derive(Default)]
pub struct ProcCpuTracker {
pub last_total: u64,
pub last_per_pid: HashMap<u32, u64>,
}
#[cfg(not(target_os = "linux"))]
pub struct ProcessCache {
pub names: HashMap<u32, String>,
pub reusable_vec: Vec<crate::types::ProcessInfo>,
}
#[cfg(not(target_os = "linux"))]
impl Default for ProcessCache {
fn default() -> Self {
Self {
names: HashMap::with_capacity(1000), // Pre-allocate for typical modern system process count
reusable_vec: Vec::with_capacity(1000),
}
}
}
#[derive(Clone)]
pub struct AppState {
pub sys: SharedSystem,
pub components: SharedComponents,
pub disks: SharedDisks,
pub networks: SharedNetworks,
pub hostname: String,
// For correct per-process CPU% using /proc deltas
// For correct per-process CPU% using /proc deltas (Linux only path uses this tracker)
#[cfg(target_os = "linux")]
pub proc_cpu: Arc<Mutex<ProcCpuTracker>>,
// Process name caching and vector reuse for non-Linux to reduce allocations
#[cfg(not(target_os = "linux"))]
pub proc_cache: Arc<Mutex<ProcessCache>>,
// Connection tracking (to allow future idle sleeps if desired)
pub client_count: Arc<AtomicUsize>,
pub auth_token: Option<String>,
// GPU negative cache (probe once). gpu_checked=true after first attempt; gpu_present reflects result.
pub gpu_checked: Arc<AtomicBool>,
pub gpu_present: Arc<AtomicBool>,
// Lightweight on-demand caches (TTL based) to cap CPU under bursty polling.
pub cache_metrics: Arc<Mutex<CacheEntry<crate::types::Metrics>>>,
pub cache_disks: Arc<Mutex<CacheEntry<Vec<crate::types::DiskInfo>>>>,
pub cache_processes: Arc<Mutex<CacheEntry<crate::types::ProcessesPayload>>>,
}
#[derive(Clone, Debug)]
pub struct CacheEntry<T> {
pub at: Option<Instant>,
pub value: Option<T>,
}
impl<T> CacheEntry<T> {
pub fn new() -> Self {
Self {
at: None,
value: None,
}
}
pub fn is_fresh(&self, ttl: Duration) -> bool {
self.at.is_some_and(|t| t.elapsed() < ttl) && self.value.is_some()
}
pub fn set(&mut self, v: T) {
self.value = Some(v);
self.at = Some(Instant::now());
}
pub fn get(&self) -> Option<&T> {
self.value.as_ref()
}
}
impl AppState {
@@ -45,11 +102,20 @@ impl AppState {
components: Arc::new(Mutex::new(components)),
disks: Arc::new(Mutex::new(disks)),
networks: Arc::new(Mutex::new(networks)),
hostname: System::host_name().unwrap_or_else(|| "unknown".into()),
#[cfg(target_os = "linux")]
proc_cpu: Arc::new(Mutex::new(ProcCpuTracker::default())),
#[cfg(not(target_os = "linux"))]
proc_cache: Arc::new(Mutex::new(ProcessCache::default())),
client_count: Arc::new(AtomicUsize::new(0)),
auth_token: std::env::var("SOCKTOP_TOKEN")
.ok()
.filter(|s| !s.is_empty()),
gpu_checked: Arc::new(AtomicBool::new(false)),
gpu_present: Arc::new(AtomicBool::new(false)),
cache_metrics: Arc::new(Mutex::new(CacheEntry::new())),
cache_disks: Arc::new(Mutex::new(CacheEntry::new())),
cache_processes: Arc::new(Mutex::new(CacheEntry::new())),
}
}
}
+91
View File
@@ -0,0 +1,91 @@
use rcgen::{CertificateParams, DistinguishedName, DnType, IsCa, SanType};
use std::{
fs,
io::Write,
net::{IpAddr, Ipv4Addr},
path::{Path, PathBuf},
};
use time::{Duration, OffsetDateTime};
fn config_dir() -> PathBuf {
std::env::var_os("XDG_CONFIG_HOME")
.map(PathBuf::from)
.or_else(|| std::env::var_os("HOME").map(|h| Path::new(&h).join(".config")))
.unwrap_or_else(|| PathBuf::from("."))
.join("socktop_agent")
.join("tls")
}
pub fn cert_paths() -> (PathBuf, PathBuf) {
let dir = config_dir();
(dir.join("cert.pem"), dir.join("key.pem"))
}
pub fn ensure_self_signed_cert() -> anyhow::Result<(PathBuf, PathBuf)> {
let (cert_path, key_path) = cert_paths();
if cert_path.exists() && key_path.exists() {
return Ok((cert_path, key_path));
}
fs::create_dir_all(cert_path.parent().unwrap())?;
let hostname = hostname::get()
.ok()
.and_then(|s| s.into_string().ok())
.unwrap_or_else(|| "localhost".to_string());
let mut params = CertificateParams::new(vec![hostname.clone(), "localhost".into()])?;
params
.subject_alt_names
.push(SanType::IpAddress(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))));
params.subject_alt_names.push(SanType::IpAddress(IpAddr::V6(
::std::net::Ipv6Addr::LOCALHOST,
)));
params
.subject_alt_names
.push(SanType::IpAddress(IpAddr::V4(Ipv4Addr::UNSPECIFIED)));
// Allow operator to provide extra SANs (comma-separated), e.g. IPs or DNS names
if let Ok(extra) = std::env::var("SOCKTOP_AGENT_EXTRA_SANS") {
for raw in extra.split(',') {
let s = raw.trim();
if s.is_empty() {
continue;
}
if let Ok(ip) = s.parse::<IpAddr>() {
params.subject_alt_names.push(SanType::IpAddress(ip));
} else {
match s.to_string().try_into() {
Ok(dns) => params.subject_alt_names.push(SanType::DnsName(dns)),
Err(_) => eprintln!("socktop_agent: ignoring invalid SAN entry: {s}"),
}
}
}
}
let mut dn = DistinguishedName::new();
dn.push(DnType::CommonName, hostname.clone());
params.distinguished_name = dn;
params.is_ca = IsCa::NoCa;
// Dynamic validity: start slightly in the past to avoid clock skew issues, end ~397 days later
let now = OffsetDateTime::now_utc();
params.not_before = now - Duration::minutes(5);
params.not_after = now + Duration::days(397);
// Generate key pair (default is ECDSA P256 SHA256)
let key_pair = rcgen::KeyPair::generate()?; // defaults to ECDSA P256 SHA256
let cert = params.self_signed(&key_pair)?;
let cert_pem = cert.pem();
let key_pem = key_pair.serialize_pem();
let mut f = fs::File::create(&cert_path)?;
f.write_all(cert_pem.as_bytes())?;
let mut k = fs::File::create(&key_path)?;
k.write_all(key_pem.as_bytes())?;
println!(
"socktop_agent: generated self-signed TLS certificate at {}",
cert_path.display()
);
println!("socktop_agent: private key at {}", key_path.display());
Ok((cert_path, key_path))
}
+148 -4
View File
@@ -7,12 +7,33 @@ use axum::{
};
use flate2::{write::GzEncoder, Compression};
use futures_util::StreamExt;
use once_cell::sync::OnceCell;
use std::collections::HashMap;
use std::io::Write;
use tokio::sync::Mutex;
use crate::metrics::{collect_disks, collect_fast_metrics, collect_processes_top_k};
use crate::metrics::{collect_disks, collect_fast_metrics, collect_processes_all};
use crate::proto::pb;
use crate::state::AppState;
// Compression threshold based on typical payload size
const COMPRESSION_THRESHOLD: usize = 768;
// Reusable buffer for compression to avoid allocations
struct CompressionCache {
processes_vec: Vec<pb::Process>,
}
impl CompressionCache {
fn new() -> Self {
Self {
processes_vec: Vec::with_capacity(512), // Typical process count
}
}
}
static COMPRESSION_CACHE: OnceCell<Mutex<CompressionCache>> = OnceCell::new();
pub async fn ws_handler(
ws: WebSocketUpgrade,
State(state): State<AppState>,
@@ -44,8 +65,51 @@ async fn handle_socket(mut socket: WebSocket, state: AppState) {
let _ = send_json(&mut socket, &d).await;
}
Message::Text(ref text) if text == "get_processes" => {
let p = collect_processes_top_k(&state, 50).await;
let _ = send_json(&mut socket, &p).await;
let payload = collect_processes_all(&state).await;
// Map to protobuf message
// Get cached buffers
let cache = COMPRESSION_CACHE.get_or_init(|| Mutex::new(CompressionCache::new()));
let mut cache = cache.lock().await;
// Reuse process vector to build the list
cache.processes_vec.clear();
cache
.processes_vec
.extend(payload.top_processes.into_iter().map(|p| pb::Process {
pid: p.pid,
name: p.name,
cpu_usage: p.cpu_usage,
mem_bytes: p.mem_bytes,
}));
let pb = pb::Processes {
process_count: payload.process_count as u64,
rows: std::mem::take(&mut cache.processes_vec),
};
let mut buf = Vec::with_capacity(8 * 1024);
if prost::Message::encode(&pb, &mut buf).is_err() {
let _ = socket.send(Message::Close(None)).await;
} else {
// compress if large
if buf.len() <= COMPRESSION_THRESHOLD {
let _ = socket.send(Message::Binary(buf)).await;
} else {
// Create a new encoder for each message to ensure proper gzip headers
let mut encoder =
GzEncoder::new(Vec::with_capacity(buf.len()), Compression::fast());
match encoder.write_all(&buf).and_then(|_| encoder.finish()) {
Ok(compressed) => {
let _ = socket.send(Message::Binary(compressed)).await;
}
Err(_) => {
let _ = socket.send(Message::Binary(buf)).await;
}
}
}
}
drop(cache); // Explicit drop to release mutex early
}
Message::Close(_) => break,
_ => {}
@@ -59,7 +123,7 @@ async fn handle_socket(mut socket: WebSocket, state: AppState) {
// Small, cheap gzip for larger payloads; send text for small.
async fn send_json<T: serde::Serialize>(ws: &mut WebSocket, value: &T) -> Result<(), axum::Error> {
let json = serde_json::to_string(value).expect("serialize");
if json.len() <= 768 {
if json.len() <= COMPRESSION_THRESHOLD {
return ws.send(Message::Text(json)).await;
}
let mut enc = GzEncoder::new(Vec::new(), Compression::fast());
@@ -67,3 +131,83 @@ async fn send_json<T: serde::Serialize>(ws: &mut WebSocket, value: &T) -> Result
let bin = enc.finish().unwrap_or_else(|_| json.into_bytes());
ws.send(Message::Binary(bin)).await
}
#[cfg(test)]
mod tests {
use super::*;
use prost::Message as ProstMessage;
use sysinfo::System;
#[tokio::test]
async fn test_process_list_not_empty() {
// Initialize system data first to ensure we have processes
let mut sys = System::new_all();
sys.refresh_all();
// Create state and put the refreshed system in it
let state = AppState::new();
{
let mut sys_lock = state.sys.lock().await;
*sys_lock = sys;
}
// Get processes directly using the collection function
let processes = collect_processes_all(&state).await;
// Convert to protobuf message format
let cache = COMPRESSION_CACHE.get_or_init(|| Mutex::new(CompressionCache::new()));
let mut cache = cache.lock().await;
// Reuse process vector to build the list
cache.processes_vec.clear();
cache
.processes_vec
.extend(processes.top_processes.into_iter().map(|p| pb::Process {
pid: p.pid,
name: p.name,
cpu_usage: p.cpu_usage,
mem_bytes: p.mem_bytes,
}));
// Create the protobuf message
let pb = pb::Processes {
process_count: processes.process_count as u64,
rows: cache.processes_vec.clone(),
};
// Test protobuf encoding/decoding
let mut buf = Vec::new();
prost::Message::encode(&pb, &mut buf).expect("Failed to encode protobuf");
let decoded = pb::Processes::decode(buf.as_slice()).expect("Failed to decode protobuf");
// Print debug info
println!("Process count: {}", pb.process_count);
println!("Process vector length: {}", pb.rows.len());
println!("Encoded size: {} bytes", buf.len());
println!("Decoded process count: {}", decoded.rows.len());
// Print first few processes if available
for (i, process) in pb.rows.iter().take(5).enumerate() {
println!(
"Process {}: {} (PID: {}) CPU: {:.1}% MEM: {} bytes",
i + 1,
process.name,
process.pid,
process.cpu_usage,
process.mem_bytes
);
}
// Validate
assert!(!pb.rows.is_empty(), "Process list should not be empty");
assert!(
pb.process_count > 0,
"Process count should be greater than 0"
);
assert_eq!(
pb.process_count as usize,
pb.rows.len(),
"Process count mismatch with actual rows"
);
}
}
+28
View File
@@ -0,0 +1,28 @@
//! CLI arg parsing tests for socktop_agent (server)
use std::process::Command;
#[test]
fn test_help_and_port_short_long() {
// We verify port flags are accepted by ensuring the process starts (then we kill quickly).
// Use an unlikely port to avoid conflicts.
let exe = env!("CARGO_BIN_EXE_socktop_agent");
// TLS enabled with long --port
let mut child = Command::new(exe)
.args(["--enableSSL", "--port", "9555"])
.spawn()
.expect("spawn agent");
// Give it a moment to bind
std::thread::sleep(std::time::Duration::from_millis(150));
let _ = child.kill();
let _ = child.wait();
// TLS enabled with short -p
let mut child2 = Command::new(exe)
.args(["--enableSSL", "-p", "9556"])
.spawn()
.expect("spawn agent");
std::thread::sleep(std::time::Duration::from_millis(150));
let _ = child2.kill();
let _ = child2.wait();
}
+40
View File
@@ -0,0 +1,40 @@
//! Unit test for port parsing logic moved out of `main.rs`.
fn parse_port<I: IntoIterator<Item = String>>(args: I, default_port: u16) -> u16 {
let mut it = args.into_iter();
let _ = it.next(); // program name
let mut long: Option<String> = None;
let mut short: Option<String> = None;
while let Some(a) = it.next() {
match a.as_str() {
"--port" => long = it.next(),
"-p" => short = it.next(),
_ if a.starts_with("--port=") => {
if let Some((_, v)) = a.split_once('=') {
long = Some(v.to_string());
}
}
_ => {}
}
}
long.or(short)
.and_then(|s| s.parse::<u16>().ok())
.unwrap_or(default_port)
}
#[test]
fn port_long_short_and_assign() {
assert_eq!(
parse_port(vec!["agent".into(), "--port".into(), "9001".into()], 8443),
9001
);
assert_eq!(
parse_port(vec!["agent".into(), "-p".into(), "9002".into()], 8443),
9002
);
assert_eq!(
parse_port(vec!["agent".into(), "--port=9003".into()], 8443),
9003
);
assert_eq!(parse_port(vec!["agent".into()], 8443), 8443);
}
+59
View File
@@ -0,0 +1,59 @@
use assert_cmd::prelude::*;
use std::fs;
use std::path::PathBuf;
use std::process::Command;
use std::time::Duration;
use std::time::Instant;
fn expected_paths(config_home: &std::path::Path) -> (PathBuf, PathBuf) {
let base = config_home.join("socktop_agent").join("tls");
(base.join("cert.pem"), base.join("key.pem"))
}
#[test]
fn generates_self_signed_cert_and_key_in_xdg_path() {
// Create an isolated fake XDG_CONFIG_HOME
let tmpdir = tempfile::tempdir().expect("tempdir");
let xdg = tmpdir.path().to_path_buf();
// Run the agent once with --enableSSL, short timeout so it exits quickly when killed
let mut cmd = Command::cargo_bin("socktop_agent").expect("binary exists");
// Bind to an ephemeral port (-p 0) to avoid conflicts/flakes
cmd.env("XDG_CONFIG_HOME", &xdg)
.arg("--enableSSL")
.arg("-p")
.arg("0");
// Spawn the process and poll for cert generation
let mut child = cmd.spawn().expect("spawn agent");
// Poll up to ~3s for files to appear to avoid timing flakes
let (cert_path, key_path) = expected_paths(&xdg);
let start = Instant::now();
let timeout = Duration::from_millis(3000);
let interval = Duration::from_millis(50);
while start.elapsed() < timeout {
if cert_path.exists() && key_path.exists() {
break;
}
std::thread::sleep(interval);
}
// Terminate the process regardless
let _ = child.kill();
let _ = child.wait();
// Verify files exist at expected paths
assert!(
cert_path.exists(),
"cert not found at {}",
cert_path.display()
);
assert!(key_path.exists(), "key not found at {}", key_path.display());
// Also ensure they are non-empty
let cert_md = fs::metadata(&cert_path).expect("cert metadata");
let key_md = fs::metadata(&key_path).expect("key metadata");
assert!(cert_md.len() > 0, "cert is empty");
assert!(key_md.len() > 0, "key is empty");
}