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

runnerd_core/observe/
metrics.rs
