runnerd_core/observe/
metrics.rs

1//! CPU, memory and network of runner containers and of their slice, read from cgroup v2 files and
2//! `/proc/<pid>/net/dev` every `SAMPLE_INTERVAL`. A container's memory is an estimate of what its job holds without
3//! page cache: `memory.current` less the file LRU lists (`active_file`, `inactive_file`), so anonymous memory, tmpfs
4//! and shared memory (`shmem`, on the anon lists) and kernel memory, reclaimable slab included, since slab in use
5//! (an open file's dentry and inode) can't be freed. Page cache on the file lists is usually the kernel's to drop
6//! before `memory.max` OOM-kills ("reaches this limit and can't be reduced", Documentation/admin-guide/
7//! cgroup-v2.rst), though dirty, writeback or pinned pages may not be dropped at once. `docker stats` subtracts only
8//! `inactive_file`, so it counts page cache read twice. The slice's memory for the high-water mark is that
9//! conservative number, `memory.current` less `inactive_file`, as it always was: it stops starts before the slice
10//! is short of memory, not after. Whether a cgroup met its limit comes from `memory.events`, never from how close
11//! a sample came to it. A container whose files vanish ends its series; that is not an error. Beside the slice, it
12//! reads the host's memory: `MemTotal` and `MemAvailable` from `/proc/meminfo` and the stall shares of
13//! `/proc/pressure/memory`.
14
15use std::collections::HashMap;
16use std::io;
17use std::path::{Path, PathBuf};
18use std::sync::{Arc, Mutex};
19use std::time::Duration;
20
21use tokio::sync::watch;
22
23use crate::api;
24use crate::store::{SampleRecord, Store};
25use crate::sync::lock;
26use crate::time;
27
28/// How often running containers are sampled.
29pub const SAMPLE_INTERVAL: Duration = Duration::from_secs(2);
30
31/// Where the cgroup and proc file systems are mounted: the host's, or a test's fixture tree.
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub struct Paths {
34    /// Usually `/sys/fs/cgroup`.
35    pub cgroup: PathBuf,
36    /// Usually `/proc`.
37    pub proc: PathBuf,
38}
39
40impl Default for Paths {
41    fn default() -> Paths {
42        Paths {
43            cgroup: PathBuf::from("/sys/fs/cgroup"),
44            proc: PathBuf::from("/proc"),
45        }
46    }
47}
48
49impl Paths {
50    /// The cgroup of a systemd slice: `a-b.slice` is `a.slice/a-b.slice`.
51    pub fn slice(&self, slice: &str) -> PathBuf {
52        let name = slice.strip_suffix(".slice").unwrap_or(slice);
53        let mut dir = self.cgroup.clone();
54        let mut prefix = String::new();
55        for part in name.split('-') {
56            if !prefix.is_empty() {
57                prefix.push('-');
58            }
59            prefix.push_str(part);
60            dir.push(format!("{prefix}.slice"));
61        }
62        dir
63    }
64
65    /// The cgroup Docker's systemd driver gives a container in a slice.
66    pub fn container(&self, slice: &str, container: &str) -> PathBuf {
67        self.slice(slice).join(format!("docker-{container}.scope"))
68    }
69}
70
71/// A cgroup's counters at one moment.
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub struct Usage {
74    /// CPU time so far, in microseconds (`cpu.stat` `usage_usec`).
75    pub cpu_usage: u64,
76    /// `memory.current` less `active_file` and `inactive_file` from `memory.stat`: without page cache.
77    pub memory: u64,
78    /// `memory.current` less `inactive_file`: page cache read twice still counts.
79    pub memory_with_active_file: u64,
80}
81
82/// What a cgroup's `memory.events` counted so far, its descendants' included; `None` for a counter the file lacks
83/// or that doesn't parse, which says nothing either way.
84#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
85pub struct Events {
86    /// Times its memory, or a descendant's under its own limit, reached the limit, reclaim could not bring it down
87    /// and an allocation was about to fail (`oom`). Not `max`, which counts every reclaim at the limit, page
88    /// cache's included.
89    pub oom: Option<u64>,
90    /// Processes the kernel killed in it for memory (`oom_kill`).
91    pub oom_kill: Option<u64>,
92    /// Times a socket's receive window was held small because its memory, or a descendant's, was under pressure
93    /// (`sock_throttled`): reclaim that went badly, of the host or of a limit, slows its downloads. `None` on a
94    /// kernel without the counter.
95    pub sock_throttled: Option<u64>,
96}
97
98fn read(path: &Path) -> io::Result<String> {
99    std::fs::read_to_string(path)
100}
101
102fn field(text: &str, key: &str) -> Option<u64> {
103    text.lines().find_map(|line| {
104        let (name, value) = line.split_once(' ')?;
105        (name == key).then(|| value.trim().parse().ok()).flatten()
106    })
107}
108
109fn invalid(path: &Path) -> io::Error {
110    io::Error::new(
111        io::ErrorKind::InvalidData,
112        format!("{} has an unexpected format", path.display()),
113    )
114}
115
116/// Reads a cgroup's CPU and memory counters.
117pub fn read_usage(cgroup: &Path) -> io::Result<Usage> {
118    let current_path = cgroup.join("memory.current");
119    let current: u64 = read(&current_path)?
120        .trim()
121        .parse()
122        .map_err(|_| invalid(&current_path))?;
123    let stat_path = cgroup.join("memory.stat");
124    let stat = read(&stat_path)?;
125    let (Some(active_file), Some(inactive_file)) =
126        (field(&stat, "active_file"), field(&stat, "inactive_file"))
127    else {
128        return Err(invalid(&stat_path));
129    };
130    let cpu_path = cgroup.join("cpu.stat");
131    let cpu_usage = field(&read(&cpu_path)?, "usage_usec").ok_or_else(|| invalid(&cpu_path))?;
132    Ok(Usage {
133        cpu_usage,
134        memory: current.saturating_sub(active_file + inactive_file),
135        memory_with_active_file: current.saturating_sub(inactive_file),
136    })
137}
138
139/// What a cgroup holds that `MemAvailable` does not count as available: its memory without page cache less its
140/// reclaimable slab (`slab_reclaimable`), which `MemAvailable` counts already. 0 when it can't be read.
141pub fn read_unavailable(cgroup: &Path) -> u64 {
142    let slab = read(&cgroup.join("memory.stat"))
143        .ok()
144        .and_then(|stat| field(&stat, "slab_reclaimable"))
145        .unwrap_or(0);
146    read_usage(cgroup).map_or(0, |usage| usage.memory.saturating_sub(slab))
147}
148
149/// A cgroup's `memory.events`; every counter `None` when the file can't be read.
150pub fn read_events(cgroup: &Path) -> Events {
151    let Ok(events) = read(&cgroup.join("memory.events")) else {
152        return Events::default();
153    };
154    Events {
155        oom: field(&events, "oom"),
156        oom_kill: field(&events, "oom_kill"),
157        sock_throttled: field(&events, "sock_throttled"),
158    }
159}
160
161/// The host's memory from `/proc/meminfo` and `/proc/pressure/memory`, beside `slice` bytes the runners hold
162/// without page cache, at `at` (seconds). `None` without `MemTotal` and `MemAvailable`; the pressure lines are `None`
163/// on a kernel without pressure stall information.
164pub fn read_host_memory(paths: &Paths, slice: u64, at: u64) -> Option<api::HostMemory> {
165    let meminfo = read(&paths.proc.join("meminfo")).ok()?;
166    let kib = |key: &str| {
167        meminfo.lines().find_map(|line| {
168            let value = line.strip_prefix(key)?.strip_prefix(':')?;
169            value.trim().strip_suffix("kB")?.trim().parse::<u64>().ok()
170        })
171    };
172    let pressure = read(&paths.proc.join("pressure/memory")).unwrap_or_default();
173    let line = |kind: &str| {
174        let fields = pressure
175            .lines()
176            .find_map(|line| line.strip_prefix(kind)?.strip_prefix(' '))?;
177        let avg = |key: &str| {
178            fields.split_whitespace().find_map(|field| {
179                let value = field.strip_prefix(key)?.strip_prefix('=')?;
180                let (whole, fraction) = value.split_once('.').unwrap_or((value, "0"));
181                let hundredths: u32 = format!("{fraction:0<2}").get(..2)?.parse().ok()?;
182                Some(
183                    whole
184                        .parse::<u32>()
185                        .ok()?
186                        .saturating_mul(100)
187                        .saturating_add(hundredths),
188                )
189            })
190        };
191        Some(api::Pressure {
192            avg10: avg("avg10")?,
193            avg60: avg("avg60")?,
194        })
195    };
196    Some(api::HostMemory {
197        at,
198        total: kib("MemTotal")?.saturating_mul(1024),
199        available: kib("MemAvailable")?.saturating_mul(1024),
200        slice,
201        some: line("some"),
202        full: line("full"),
203    })
204}
205
206/// Bytes received and sent by a process's network namespace on its `eth*` interfaces: the container's own
207/// traffic, without its loopback or the bridge of the Docker Engine inside it, whose traffic also crosses `eth0`.
208pub fn read_network(paths: &Paths, pid: u32) -> io::Result<(u64, u64)> {
209    let path = paths.proc.join(pid.to_string()).join("net/dev");
210    let text = read(&path)?;
211    let mut totals = (0, 0);
212    for line in text.lines().skip(2) {
213        let Some((name, counters)) = line.split_once(':') else {
214            return Err(invalid(&path));
215        };
216        if !name.trim().starts_with("eth") {
217            continue;
218        }
219        let counters: Vec<u64> = counters
220            .split_whitespace()
221            .map(str::parse)
222            .collect::<Result<_, _>>()
223            .map_err(|_| invalid(&path))?;
224        // Receive: bytes packets errs drop fifo frame compressed multicast; then transmit, bytes first.
225        let (Some(received), Some(sent)) = (counters.first(), counters.get(8)) else {
226            return Err(invalid(&path));
227        };
228        totals.0 += received;
229        totals.1 += sent;
230    }
231    Ok(totals)
232}
233
234/// A runner container to sample.
235#[derive(Debug, Clone, PartialEq, Eq)]
236pub struct Tracked {
237    /// The runner's name, which keys its samples.
238    pub runner: String,
239    /// The full Docker container ID.
240    pub container: String,
241    /// The container's main process on the host, for its network counters.
242    pub pid: u32,
243}
244
245/// Samples containers and their slice, turning CPU time into CPU use between samples.
246#[derive(Debug)]
247pub struct Sampler {
248    paths: Paths,
249    slice: String,
250    previous: HashMap<String, (u64, u64)>,
251    previous_host: Option<(u64, u64)>,
252}
253
254/// Millicores used between two readings of CPU time (µs) taken at two times (ms).
255fn millicores(previous: Option<(u64, u64)>, at: u64, cpu_usage: u64) -> u64 {
256    previous.map_or(0, |(then, usage)| {
257        cpu_usage
258            .saturating_sub(usage)
259            .checked_div(at.saturating_sub(then))
260            .unwrap_or(0)
261    })
262}
263
264impl Sampler {
265    /// A sampler for containers in `slice`.
266    pub fn new(paths: Paths, slice: &str) -> Sampler {
267        Sampler {
268            paths,
269            slice: slice.to_owned(),
270            previous: HashMap::new(),
271            previous_host: None,
272        }
273    }
274
275    /// The host's memory at `at` (ms), beside what the slice holds that `MemAvailable` leaves out
276    /// (`read_unavailable`).
277    pub fn host_memory(&self, at: u64) -> Option<api::HostMemory> {
278        let slice = read_unavailable(&self.paths.slice(&self.slice));
279        read_host_memory(&self.paths, slice, at / 1000)
280    }
281
282    /// One sample of every tracked container that still exists, and of the slice, at `at` (ms). A container
283    /// whose files are gone gets no sample, and is forgotten.
284    pub fn sample(
285        &mut self,
286        tracked: &[Tracked],
287        at: u64,
288    ) -> (Vec<SampleRecord>, Option<api::HostSample>) {
289        let mut samples = Vec::new();
290        let mut seen = HashMap::new();
291        for container in tracked {
292            let cgroup = self.paths.container(&self.slice, &container.container);
293            let (Ok(usage), Ok((received, sent))) = (
294                read_usage(&cgroup),
295                read_network(&self.paths, container.pid),
296            ) else {
297                continue;
298            };
299            let cpu = millicores(
300                self.previous.get(&container.runner).copied(),
301                at,
302                usage.cpu_usage,
303            );
304            seen.insert(container.runner.clone(), (at, usage.cpu_usage));
305            samples.push(SampleRecord {
306                runner: container.runner.clone(),
307                at,
308                cpu,
309                cpu_usage: usage.cpu_usage,
310                memory: usage.memory,
311                network_received: received,
312                network_sent: sent,
313                events: read_events(&cgroup),
314            });
315        }
316        self.previous = seen;
317        let host = read_usage(&self.paths.slice(&self.slice))
318            .ok()
319            .map(|usage| {
320                let cpu = millicores(self.previous_host, at, usage.cpu_usage);
321                self.previous_host = Some((at, usage.cpu_usage));
322                api::HostSample {
323                    at: at / 1000,
324                    cpu,
325                    memory: usage.memory_with_active_file,
326                }
327            });
328        (samples, host)
329    }
330}
331
332/// Samples the tracked containers every `SAMPLE_INTERVAL` into the store, until the engine drops its sender.
333pub async fn run(state: Sampler, tracked: watch::Receiver<Vec<Tracked>>, store: Store) {
334    let state = Arc::new(Mutex::new(state));
335    let mut interval = tokio::time::interval(SAMPLE_INTERVAL);
336    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
337    loop {
338        interval.tick().await;
339        if tracked.has_changed().is_err() {
340            return;
341        }
342        let containers = tracked.borrow().clone();
343        let state = state.clone();
344        // Reading cgroup files is quick but blocking.
345        let Ok(((samples, host), memory)) = tokio::task::spawn_blocking(move || {
346            // A panic midway through a sample leaves the counters of the last or of this one, either of which the
347            // next sample measures from.
348            let mut state = lock(&state);
349            let at = time::now_ms();
350            (state.sample(&containers, at), state.host_memory(at))
351        })
352        .await
353        else {
354            continue;
355        };
356        if let Some(memory) = memory {
357            store.host_memory(memory);
358        }
359        if !samples.is_empty() {
360            store.samples(samples);
361        }
362        if let Some(host) = host {
363            store.host_sample(host);
364        }
365    }
366}
runnerd_core/observe/metrics.rs

runnerd_core/observe/
metrics.rs
