micromegas_telemetry_sink/
system_monitor.rs

1use micromegas_tracing::{fmetric, imetric};
2use std::time::{Duration, Instant};
3use sysinfo::{CpuRefreshKind, MemoryRefreshKind, RefreshKind, System};
4
5/// The sampling frequency for the process/jemalloc memory gauges -- much
6/// coarser than the loop's own ~200ms `sysinfo` CPU-update floor (that floor
7/// is only needed for a valid CPU-usage delta). Matches the cadence
8/// object-cache-srv's saturation_monitor.rs already chose for the same
9/// telemetry-volume reason; these gauges don't need 200ms resolution to
10/// catch an hour-plus OOM climb.
11const SLOW_SAMPLE_INTERVAL: Duration = Duration::from_secs(5);
12
13/// True once at least `SLOW_SAMPLE_INTERVAL` has elapsed since the last slow
14/// sample. Extracted as a pure function (mirroring `saturation_monitor.rs`'s
15/// `sample_once` extraction) so the gating decision itself is directly
16/// testable with plain `Duration` values, rather than only reachable from
17/// inside the infinite loop below.
18pub fn should_sample_slow(elapsed_since_last_sample: Duration) -> bool {
19    elapsed_since_last_sample >= SLOW_SAMPLE_INTERVAL
20}
21
22/// Emits this process's own RSS/virtual memory size. Allocator-agnostic
23/// (reads from the OS via `sysinfo`), so this runs for every consumer of
24/// `spawn_system_monitor`, including non-jemalloc binaries.
25///
26/// Takes a caller-owned `System` so repeated calls reuse it instead of each
27/// constructing a fresh instance (which re-reads `/proc/stat` etc.).
28pub fn emit_process_memory_stats(system: &mut System) {
29    use sysinfo::{ProcessRefreshKind, ProcessesToUpdate};
30    let Ok(pid) = sysinfo::get_current_pid() else {
31        return;
32    };
33    system.refresh_processes_specifics(
34        ProcessesToUpdate::Some(&[pid]),
35        false,
36        ProcessRefreshKind::nothing().with_memory(),
37    );
38    if let Some(process) = system.process(pid) {
39        imetric!("process_resident_bytes", "bytes", process.memory());
40        imetric!("process_virtual_bytes", "bytes", process.virtual_memory());
41    }
42}
43
44/// Emits jemalloc's own runtime accounting (`stats.allocated`/`resident`/
45/// `mapped`/`retained`). Only meaningful when jemalloc is this process's
46/// global allocator, hence the feature + platform gate.
47#[cfg(all(feature = "jemalloc", not(target_os = "windows")))]
48pub fn emit_jemalloc_stats() {
49    use tikv_jemalloc_ctl::{epoch, stats};
50    // jemalloc caches these counters; advance the epoch to refresh them
51    // before reading, per tikv-jemalloc-ctl's documented usage.
52    if epoch::advance().is_err() {
53        return;
54    }
55    if let Ok(v) = stats::allocated::read() {
56        imetric!("jemalloc_allocated_bytes", "bytes", v as u64);
57    }
58    if let Ok(v) = stats::resident::read() {
59        imetric!("jemalloc_resident_bytes", "bytes", v as u64);
60    }
61    if let Ok(v) = stats::mapped::read() {
62        imetric!("jemalloc_mapped_bytes", "bytes", v as u64);
63    }
64    if let Ok(v) = stats::retained::read() {
65        imetric!("jemalloc_retained_bytes", "bytes", v as u64);
66    }
67}
68
69#[cfg(not(all(feature = "jemalloc", not(target_os = "windows"))))]
70pub fn emit_jemalloc_stats() {}
71
72/// Continuously sends system-wide CPU and memory usage metrics.
73///
74/// This function runs in a loop, refreshing system information at regular intervals
75/// and emitting `imetric!` and `fmetric!` events for total memory, used memory,
76/// free memory, and global CPU usage. Every `SLOW_SAMPLE_INTERVAL`, it also
77/// emits this process's RSS/virtual memory size and, when built with the
78/// `jemalloc` feature on a non-Windows target, jemalloc's own runtime stats.
79pub fn send_system_metrics_forever() {
80    let what_to_refresh = RefreshKind::nothing()
81        .with_cpu(CpuRefreshKind::nothing().with_cpu_usage())
82        .with_memory(MemoryRefreshKind::nothing().with_ram());
83    let mut system = System::new_with_specifics(what_to_refresh);
84    imetric!("total_memory", "bytes", system.total_memory());
85    let mut process_system = System::new();
86    let mut last_slow_sample = Instant::now();
87    loop {
88        std::thread::sleep(sysinfo::MINIMUM_CPU_UPDATE_INTERVAL);
89        system.refresh_specifics(what_to_refresh);
90        imetric!("used_memory", "bytes", system.used_memory());
91        imetric!("free_memory", "bytes", system.free_memory());
92        fmetric!("cpu_usage", "percent", system.global_cpu_usage() as f64);
93
94        if should_sample_slow(last_slow_sample.elapsed()) {
95            emit_process_memory_stats(&mut process_system);
96            emit_jemalloc_stats();
97            last_slow_sample = Instant::now();
98        }
99    }
100}
101
102/// Spawns a new thread to run the `send_system_metrics_forever` function.
103///
104/// This allows system metrics to be collected and reported in the background.
105pub fn spawn_system_monitor() {
106    std::thread::spawn(send_system_metrics_forever);
107}
micromegas_telemetry_sink/system_monitor.rs

micromegas_telemetry_sink/
system_monitor.rs
