runnerd_core/observe/
mirror.rs

1//! Which runner the Docker Hub mirror served, and how many bytes: the mirror's access log (Docker log lines with
2//! timestamps, in the combined log format) attributed by source address. A runner container's Docker Engine pulls
3//! for every container of its job, so the mirror sees the runner container's address. An address belongs to a
4//! runner only between its container's start and its removal, so a later container given the same address never
5//! gets an earlier one's bytes. Docker's `--since` takes whole seconds, so each read starts with lines already
6//! counted; a `MirrorPosition` skips them by their exact time, and among lines of the same time, by how many.
7
8use crate::store::MirrorPosition;
9
10/// One response the mirror logged.
11#[derive(Debug, Clone, PartialEq, Eq)]
12pub struct Served {
13    /// When Docker received the log line, in nanoseconds since the epoch.
14    pub at: u64,
15    /// The client's address.
16    pub address: String,
17    /// Bytes in the response body.
18    pub bytes: u64,
19}
20
21/// Parses a line of `docker logs --timestamps` of the mirror: `<RFC 3339> <address> - - [<time>] "<request>"
22/// <status> <bytes> …`. Lines of another format (the registry's own messages) are `None`.
23pub fn parse(line: &str) -> Option<Served> {
24    let (timestamp, line) = line.split_once(' ')?;
25    let at = crate::time::parse_ns(timestamp)?;
26    let (address, rest) = line.split_once(" - - [")?;
27    let (_, rest) = rest.split_once("] \"")?;
28    let (_request, rest) = rest.split_once("\" ")?;
29    let mut fields = rest.split(' ');
30    let _status: u16 = fields.next()?.parse().ok()?;
31    let bytes = match fields.next()? {
32        "-" => 0,
33        bytes => bytes.parse().ok()?,
34    };
35    if address.parse::<std::net::IpAddr>().is_err() {
36        return None;
37    }
38    Some(Served {
39        at,
40        address: address.to_owned(),
41        bytes,
42    })
43}
44
45/// A runner's hold on an address.
46#[derive(Debug, Clone, PartialEq, Eq)]
47struct Lease {
48    address: String,
49    runner: String,
50    from: u64,
51    until: Option<u64>,
52}
53
54/// Which runner held which address when.
55#[derive(Debug, Default)]
56pub struct Leases {
57    leases: Vec<Lease>,
58}
59
60impl Leases {
61    /// A runner's container got `address` at `from` (ms).
62    pub fn start(&mut self, address: &str, runner: &str, from: u64) {
63        self.leases.push(Lease {
64            address: address.to_owned(),
65            runner: runner.to_owned(),
66            from,
67            until: None,
68        });
69    }
70
71    /// A runner's container was removed at `until` (ms).
72    pub fn end(&mut self, runner: &str, until: u64) {
73        for lease in &mut self.leases {
74            if lease.runner == runner && lease.until.is_none() {
75                lease.until = Some(until);
76            }
77        }
78    }
79
80    /// Forgets leases that ended before `before` (ms): the log was read past them.
81    pub fn forget_before(&mut self, before: u64) {
82        self.leases
83            .retain(|lease| lease.until.is_none_or(|until| until >= before));
84    }
85
86    /// The runner that held the address at `at` (ms).
87    pub fn holder(&self, address: &str, at: u64) -> Option<&str> {
88        self.leases
89            .iter()
90            .find(|lease| {
91                lease.address == address
92                    && lease.from <= at
93                    && lease.until.is_none_or(|until| at < until)
94            })
95            .map(|lease| lease.runner.as_str())
96    }
97
98    /// Bytes per runner in log lines after `after`, and where the lines read end.
99    pub fn attribute(
100        &self,
101        log: &str,
102        after: MirrorPosition,
103    ) -> (Vec<(String, u64)>, MirrorPosition) {
104        let mut totals: Vec<(String, u64)> = Vec::new();
105        let mut until = after;
106        // Lines logged at `after.at` in this read; the first `after.seen` were counted before.
107        let mut at_after = 0;
108        for served in log.lines().filter_map(parse) {
109            if served.at < after.at {
110                continue;
111            }
112            if served.at == after.at {
113                at_after += 1;
114                if at_after <= after.seen {
115                    continue;
116                }
117            }
118            if served.at == until.at {
119                until.seen += 1;
120            } else {
121                until = MirrorPosition {
122                    at: served.at,
123                    seen: 1,
124                };
125            }
126            let Some(runner) = self.holder(&served.address, served.at / 1_000_000) else {
127                continue;
128            };
129            match totals.iter_mut().find(|(name, _)| name == runner) {
130                Some((_, total)) => *total += served.bytes,
131                None => totals.push((runner.to_owned(), served.bytes)),
132            }
133        }
134        (totals, until)
135    }
136}
runnerd_core/observe/mirror.rs

runnerd_core/observe/
mirror.rs
