From 4a4acedbc3e78a82e19ce1c2fc81223091628c8d Mon Sep 17 00:00:00 2001 From: lda Date: Mon, 12 Oct 2026 16:28:21 +0700 Subject: [PATCH] feat(telemetry): report process memory and process and host uptime --- Cargo.lock | 24 +++++++++++ docs/roadmap.md | 3 +- docs/telemetry-recipes.md | 35 +++++++++++++++ docs/telemetry.md | 26 +++++++++++- ipjs/README.md | 3 ++ ipjs/src/subcommands/neighbor/nl.rs | 2 +- wakey-agent/Cargo.toml | 3 ++ wakey-agent/src/tracing.rs | 7 ++- wakey-control-plane/Cargo.toml | 3 ++ wakey-control-plane/src/tracing.rs | 7 ++- wakey-core/src/telemetry.rs | 17 ++++++++ wakey-core/src/telemetry/process.rs | 66 +++++++++++++++++++++++++++++ wakey-core/tests/telemetry.rs | 55 +++++++++++++++++++++++- wakey-linux/Cargo.toml | 6 +++ wakey-linux/src/lib.rs | 2 + wakey-linux/src/telemetry.rs | 60 ++++++++++++++++++++++++++ 16 files changed, 311 insertions(+), 8 deletions(-) create mode 100644 wakey-core/src/telemetry/process.rs create mode 100644 wakey-linux/src/telemetry.rs diff --git a/Cargo.lock b/Cargo.lock index edb2b13..17d5ee6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1977,6 +1977,27 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "procfs" +version = "0.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25485360a54d6861439d60facef26de713b1e126bf015ec8f98239467a2b82f7" +dependencies = [ + "bitflags", + "procfs-core", + "rustix", +] + +[[package]] +name = "procfs-core" +version = "0.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6401bf7b6af22f78b563665d15a22e9aef27775b79b149a66ca022468a4e405" +dependencies = [ + "bitflags", + "hex", +] + [[package]] name = "prost" version = "0.14.3" @@ -3671,6 +3692,7 @@ dependencies = [ "url", "wakey", "wakey-core", + "wakey-linux", "wakey-vt100", ] @@ -3696,6 +3718,7 @@ dependencies = [ "uuid", "wakey-agent", "wakey-core", + "wakey-linux", ] [[package]] @@ -3733,6 +3756,7 @@ dependencies = [ "lda-ipjs", "macaddr", "nix 0.31.2", + "procfs", "pty-process", "serde", "serde_json", diff --git a/docs/roadmap.md b/docs/roadmap.md index ac21364..8cdccb7 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -55,7 +55,8 @@ or current fleet size. Metric labels should remain bounded; no per-device labels ## Process And Host Visibility -- [ ] Add shared process measurements for agent and CC: start time/uptime, resident memory, CPU time, and open file descriptors where supported. Keep version as resource metadata rather than repeating it on every sample. +- [x] Add Linux process age/RSS and host uptime gauges for agent and CC through opt-in OTLP, using the existing collection interval. Core owns OS-independent instruments; wakey-linux supplies the optional procfs reader. Missing data omits observations; no added labels, config fields, health routes or polling tasks. Collector coverage verifies units and values; router/live dashboard validation pending. +- [ ] Extend process measurements with CPU time and open file descriptors where supported, plus start-time/version metadata if needed. Keep version as resource metadata rather than repeating it on every sample. - [ ] Add a small host summary: hostname, OS/kernel, CPU count, host uptime, load, and available/total memory. Distinguish host uptime from process uptime so router reboots and agent restarts are visible separately. Sources must tolerate missing Linux files; not every agent host is a router. - [ ] Expose changing measurements through OTLP on a modest sampling interval; show selected process/host details on Fleet Health and agent details. Keep host information behind the protected operator interface, not public health endpoints. Avoid expensive reads on every request and unbounded labels. - [ ] Keep `/healthz` as cheap liveness; consider a separate bounded `/readyz` database-access check for CC. Collector outages and disconnected fleet agents must not make a functioning CC unready. diff --git a/docs/telemetry-recipes.md b/docs/telemetry-recipes.md index 445397e..b2718c7 100644 --- a/docs/telemetry-recipes.md +++ b/docs/telemetry-recipes.md @@ -26,6 +26,41 @@ that includes recent activity and allow at least two exports for rate queries. Counter resets on restart are handled by `rate` and `increase`. Apply those functions **before** summing across instances so individual resets remain visible. +## Process And Host + +Discover the new gauges first (these are Linux-only): + +```promql +count by (__name__, job, instance) ({__name__=~"wakey_(process|host)_.*"}) +``` + +Process age and host uptime, in seconds: + +```promql +wakey_process_uptime_seconds +``` + +```promql +wakey_host_uptime_seconds +``` + +Set Grafana's unit to a duration. Compare the two: process age resetting while +host uptime continues suggests a daemon restart; both resetting suggests a host +restart or a changed namespace/view. Missing samples are not a zero uptime or +proof of a restart. These are gauges: do not use `rate` or `increase` on them. + +Resident process memory, in bytes: + +```promql +wakey_process_memory_resident_bytes +``` + +Set the unit to **bytes (IEC)**. Keep `job` and `instance` in the legend to compare +agent and CC separately. RSS includes shared/file-backed pages and is not heap +allocation alone. A missing series can mean disabled telemetry, an unsupported +platform, unavailable procfs, or export trouble. No new health endpoint or host +inventory is provided here; do not sum host uptime from multiple daemons. + ## Operations Completed attempts per second, including errors and timeouts: diff --git a/docs/telemetry.md b/docs/telemetry.md index a976548..47b5742 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -157,6 +157,9 @@ queries, rates, averages, percentiles, and Grafana units. | `wakey.stored.devices` | CC gauge of retained device rows, labeled by `agent_id` | | `wakey.stored.endpoints` | CC gauge of retained endpoint rows, including inactive entries, labeled by `agent_id` | | `wakey.fleet.devices` | CC gauge of unfiltered grouped identities, labeled by `identity` (`saved` or `unsaved`) | +| `wakey.process.uptime` | Linux gauge: seconds since this process's kernel-reported start, including suspend | +| `wakey.process.memory.resident` | Linux gauge: approximate resident memory in bytes, from VmRSS | +| `wakey.host.uptime` | Linux gauge: uptime in seconds reported by the process's procfs view, including suspend | | `wakey.terminal.active` | Agent PTY workers currently alive | | `wakey.terminal.input.bytes` | Input bytes received by the terminal relay | | `wakey.terminal.output.bytes` | Output bytes read from the PTY, including while detached | @@ -164,9 +167,30 @@ queries, rates, averages, percentiles, and Grafana units. Metric labels do not include request/session IDs, MACs, or IPs. New spans use `skip_all` and explicitly select safe fields; terminal bytes and credentials are -not attached. Profiles, host CPU/RSS metrics, connected-agent gauges, and export-loss +not attached. Profiles, host CPU/total-memory metrics, connected-agent gauges, and export-loss dashboards are not part of this initial slice. +Process and host gauges are registered automatically for Linux agent/CC daemons +when OTLP is enabled. `wakey-core` owns the OS-independent instruments and +`ProcessMetricsSource` interface; `wakey-linux` supplies the procfs reader via +its optional `telemetry` feature. Each daemon attaches that reader to its +telemetry guard, so core has no Linux dependency or filesystem reads. +Their small procfs reads run during the existing 30-second +metric collection and final shutdown collection; no independent poller or task +is started. Non-Linux builds omit these gauges. Missing/unreadable procfs data +omits the affected observation instead of reporting zero or failing the daemon. +No hostname, PID, command line or environment values are exported by this slice. +Service/instance resource attributes identify each daemon; two daemons on the +same host can report the same host uptime, so do not sum it across processes. + +RSS includes anonymous, file-backed and shared resident pages, not just heap, +and is an approximate kernel measurement. Process age uses procfs start ticks +and uptime, not wall-clock subtraction or telemetry initialization time. +Container/time namespaces can change the visible host clock; this is the +process's Linux view, not an assertion about a physical machine's boot time. +The Linux parser is [procfs](https://docs.rs/procfs/0.18.0/procfs/), with optional +default features disabled. Public `/healthz` remains unchanged. + Current-state fleet gauges sample independently every 30 seconds, with bounded revision retries to avoid mixed-state counts. Enrolled empty agents report zero; removed labels disappear when no enrollment or retained rows remain. Failed samples diff --git a/ipjs/README.md b/ipjs/README.md index 0b4ab9f..de693bd 100644 --- a/ipjs/README.md +++ b/ipjs/README.md @@ -53,6 +53,9 @@ Use netlink only when: Today that mainly applies to `neighbor::nl`. +`wakey-linux` explicitly uses netlink for neighbor and address inventory on its +Unix-gated path, without automatic JSON fallback. + ## Relationship to wakey `wakey` is the product. diff --git a/ipjs/src/subcommands/neighbor/nl.rs b/ipjs/src/subcommands/neighbor/nl.rs index 8940225..fff9aa6 100644 --- a/ipjs/src/subcommands/neighbor/nl.rs +++ b/ipjs/src/subcommands/neighbor/nl.rs @@ -1,4 +1,4 @@ -//! rtnetlink-based neighbor table query. One syscall, filter in userspace. +//! rtnetlink-based neighbor query: dump links and neighbors, then filter in userspace. #![cfg(unix)] use std::{ collections::{HashMap, HashSet}, diff --git a/wakey-agent/Cargo.toml b/wakey-agent/Cargo.toml index 91fc776..7e4001d 100644 --- a/wakey-agent/Cargo.toml +++ b/wakey-agent/Cargo.toml @@ -37,3 +37,6 @@ url = "2" vt100 = { package = "wakey-vt100", path = "../vendor/vt100", registry = "gitea", version = "=0.16.2-wakey.2" } wakey = { path = "..", registry = "gitea", version = "0" } wakey-core = { path = "../wakey-core", registry = "gitea", version = "0", features = ["telemetry"] } + +[target.'cfg(target_os = "linux")'.dependencies] +wakey-linux = { path = "../wakey-linux", registry = "gitea", version = "0", features = ["telemetry"] } diff --git a/wakey-agent/src/tracing.rs b/wakey-agent/src/tracing.rs index c87af6d..75897d4 100644 --- a/wakey-agent/src/tracing.rs +++ b/wakey-agent/src/tracing.rs @@ -6,7 +6,7 @@ pub fn init( config: &TelemetryConfig, agent_id: Option<&str>, ) -> Result { - wakey_core::telemetry::init( + let guard = wakey_core::telemetry::init( config, "wakey-agent", match verbose { @@ -15,5 +15,8 @@ pub fn init( _ => "wakey_agent=trace,wakey=debug,wakey_core=info,opentelemetry=warn", }, agent_id, - ) + )?; + #[cfg(target_os = "linux")] + let guard = guard.with_process_metrics(wakey_linux::telemetry::ProcfsProcessMetrics); + Ok(guard) } diff --git a/wakey-control-plane/Cargo.toml b/wakey-control-plane/Cargo.toml index 3ce90c3..01cd2f9 100644 --- a/wakey-control-plane/Cargo.toml +++ b/wakey-control-plane/Cargo.toml @@ -37,3 +37,6 @@ tracing = { workspace = true } uuid = { workspace = true } wakey-agent = { path = "../wakey-agent", registry = "gitea", version = "0" } wakey-core = { path = "../wakey-core", registry = "gitea", version = "0", features = ["telemetry"] } + +[target.'cfg(target_os = "linux")'.dependencies] +wakey-linux = { path = "../wakey-linux", registry = "gitea", version = "0", features = ["telemetry"] } diff --git a/wakey-control-plane/src/tracing.rs b/wakey-control-plane/src/tracing.rs index 90d58e5..59825e1 100644 --- a/wakey-control-plane/src/tracing.rs +++ b/wakey-control-plane/src/tracing.rs @@ -3,7 +3,7 @@ use anyhow::Result; pub use wakey_core::telemetry::TelemetryGuard; pub fn init(verbose: u8, telemetry: &TelemetryConfig) -> Result { - wakey_core::telemetry::init( + let guard = wakey_core::telemetry::init( telemetry, "wakey-control-plane", match verbose { @@ -12,5 +12,8 @@ pub fn init(verbose: u8, telemetry: &TelemetryConfig) -> Result _ => "wakey_control_plane=trace,wakey_core=info,opentelemetry=warn", }, None, - ) + )?; + #[cfg(target_os = "linux")] + let guard = guard.with_process_metrics(wakey_linux::telemetry::ProcfsProcessMetrics); + Ok(guard) } diff --git a/wakey-core/src/telemetry.rs b/wakey-core/src/telemetry.rs index 7903e3a..2546157 100644 --- a/wakey-core/src/telemetry.rs +++ b/wakey-core/src/telemetry.rs @@ -43,9 +43,23 @@ pub struct TelemetryGuard { traces: Option, logs: Option, metrics: Option, + process: Option, } impl TelemetryGuard { + /// Attach platform measurements only when OTLP metrics are enabled. Core + /// owns instruments; the supplied reader owns all OS-specific collection. + pub fn with_process_metrics(mut self, source: impl ProcessMetricsSource) -> Self { + if let Some(metrics) = &self.metrics { + use opentelemetry::metrics::MeterProvider as _; + self.process = Some(process::ProcessMetrics::new( + &metrics.meter("wakey"), + source, + )); + } + self + } + pub fn shutdown(self) { // Export failures must not turn an otherwise successful daemon exit into a failure. if let Some(provider) = self.metrics { @@ -198,6 +212,7 @@ pub fn init( traces: Some(traces), logs: Some(logs), metrics: Some(metrics), + process: None, }; } else if config.json_logs { tracing_subscriber::registry() @@ -292,7 +307,9 @@ fn local_timer() -> fmt::time::OffsetTime = OnceLock::new(); mod fleet; +mod process; pub use fleet::{AgentDeviceCounts, FleetCounts, FleetMetrics}; +pub use process::ProcessMetricsSource; struct Instruments { operations: opentelemetry::metrics::Counter, duration: opentelemetry::metrics::Histogram, diff --git a/wakey-core/src/telemetry/process.rs b/wakey-core/src/telemetry/process.rs new file mode 100644 index 0000000..8d5af68 --- /dev/null +++ b/wakey-core/src/telemetry/process.rs @@ -0,0 +1,66 @@ +//! OS-independent process/host instruments. Callers supply a platform reader. +//! Failed reads omit a sample: zero must not masquerade as a restart or freed RAM. + +use opentelemetry::metrics::{Meter, ObservableGauge}; +use std::sync::Arc; + +/// Values are seconds/bytes, or None when unsupported/unavailable. Readers run +/// on the metric collection thread: keep work small and local, never perform +/// network requests or emit exporter-directed logs from these callbacks. +pub trait ProcessMetricsSource: Send + Sync + 'static { + fn process_uptime_seconds(&self) -> Option; + fn resident_memory_bytes(&self) -> Option; + fn host_uptime_seconds(&self) -> Option; +} + +pub(super) struct ProcessMetrics { + _uptime: ObservableGauge, + _resident: ObservableGauge, + _host_uptime: ObservableGauge, +} + +impl ProcessMetrics { + pub(super) fn new(meter: &Meter, source: impl ProcessMetricsSource) -> Self { + let source = Arc::new(source); + let uptime = source.clone(); + let resident = source.clone(); + Self { + _uptime: meter + .f64_observable_gauge("wakey.process.uptime") + .with_unit("s") + .with_description("Process age reported by the platform reader") + .with_callback(move |observer| { + if let Some(seconds) = uptime.process_uptime_seconds() + && seconds.is_finite() + && seconds >= 0.0 + { + observer.observe(seconds, &[]); + } + }) + .build(), + _resident: meter + .u64_observable_gauge("wakey.process.memory.resident") + .with_unit("By") + .with_description("Resident process memory reported by the platform reader") + .with_callback(move |observer| { + if let Some(bytes) = resident.resident_memory_bytes() { + observer.observe(bytes, &[]); + } + }) + .build(), + _host_uptime: meter + .f64_observable_gauge("wakey.host.uptime") + .with_unit("s") + .with_description("Host uptime reported by the platform reader") + .with_callback(move |observer| { + if let Some(seconds) = source.host_uptime_seconds() + && seconds.is_finite() + && seconds >= 0.0 + { + observer.observe(seconds, &[]); + } + }) + .build(), + } + } +} diff --git a/wakey-core/tests/telemetry.rs b/wakey-core/tests/telemetry.rs index cbe6a4f..9bcf931 100644 --- a/wakey-core/tests/telemetry.rs +++ b/wakey-core/tests/telemetry.rs @@ -197,6 +197,20 @@ async fn exports_all_signals_and_preserves_remote_command_parent() { }, ); let server = tokio::spawn(server); + // The exporter contract is OS-independent; Linux procfs collection is + // tested in wakey-linux rather than imported into core. + struct ProcessSample; + impl telemetry::ProcessMetricsSource for ProcessSample { + fn process_uptime_seconds(&self) -> Option { + Some(120.5) + } + fn resident_memory_bytes(&self) -> Option { + Some(32 * 1024 * 1024) + } + fn host_uptime_seconds(&self) -> Option { + Some(10_000.0) + } + } let guard = telemetry::init( &TelemetryConfig { otlp_endpoint: Some(endpoint), @@ -211,7 +225,8 @@ async fn exports_all_signals_and_preserves_remote_command_parent() { "info", Some("test-agent"), ) - .unwrap(); + .unwrap() + .with_process_metrics(ProcessSample); let fleet_metrics = telemetry::FleetMetrics::new().unwrap(); fleet_metrics.replace(Some(telemetry::FleetCounts { @@ -344,6 +359,44 @@ async fn exports_all_signals_and_preserves_remote_command_parent() { assert!(names.contains(&name), "missing {name}"); } let latest = metrics.last().unwrap(); + { + for (name, unit, expected) in [ + ( + "wakey.process.uptime", + "s", + number_data_point::Value::AsDouble(120.5), + ), + ( + "wakey.process.memory.resident", + "By", + number_data_point::Value::AsInt(32 * 1024 * 1024), + ), + ( + "wakey.host.uptime", + "s", + number_data_point::Value::AsDouble(10_000.0), + ), + ] { + let metric = latest + .resource_metrics + .iter() + .flat_map(|resource| &resource.scope_metrics) + .flat_map(|scope| &scope.metrics) + .find(|metric| metric.name == name) + .unwrap_or_else(|| panic!("missing {name}")); + assert_eq!(metric.unit, unit); + let Some(metric::Data::Gauge(gauge)) = &metric.data else { + panic!("{name} must be a gauge"); + }; + assert_eq!(gauge.data_points.len(), 1); + let point = &gauge.data_points[0]; + assert!( + point.attributes.is_empty(), + "process metrics need no PID/host/address labels" + ); + assert_eq!(point.value, Some(expected)); + } + } for (name, label, expected) in [ ( "wakey.stored.devices", diff --git a/wakey-linux/Cargo.toml b/wakey-linux/Cargo.toml index 2d9bf88..59f77f1 100644 --- a/wakey-linux/Cargo.toml +++ b/wakey-linux/Cargo.toml @@ -17,6 +17,12 @@ wakey-core = { path = "../wakey-core", registry = "gitea", version = "0"} [dev-dependencies] serial_test = "3" +[features] +telemetry = ["wakey-core/telemetry", "dep:procfs"] + +[target.'cfg(target_os = "linux")'.dependencies] +procfs = { version = "0.18", default-features = false, optional = true } + [dependencies.lda-ipjs] path = "../ipjs" registry = "gitea" diff --git a/wakey-linux/src/lib.rs b/wakey-linux/src/lib.rs index 94bd3a4..c9b8a17 100644 --- a/wakey-linux/src/lib.rs +++ b/wakey-linux/src/lib.rs @@ -3,6 +3,8 @@ pub mod devices; pub mod dhcp; pub mod observations; +#[cfg(all(target_os = "linux", feature = "telemetry"))] +pub mod telemetry; #[cfg(unix)] pub mod terminal; pub mod wake; diff --git a/wakey-linux/src/telemetry.rs b/wakey-linux/src/telemetry.rs new file mode 100644 index 0000000..3960d66 --- /dev/null +++ b/wakey-linux/src/telemetry.rs @@ -0,0 +1,60 @@ +//! Linux procfs reader for core's OS-independent process/host instruments. +use procfs::Current; +use wakey_core::telemetry::ProcessMetricsSource; + +/// Reads only this process and the visible host clock; never scans other PIDs. +/// Constructing the reader does no I/O, so disabled OTLP remains local-only. +pub struct ProcfsProcessMetrics; + +impl ProcessMetricsSource for ProcfsProcessMetrics { + fn process_uptime_seconds(&self) -> Option { + process_uptime() + } + fn resident_memory_bytes(&self) -> Option { + resident_bytes() + } + fn host_uptime_seconds(&self) -> Option { + host_uptime() + } +} + +fn host_uptime() -> Option { + let seconds = procfs::Uptime::current().ok()?.uptime; + (seconds.is_finite() && seconds >= 0.0).then_some(seconds) +} + +fn process_uptime() -> Option { + let stat = procfs::process::Process::myself().ok()?.stat().ok()?; + let ticks = procfs::ticks_per_second(); + if ticks == 0 { + return None; + } + // Both values use the kernel's boot-time clock, not wall time or telemetry + // initialization time. This also avoids parsing process names in stat ourselves. + let age = host_uptime()? - stat.starttime as f64 / ticks as f64; + (age.is_finite() && age >= 0.0).then_some(age) +} + +fn resident_bytes() -> Option { + let status = procfs::process::Process::myself().ok()?.status().ok()?; + // procfs documents VmRSS in KiB even though the Linux file spells it "kB". + // It includes anonymous, file-backed and shared resident pages, not heap alone. + status.vmrss?.checked_mul(1024) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn current_process_and_visible_host_can_be_sampled() { + let source = ProcfsProcessMetrics; + let age = source + .process_uptime_seconds() + .expect("process start/uptime available"); + let host = source.host_uptime_seconds().expect("host uptime available"); + assert!(age.is_finite() && age >= 0.0); + assert!(host.is_finite() && host >= age); + assert!(source.resident_memory_bytes().expect("VmRSS available") > 0); + } +}