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.
12
13use std::collections::HashMap;
14use std::io;
15use std::path::{Path, PathBuf};
16use std::sync::{Arc, Mutex};
17use std::time::Duration;
18
19use tokio::sync::watch;
20
21use crate::api;
22use crate::store::{SampleRecord, Store};
23use crate::sync::lock;
24use crate::time;
25
26/// How often running containers are sampled.
27pub const SAMPLE_INTERVAL: Duration = Duration::from_secs(2);
28
29/// Where the cgroup and proc file systems are mounted: the host's, or a test's fixture tree.
30#[derive(Debug, Clone, PartialEq, Eq)]
31pub struct Paths {
32    /// Usually `/sys/fs/cgroup`.
33    pub cgroup: PathBuf,
34    /// Usually `/proc`.
35    pub proc: PathBuf,
36}
37
38impl Default for Paths {
39    fn default() -> Paths {
40        Paths {
41            cgroup: PathBuf::from("/sys/fs/cgroup"),
42            proc: PathBuf::from("/proc"),
43        }
44    }
45}
46
47impl Paths {
48    /// The cgroup of a systemd slice: `a-b.slice` is `a.slice/a-b.slice`.
49    pub fn slice(&self, slice: &str) -> PathBuf {
50        let name = slice.strip_suffix(".slice").unwrap_or(slice);
51        let mut dir = self.cgroup.clone();
52        let mut prefix = String::new();
53        for part in name.split('-') {
54            if !prefix.is_empty() {
55                prefix.push('-');
56            }
57            prefix.push_str(part);
58            dir.push(format!("{prefix}.slice"));
59        }
60        dir
61    }
62
63    /// The cgroup Docker's systemd driver gives a container in a slice.
64    pub fn container(&self, slice: &str, container: &str) -> PathBuf {
65        self.slice(slice).join(format!("docker-{container}.scope"))
66    }
67}
68
69/// A cgroup's counters at one moment.
70#[derive(Debug, Clone, Copy, PartialEq, Eq)]
71pub struct Usage {
72    /// CPU time so far, in microseconds (`cpu.stat` `usage_usec`).
73    pub cpu_usage: u64,
74    /// `memory.current` less `active_file` and `inactive_file` from `memory.stat`: without page cache.
75    pub memory: u64,
76    /// `memory.current` less `inactive_file`: page cache read twice still counts.
77    pub memory_with_active_file: u64,
78}
79
80/// What a cgroup's `memory.events` counted so far, its descendants' included; `None` for a counter the file lacks
81/// or that doesn't parse, which says nothing either way.
82#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
83pub struct Events {
84    /// Times its memory, or a descendant's under its own limit, reached the limit, reclaim could not bring it down
85    /// and an allocation was about to fail (`oom`). Not `max`, which counts every reclaim at the limit, page
86    /// cache's included.
87    pub oom: Option<u64>,
88    /// Processes the kernel killed in it for memory (`oom_kill`).
89    pub oom_kill: Option<u64>,
90}
91
92fn read(path: &Path) -> io::Result<String> {
93    std::fs::read_to_string(path)
94}
95
96fn field(text: &str, key: &str) -> Option<u64> {
97    text.lines().find_map(|line| {
98        let (name, value) = line.split_once(' ')?;
99        (name == key).then(|| value.trim().parse().ok()).flatten()
100    })
101}
102
103fn invalid(path: &Path) -> io::Error {
104    io::Error::new(
105        io::ErrorKind::InvalidData,
106        format!("{} has an unexpected format", path.display()),
107    )
108}
109
110/// Reads a cgroup's CPU and memory counters.
111pub fn read_usage(cgroup: &Path) -> io::Result<Usage> {
112    let current_path = cgroup.join("memory.current");
113    let current: u64 = read(&current_path)?
114        .trim()
115        .parse()
116        .map_err(|_| invalid(&current_path))?;
117    let stat_path = cgroup.join("memory.stat");
118    let stat = read(&stat_path)?;
119    let (Some(active_file), Some(inactive_file)) =
120        (field(&stat, "active_file"), field(&stat, "inactive_file"))
121    else {
122        return Err(invalid(&stat_path));
123    };
124    let cpu_path = cgroup.join("cpu.stat");
125    let cpu_usage = field(&read(&cpu_path)?, "usage_usec").ok_or_else(|| invalid(&cpu_path))?;
126    Ok(Usage {
127        cpu_usage,
128        memory: current.saturating_sub(active_file + inactive_file),
129        memory_with_active_file: current.saturating_sub(inactive_file),
130    })
131}
132
133/// A cgroup's `memory.events`; every counter `None` when the file can't be read.
134pub fn read_events(cgroup: &Path) -> Events {
135    let Ok(events) = read(&cgroup.join("memory.events")) else {
136        return Events::default();
137    };
138    Events {
139        oom: field(&events, "oom"),
140        oom_kill: field(&events, "oom_kill"),
141    }
142}
143
144/// Bytes received and sent by a process's network namespace on its `eth*` interfaces: the container's own
145/// traffic, without its loopback or the bridge of the Docker Engine inside it, whose traffic also crosses `eth0`.
146pub fn read_network(paths: &Paths, pid: u32) -> io::Result<(u64, u64)> {
147    let path = paths.proc.join(pid.to_string()).join("net/dev");
148    let text = read(&path)?;
149    let mut totals = (0, 0);
150    for line in text.lines().skip(2) {
151        let Some((name, counters)) = line.split_once(':') else {
152            return Err(invalid(&path));
153        };
154        if !name.trim().starts_with("eth") {
155            continue;
156        }
157        let counters: Vec<u64> = counters
158            .split_whitespace()
159            .map(str::parse)
160            .collect::<Result<_, _>>()
161            .map_err(|_| invalid(&path))?;
162        // Receive: bytes packets errs drop fifo frame compressed multicast; then transmit, bytes first.
163        let (Some(received), Some(sent)) = (counters.first(), counters.get(8)) else {
164            return Err(invalid(&path));
165        };
166        totals.0 += received;
167        totals.1 += sent;
168    }
169    Ok(totals)
170}
171
172/// A runner container to sample.
173#[derive(Debug, Clone, PartialEq, Eq)]
174pub struct Tracked {
175    /// The runner's name, which keys its samples.
176    pub runner: String,
177    /// The full Docker container ID.
178    pub container: String,
179    /// The container's main process on the host, for its network counters.
180    pub pid: u32,
181}
182
183/// Samples containers and their slice, turning CPU time into CPU use between samples.
184#[derive(Debug)]
185pub struct Sampler {
186    paths: Paths,
187    slice: String,
188    previous: HashMap<String, (u64, u64)>,
189    previous_host: Option<(u64, u64)>,
190}
191
192/// Millicores used between two readings of CPU time (µs) taken at two times (ms).
193fn millicores(previous: Option<(u64, u64)>, at: u64, cpu_usage: u64) -> u64 {
194    previous.map_or(0, |(then, usage)| {
195        cpu_usage
196            .saturating_sub(usage)
197            .checked_div(at.saturating_sub(then))
198            .unwrap_or(0)
199    })
200}
201
202impl Sampler {
203    /// A sampler for containers in `slice`.
204    pub fn new(paths: Paths, slice: &str) -> Sampler {
205        Sampler {
206            paths,
207            slice: slice.to_owned(),
208            previous: HashMap::new(),
209            previous_host: None,
210        }
211    }
212
213    /// One sample of every tracked container that still exists, and of the slice, at `at` (ms). A container
214    /// whose files are gone gets no sample, and is forgotten.
215    pub fn sample(
216        &mut self,
217        tracked: &[Tracked],
218        at: u64,
219    ) -> (Vec<SampleRecord>, Option<api::HostSample>) {
220        let mut samples = Vec::new();
221        let mut seen = HashMap::new();
222        for container in tracked {
223            let cgroup = self.paths.container(&self.slice, &container.container);
224            let (Ok(usage), Ok((received, sent))) = (
225                read_usage(&cgroup),
226                read_network(&self.paths, container.pid),
227            ) else {
228                continue;
229            };
230            let cpu = millicores(
231                self.previous.get(&container.runner).copied(),
232                at,
233                usage.cpu_usage,
234            );
235            seen.insert(container.runner.clone(), (at, usage.cpu_usage));
236            samples.push(SampleRecord {
237                runner: container.runner.clone(),
238                at,
239                cpu,
240                cpu_usage: usage.cpu_usage,
241                memory: usage.memory,
242                network_received: received,
243                network_sent: sent,
244                events: read_events(&cgroup),
245            });
246        }
247        self.previous = seen;
248        let host = read_usage(&self.paths.slice(&self.slice))
249            .ok()
250            .map(|usage| {
251                let cpu = millicores(self.previous_host, at, usage.cpu_usage);
252                self.previous_host = Some((at, usage.cpu_usage));
253                api::HostSample {
254                    at: at / 1000,
255                    cpu,
256                    memory: usage.memory_with_active_file,
257                }
258            });
259        (samples, host)
260    }
261}
262
263/// Samples the tracked containers every `SAMPLE_INTERVAL` into the store, until the engine drops its sender.
264pub async fn run(state: Sampler, tracked: watch::Receiver<Vec<Tracked>>, store: Store) {
265    let state = Arc::new(Mutex::new(state));
266    let mut interval = tokio::time::interval(SAMPLE_INTERVAL);
267    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
268    loop {
269        interval.tick().await;
270        if tracked.has_changed().is_err() {
271            return;
272        }
273        let containers = tracked.borrow().clone();
274        let state = state.clone();
275        // Reading cgroup files is quick but blocking.
276        let Ok((samples, host)) = tokio::task::spawn_blocking(move || {
277            // A panic midway through a sample leaves the counters of the last or of this one, either of which the
278            // next sample measures from.
279            lock(&state).sample(&containers, time::now_ms())
280        })
281        .await
282        else {
283            continue;
284        };
285        if !samples.is_empty() {
286            store.samples(samples);
287        }
288        if let Some(host) = host {
289            store.host_sample(host);
290        }
291    }
292}
runnerd_core/observe/metrics.rs

runnerd_core/observe/
metrics.rs
