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/// Bytes received and sent by a process's network namespace on its `eth*` interfaces: the container's own
108/// traffic, without its loopback or the bridge of the Docker Engine inside it, whose traffic also crosses `eth0`.
109pub fn read_network(paths: &Paths, pid: u32) -> io::Result<(u64, u64)> {
110    let path = paths.proc.join(pid.to_string()).join("net/dev");
111    let text = read(&path)?;
112    let mut totals = (0, 0);
113    for line in text.lines().skip(2) {
114        let Some((name, counters)) = line.split_once(':') else {
115            return Err(invalid(&path));
116        };
117        if !name.trim().starts_with("eth") {
118            continue;
119        }
120        let counters: Vec<u64> = counters
121            .split_whitespace()
122            .map(str::parse)
123            .collect::<Result<_, _>>()
124            .map_err(|_| invalid(&path))?;
125        // Receive: bytes packets errs drop fifo frame compressed multicast; then transmit, bytes first.
126        let (Some(received), Some(sent)) = (counters.first(), counters.get(8)) else {
127            return Err(invalid(&path));
128        };
129        totals.0 += received;
130        totals.1 += sent;
131    }
132    Ok(totals)
133}
134
135/// A runner container to sample.
136#[derive(Debug, Clone, PartialEq, Eq)]
137pub struct Tracked {
138    /// The runner's name, which keys its samples.
139    pub runner: String,
140    /// The full Docker container ID.
141    pub container: String,
142    /// The container's main process on the host, for its network counters.
143    pub pid: u32,
144}
145
146/// Samples containers and their slice, turning CPU time into CPU use between samples.
147#[derive(Debug)]
148pub struct Sampler {
149    paths: Paths,
150    slice: String,
151    previous: HashMap<String, (u64, u64)>,
152    previous_host: Option<(u64, u64)>,
153}
154
155/// Millicores used between two readings of CPU time (µs) taken at two times (ms).
156fn millicores(previous: Option<(u64, u64)>, at: u64, cpu_usage: u64) -> u64 {
157    previous.map_or(0, |(then, usage)| {
158        cpu_usage
159            .saturating_sub(usage)
160            .checked_div(at.saturating_sub(then))
161            .unwrap_or(0)
162    })
163}
164
165impl Sampler {
166    /// A sampler for containers in `slice`.
167    pub fn new(paths: Paths, slice: &str) -> Sampler {
168        Sampler {
169            paths,
170            slice: slice.to_owned(),
171            previous: HashMap::new(),
172            previous_host: None,
173        }
174    }
175
176    /// One sample of every tracked container that still exists, and of the slice, at `at` (ms). A container
177    /// whose files are gone gets no sample, and is forgotten.
178    pub fn sample(
179        &mut self,
180        tracked: &[Tracked],
181        at: u64,
182    ) -> (Vec<SampleRecord>, Option<api::HostSample>) {
183        let mut samples = Vec::new();
184        let mut seen = HashMap::new();
185        for container in tracked {
186            let cgroup = self.paths.container(&self.slice, &container.container);
187            let (Ok(usage), Ok((received, sent))) = (
188                read_usage(&cgroup),
189                read_network(&self.paths, container.pid),
190            ) else {
191                continue;
192            };
193            let cpu = millicores(
194                self.previous.get(&container.runner).copied(),
195                at,
196                usage.cpu_usage,
197            );
198            seen.insert(container.runner.clone(), (at, usage.cpu_usage));
199            samples.push(SampleRecord {
200                runner: container.runner.clone(),
201                at,
202                cpu,
203                cpu_usage: usage.cpu_usage,
204                memory: usage.memory,
205                network_received: received,
206                network_sent: sent,
207            });
208        }
209        self.previous = seen;
210        let host = read_usage(&self.paths.slice(&self.slice))
211            .ok()
212            .map(|usage| {
213                let cpu = millicores(self.previous_host, at, usage.cpu_usage);
214                self.previous_host = Some((at, usage.cpu_usage));
215                api::HostSample {
216                    at: at / 1000,
217                    cpu,
218                    memory: usage.memory,
219                }
220            });
221        (samples, host)
222    }
223}
224
225/// Samples the tracked containers every `SAMPLE_INTERVAL` into the store, until the engine drops its sender.
226pub async fn run(state: Sampler, tracked: watch::Receiver<Vec<Tracked>>, store: Store) {
227    let state = Arc::new(Mutex::new(state));
228    let mut interval = tokio::time::interval(SAMPLE_INTERVAL);
229    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
230    loop {
231        interval.tick().await;
232        if tracked.has_changed().is_err() {
233            return;
234        }
235        let containers = tracked.borrow().clone();
236        let state = state.clone();
237        // Reading cgroup files is quick but blocking.
238        let Ok((samples, host)) = tokio::task::spawn_blocking(move || {
239            // A panic midway through a sample leaves the counters of the last or of this one, either of which the
240            // next sample measures from.
241            lock(&state).sample(&containers, time::now_ms())
242        })
243        .await
244        else {
245            continue;
246        };
247        if !samples.is_empty() {
248            store.samples(samples);
249        }
250        if let Some(host) = host {
251            store.host_sample(host);
252        }
253    }
254}
runnerd_core/observe/metrics.rs

runnerd_core/observe/
metrics.rs
