optirs_nas/nas_engine/
telemetry.rs

1//! Injectable system telemetry (F12).
2//!
3//! `SystemResourceTracker` used to fabricate its readings: 32 GB total memory,
4//! 16 GB used, 60% CPU, 4 GPUs at 80%, 1 TB disk half full, 1 GB/s network, 65 C,
5//! 250 W — all hard-coded literals behind
6//! `// In a real implementation, this would query system APIs`. Those numbers were
7//! then compared against the user's
8//! [`crate::nas_engine::config::ResourceConstraints`], so a search
9//! configured with (say) an 8 GB memory budget aborted immediately on the strength
10//! of an invented 16 GB reading.
11//!
12//! The rule here is: **report only what can actually be measured, and `None`
13//! otherwise.** Consumers must treat `None` as "unconstrained", never as zero and
14//! never as a violation.
15//!
16//! What [`StdTelemetry`] can honestly measure, in pure Rust with no FFI:
17//!
18//! | field | source | portable? |
19//! |---|---|---|
20//! | `logical_cpus` | [`std::thread::available_parallelism`] | yes |
21//! | `process_memory_gb` | `/proc/self/status` `VmRSS` | Linux only |
22//! | `total_memory_gb`, `available_memory_gb` | `/proc/meminfo` | Linux only |
23//! | `load_average_per_cpu` | `/proc/loadavg` divided by the CPU count | Linux only |
24//! | `process_count` | entries in `/proc` | Linux only |
25//!
26//! Everything else — GPU count and utilization, disk capacity, network bandwidth,
27//! die temperature, package power — needs a platform API or a vendor library. None
28//! of those can be reached without FFI, so [`StdTelemetry`] reports `None` for
29//! them rather than inventing a plausible-looking constant. Supply a
30//! [`TelemetrySource`] of your own (e.g. backed by a monitoring agent you already
31//! run) to fill them in.
32//!
33//! The `/proc` parsers are pure functions over `&str`
34//! ([`parse_vm_rss_kib`], [`parse_meminfo`], [`parse_loadavg`]) so the Linux path
35//! is unit-tested from fixture strings on every platform, including the macOS and
36//! Windows hosts where the files do not exist.
37
38use std::fmt::Debug;
39use std::path::Path;
40
41/// One telemetry reading. Every field is `Option`: `None` means *not measurable
42/// here*, which consumers must treat as unconstrained.
43#[derive(Debug, Clone, Default, PartialEq)]
44pub struct TelemetrySample {
45    /// Resident set size of *this process*, in GiB.
46    ///
47    /// This — not system-wide usage — is what belongs in
48    /// [`crate::nas_engine::results::ResourceUsage::memory_gb`], because that field
49    /// is compared against a budget for the search itself. Charging the search for
50    /// every other process on the machine would abort it whenever the host is busy.
51    pub process_memory_gb: Option<f64>,
52
53    /// Total physical memory, in GiB (system capacity, not this search's usage).
54    pub total_memory_gb: Option<f64>,
55
56    /// Memory the OS reports as available for new allocations, in GiB.
57    pub available_memory_gb: Option<f64>,
58
59    /// Logical CPUs usable by this process.
60    pub logical_cpus: Option<usize>,
61
62    /// 1-minute load average divided by the logical CPU count, so `1.0` means
63    /// "fully loaded" independent of core count.
64    pub load_average_per_cpu: Option<f64>,
65
66    /// Number of processes visible on the system.
67    pub process_count: Option<usize>,
68
69    /// GPU devices visible to this process.
70    pub gpu_devices: Option<usize>,
71
72    /// GPU utilization in `[0, 1]`.
73    pub gpu_utilization: Option<f64>,
74
75    /// Total capacity of the filesystem holding the working directory, in GiB.
76    pub total_disk_gb: Option<f64>,
77
78    /// Free space on that filesystem, in GiB.
79    pub available_disk_gb: Option<f64>,
80
81    /// Network bandwidth, in MB/s.
82    pub network_bandwidth_mbps: Option<f64>,
83
84    /// Package/die temperature, in degrees Celsius.
85    pub temperature_celsius: Option<f64>,
86
87    /// Instantaneous power draw, in watts.
88    pub power_watts: Option<f64>,
89}
90
91impl TelemetrySample {
92    /// A sample in which nothing is known. Equivalent to
93    /// `TelemetrySample::default()`, spelled out so the intent is obvious at call
94    /// sites.
95    pub fn unknown() -> Self {
96        Self::default()
97    }
98
99    /// Whether every field is `None`.
100    pub fn is_fully_unknown(&self) -> bool {
101        *self == Self::default()
102    }
103}
104
105/// A source of system telemetry.
106///
107/// Injecting one is how a caller upgrades the honest-but-sparse default: wire in a
108/// source backed by whatever monitoring you already have, and the resource monitor
109/// starts enforcing the corresponding constraints. Constraints whose measurement is
110/// still `None` stay unenforced.
111pub trait TelemetrySource: Send + Sync + Debug {
112    /// Human-readable name of the source, used in log messages.
113    fn name(&self) -> &str;
114
115    /// Take a reading.
116    fn sample(&self) -> TelemetrySample;
117}
118
119/// The default source: standard library plus the Linux `/proc` files, and `None`
120/// for everything that cannot be measured without FFI.
121#[derive(Debug, Clone, Default)]
122pub struct StdTelemetry;
123
124impl StdTelemetry {
125    /// Create the default telemetry source.
126    pub fn new() -> Self {
127        Self
128    }
129}
130
131impl TelemetrySource for StdTelemetry {
132    fn name(&self) -> &str {
133        "std"
134    }
135
136    fn sample(&self) -> TelemetrySample {
137        let logical_cpus = std::thread::available_parallelism()
138            .ok()
139            .map(|count| count.get());
140
141        // `/proc` is absent on macOS and Windows; every read below therefore
142        // degrades to `None` rather than to a guess.
143        let process_memory_gb = read_to_string_if_present("/proc/self/status")
144            .and_then(|status| parse_vm_rss_kib(&status))
145            .map(|kib| kib as f64 / (1024.0 * 1024.0));
146
147        let (total_memory_gb, available_memory_gb) = match read_to_string_if_present(
148            "/proc/meminfo",
149        )
150        .and_then(|info| parse_meminfo(&info))
151        {
152            Some((total_kib, available_kib)) => (
153                Some(total_kib as f64 / (1024.0 * 1024.0)),
154                available_kib.map(|kib| kib as f64 / (1024.0 * 1024.0)),
155            ),
156            None => (None, None),
157        };
158
159        let load_average_per_cpu = read_to_string_if_present("/proc/loadavg")
160            .and_then(|loadavg| parse_loadavg(&loadavg))
161            .and_then(|load| logical_cpus.map(|cpus| load / cpus.max(1) as f64));
162
163        let process_count = count_proc_processes();
164
165        TelemetrySample {
166            process_memory_gb,
167            total_memory_gb,
168            available_memory_gb,
169            logical_cpus,
170            load_average_per_cpu,
171            process_count,
172            // Not measurable without a platform API / vendor library.
173            gpu_devices: None,
174            gpu_utilization: None,
175            total_disk_gb: None,
176            available_disk_gb: None,
177            network_bandwidth_mbps: None,
178            temperature_celsius: None,
179            power_watts: None,
180        }
181    }
182}
183
184/// Read a file, returning `None` if it does not exist or cannot be read. Used so
185/// the absence of `/proc` is an ordinary "unknown", not an error.
186fn read_to_string_if_present(path: &str) -> Option<String> {
187    let path = Path::new(path);
188    if !path.exists() {
189        return None;
190    }
191    std::fs::read_to_string(path).ok()
192}
193
194/// Extract `VmRSS` (in KiB) from the contents of `/proc/self/status`.
195///
196/// The file states its own units (`VmRSS:    123456 kB`), so no page-size lookup —
197/// and therefore no `sysconf` FFI — is needed.
198pub fn parse_vm_rss_kib(status: &str) -> Option<u64> {
199    for line in status.lines() {
200        let Some(rest) = line.strip_prefix("VmRSS:") else {
201            continue;
202        };
203        let mut fields = rest.split_whitespace();
204        let value: u64 = fields.next()?.parse().ok()?;
205        // Only kB is ever emitted by the kernel; anything else is unrecognised and
206        // is reported as unknown rather than misinterpreted.
207        return match fields.next() {
208            Some(unit) if unit.eq_ignore_ascii_case("kb") => Some(value),
209            None => Some(value),
210            Some(_) => None,
211        };
212    }
213    None
214}
215
216/// Extract `(MemTotal_kib, MemAvailable_kib)` from the contents of
217/// `/proc/meminfo`. `MemAvailable` is absent on kernels older than 3.14, hence the
218/// inner `Option`.
219pub fn parse_meminfo(meminfo: &str) -> Option<(u64, Option<u64>)> {
220    let mut total = None;
221    let mut available = None;
222    for line in meminfo.lines() {
223        let mut parts = line.split(':');
224        let key = parts.next()?.trim();
225        let Some(rest) = parts.next() else {
226            continue;
227        };
228        let Some(value) = rest.split_whitespace().next() else {
229            continue;
230        };
231        let Ok(parsed) = value.parse::<u64>() else {
232            continue;
233        };
234        match key {
235            "MemTotal" => total = Some(parsed),
236            "MemAvailable" => available = Some(parsed),
237            _ => {}
238        }
239    }
240    total.map(|total| (total, available))
241}
242
243/// Extract the 1-minute load average from the contents of `/proc/loadavg`.
244pub fn parse_loadavg(loadavg: &str) -> Option<f64> {
245    loadavg
246        .split_whitespace()
247        .next()
248        .and_then(|field| field.parse::<f64>().ok())
249        .filter(|value| value.is_finite() && *value >= 0.0)
250}
251
252/// Count numeric entries in `/proc`, which is the process count on Linux.
253fn count_proc_processes() -> Option<usize> {
254    let entries = std::fs::read_dir("/proc").ok()?;
255    let count = entries
256        .filter_map(|entry| entry.ok())
257        .filter(|entry| {
258            entry
259                .file_name()
260                .to_str()
261                .map(|name| !name.is_empty() && name.bytes().all(|b| b.is_ascii_digit()))
262                .unwrap_or(false)
263        })
264        .count();
265    Some(count)
266}
267
268/// A telemetry source returning a fixed sample. Intended for tests and for callers
269/// that already have measurements from elsewhere.
270#[derive(Debug, Clone)]
271pub struct FixedTelemetry {
272    name: String,
273    sample: TelemetrySample,
274}
275
276impl FixedTelemetry {
277    /// Wrap a pre-computed sample.
278    pub fn new(name: impl Into<String>, sample: TelemetrySample) -> Self {
279        Self {
280            name: name.into(),
281            sample,
282        }
283    }
284}
285
286impl TelemetrySource for FixedTelemetry {
287    fn name(&self) -> &str {
288        &self.name
289    }
290
291    fn sample(&self) -> TelemetrySample {
292        self.sample.clone()
293    }
294}
295
296#[cfg(test)]
297mod tests {
298    use super::*;
299
300    const STATUS_FIXTURE: &str = "Name:\tcargo\nUmask:\t0022\nState:\tR (running)\n\
301                                  VmPeak:\t 2097152 kB\nVmSize:\t 1048576 kB\n\
302                                  VmRSS:\t  524288 kB\nThreads:\t8\n";
303
304    const MEMINFO_FIXTURE: &str = "MemTotal:       32899072 kB\nMemFree:         1234567 kB\n\
305                                   MemAvailable:   20971520 kB\nBuffers:          123456 kB\n";
306
307    #[test]
308    fn vm_rss_is_parsed_from_the_status_fixture() {
309        assert_eq!(parse_vm_rss_kib(STATUS_FIXTURE), Some(524_288));
310        // 512 MiB expressed in GiB.
311        let gib: f64 = 524_288.0 / (1024.0 * 1024.0);
312        assert!((gib - 0.5).abs() < 1e-12);
313    }
314
315    #[test]
316    fn vm_rss_reports_unknown_rather_than_guessing() {
317        assert_eq!(parse_vm_rss_kib("Name:\tcargo\nThreads:\t8\n"), None);
318        assert_eq!(parse_vm_rss_kib(""), None);
319        // An unrecognised unit must not be silently treated as kB.
320        assert_eq!(parse_vm_rss_kib("VmRSS:\t 512 pages\n"), None);
321        // Missing unit (defensive): the kernel always writes kB.
322        assert_eq!(parse_vm_rss_kib("VmRSS:\t 4096\n"), Some(4096));
323        assert_eq!(parse_vm_rss_kib("VmRSS:\tnot-a-number kB\n"), None);
324    }
325
326    #[test]
327    fn meminfo_is_parsed_from_the_fixture() {
328        let (total, available) = parse_meminfo(MEMINFO_FIXTURE).expect("MemTotal present");
329        assert_eq!(total, 32_899_072);
330        assert_eq!(available, Some(20_971_520));
331    }
332
333    #[test]
334    fn meminfo_tolerates_a_kernel_without_memavailable() {
335        let (total, available) =
336            parse_meminfo("MemTotal:  1048576 kB\nMemFree:  524288 kB\n").expect("MemTotal");
337        assert_eq!(total, 1_048_576);
338        assert_eq!(available, None);
339        assert_eq!(parse_meminfo("MemFree: 1 kB\n"), None);
340        assert_eq!(parse_meminfo(""), None);
341    }
342
343    #[test]
344    fn loadavg_is_parsed_and_validated() {
345        assert_eq!(parse_loadavg("0.52 0.58 0.59 2/1234 56789\n"), Some(0.52));
346        assert_eq!(parse_loadavg(""), None);
347        assert_eq!(parse_loadavg("garbage 0.1 0.2\n"), None);
348        assert_eq!(parse_loadavg("-1.0 0.1 0.2\n"), None);
349    }
350
351    /// The whole point of F12: the default source must never invent a number.
352    #[test]
353    fn the_default_source_reports_only_measurable_values() {
354        let sample = StdTelemetry::new().sample();
355
356        // Available parallelism is portable, so this one must always be present.
357        assert!(
358            sample.logical_cpus.is_some_and(|cpus| cpus >= 1),
359            "available_parallelism must be reported"
360        );
361
362        // These are the fabricated readings the tracker used to emit. None of them
363        // is reachable without FFI, so all must be `None` on every platform.
364        assert_eq!(sample.gpu_devices, None, "the invented 4 GPUs must be gone");
365        assert_eq!(
366            sample.gpu_utilization, None,
367            "the invented 80% must be gone"
368        );
369        assert_eq!(sample.total_disk_gb, None, "the invented 1 TB must be gone");
370        assert_eq!(sample.available_disk_gb, None);
371        assert_eq!(
372            sample.network_bandwidth_mbps, None,
373            "the invented 1 GB/s must be gone"
374        );
375        assert_eq!(
376            sample.temperature_celsius, None,
377            "the invented 65 C must be gone"
378        );
379        assert_eq!(sample.power_watts, None, "the invented 250 W must be gone");
380
381        // Memory is only knowable where /proc exists. Whatever the platform, the
382        // value must be either absent or genuinely measured — never the old 32/16.
383        if let Some(total) = sample.total_memory_gb {
384            assert!(total > 0.0 && total.is_finite(), "got {total}");
385            assert_ne!(total, 32.0, "32.0 GB was the fabricated constant");
386        }
387        if let Some(process) = sample.process_memory_gb {
388            assert!(process > 0.0 && process.is_finite(), "got {process}");
389            assert_ne!(process, 16.0, "16.0 GB was the fabricated constant");
390        }
391        if let Some(load) = sample.load_average_per_cpu {
392            assert!(load >= 0.0 && load.is_finite());
393        }
394    }
395
396    #[test]
397    fn a_fixed_source_reports_exactly_what_it_was_given() {
398        let sample = TelemetrySample {
399            process_memory_gb: Some(3.5),
400            total_memory_gb: Some(64.0),
401            logical_cpus: Some(16),
402            temperature_celsius: Some(71.0),
403            ..TelemetrySample::unknown()
404        };
405        let source = FixedTelemetry::new("injected", sample.clone());
406        assert_eq!(source.name(), "injected");
407        assert_eq!(source.sample(), sample);
408        assert!(!sample.is_fully_unknown());
409        assert!(TelemetrySample::unknown().is_fully_unknown());
410    }
411}
optirs_nas/nas_engine/telemetry.rs

optirs_nas/nas_engine/
telemetry.rs
