Skip to main content

isb_core/
metrics.rs

1//! Live numbers for dashboards: the host's CPU, memory, storage, disk I/O
2//! and network, and every instance's status, address, CPU, memory, disk and
3//! network, each with a short history for sparklines, plus an hour of the
4//! host's own numbers for the monitor.
5//!
6//! One `GET /1.0/instances?recursion=2` per sample. CPU is a rate, so it needs
7//! two samples: the first one reports none. Instance CPU is a percentage of
8//! one core, as `docker stats` shows it (four busy cores read 400%).
9
10use std::collections::{BTreeMap, VecDeque};
11use std::time::{Duration, Instant};
12
13use serde::Serialize;
14use serde_json::Value;
15
16use crate::client::Client;
17use crate::error::Result;
18
19mod sys;
20
21/// This host's name.
22pub fn hostname() -> String {
23    sys::hostname()
24}
25
26/// Samples of history kept per series.
27pub const HISTORY: usize = 40;
28
29/// Disk counters come from `/1.0/metrics`, a heavier call: every this many
30/// samples.
31pub const DISK_EVERY: u32 = 5;
32
33/// Host points kept: an hour at the controller's two-second sample.
34pub const HOST_POINTS: usize = 1800;
35
36#[derive(Debug, Clone, Serialize, Default, PartialEq)]
37pub struct HostSample {
38    pub hostname: String,
39    pub cpus: u32,
40    /// Busy share of all CPUs, 0-100.
41    pub cpu_pct: Option<f32>,
42    pub cpu_history: Vec<f32>,
43    pub mem_used: u64,
44    pub mem_total: u64,
45    /// incus storage pools, used and total bytes. Pools on one filesystem
46    /// (several `dir` pools, say) are counted once.
47    pub disk_used: u64,
48    pub disk_total: u64,
49    pub load1: f32,
50    pub load5: Option<f32>,
51    pub load15: Option<f32>,
52    /// Busy share of each CPU, 0-100 (Linux only).
53    pub cpu_cores: Vec<f32>,
54    pub uptime_secs: Option<u64>,
55    pub swap_used: Option<u64>,
56    pub swap_total: Option<u64>,
57    /// Each storage pool, with its own used and total bytes.
58    pub pools: Vec<PoolSample>,
59    /// Bytes per second over the host's whole disks (not partitions, device
60    /// mapper or zvols, which would count the same bytes twice).
61    pub disk_read_rate: Option<f64>,
62    pub disk_write_rate: Option<f64>,
63    /// Bytes per second over [`HostSample::interfaces`].
64    pub net_rx_rate: Option<f64>,
65    pub net_tx_rate: Option<f64>,
66    /// Network interfaces but loopback and instances' own ends (veth, tap).
67    pub interfaces: Vec<IfaceSample>,
68}
69
70#[derive(Debug, Clone, Serialize, Default, PartialEq)]
71pub struct PoolSample {
72    pub name: String,
73    pub driver: String,
74    pub used: u64,
75    pub total: u64,
76}
77
78#[derive(Debug, Clone, Serialize, Default, PartialEq)]
79pub struct IfaceSample {
80    pub name: String,
81    /// Global addresses, IPv4 and IPv6.
82    pub addresses: Vec<String>,
83    pub up: bool,
84    pub rx_bytes: u64,
85    pub tx_bytes: u64,
86    pub rx_rate: Option<f64>,
87    pub tx_rate: Option<f64>,
88}
89
90/// One host sample, kept [`HOST_POINTS`] deep for the monitor's charts.
91#[derive(Debug, Clone, Copy, Serialize, Default, PartialEq)]
92pub struct HostPoint {
93    /// Unix seconds.
94    pub t: u64,
95    pub cpu: Option<f32>,
96    pub mem_used: Option<u64>,
97    pub net_rx: Option<f64>,
98    pub net_tx: Option<f64>,
99}
100
101#[derive(Debug, Clone, Serialize, Default, PartialEq)]
102pub struct InstanceSample {
103    pub name: String,
104    pub status: String,
105    /// `container`, `virtual-machine`, or `oci` for an application container.
106    pub kind: String,
107    pub ip: Option<String>,
108    /// Percent of one core.
109    pub cpu_pct: Option<f32>,
110    pub cpu_history: Vec<f32>,
111    pub mem_bytes: Option<u64>,
112    /// The root disk's usage, where the storage driver reports it (ZFS,
113    /// Btrfs, LVM; not `dir`).
114    pub disk_bytes: Option<u64>,
115    /// `user.*` config keys without the prefix, isb's own included.
116    pub labels: BTreeMap<String, String>,
117    pub image: String,
118    pub created_at: String,
119    /// The incus project; with isb's orgs, `isb-<org>` (or `default`).
120    pub project: String,
121    /// Bytes received and sent on every interface but loopback, since the
122    /// instance started (counters: the history turns them into rates).
123    #[serde(skip_serializing_if = "Option::is_none")]
124    pub net_rx_bytes: Option<u64>,
125    #[serde(skip_serializing_if = "Option::is_none")]
126    pub net_tx_bytes: Option<u64>,
127    /// Bytes read from and written to disks (counters), from incus'
128    /// `/1.0/metrics`, taken every [`DISK_EVERY`]th sample only.
129    #[serde(skip_serializing_if = "Option::is_none")]
130    pub disk_read_bytes: Option<u64>,
131    #[serde(skip_serializing_if = "Option::is_none")]
132    pub disk_write_bytes: Option<u64>,
133    /// The counters above as bytes per second, from the previous sample.
134    #[serde(skip_serializing_if = "Option::is_none")]
135    pub net_rx_rate: Option<f64>,
136    #[serde(skip_serializing_if = "Option::is_none")]
137    pub net_tx_rate: Option<f64>,
138    #[serde(skip_serializing_if = "Option::is_none")]
139    pub disk_read_rate: Option<f64>,
140    #[serde(skip_serializing_if = "Option::is_none")]
141    pub disk_write_rate: Option<f64>,
142}
143
144impl InstanceSample {
145    pub fn running(&self) -> bool {
146        self.status.eq_ignore_ascii_case("running")
147    }
148    /// The stack this instance is a replica of.
149    pub fn stack(&self) -> Option<&str> {
150        self.labels.get("isb.stack").map(String::as_str)
151    }
152}
153
154/// Turns successive snapshots into rates and histories.
155#[derive(Debug, Default)]
156pub struct Sampler {
157    cpu: BTreeMap<String, (u64, Instant)>,
158    hist: BTreeMap<String, VecDeque<f32>>,
159    host_cpu: Option<(u64, u64)>,
160    host_hist: VecDeque<f32>,
161    /// Pool usage moves slowly and costs a request per pool: refreshed every
162    /// [`POOLS_EVERY`].
163    pools: Option<(Instant, Vec<PoolSample>)>,
164    /// Samples taken, for [`DISK_EVERY`].
165    n: u32,
166    /// Per-core (busy, total) ticks.
167    host_cores: Vec<(u64, u64)>,
168    /// Per interface (rx, tx) bytes, and when they were read.
169    host_net: Option<(Instant, IfaceCounters)>,
170    host_disk: Option<(Instant, (u64, u64))>,
171    /// Per instance key: the last counters, for rates.
172    io: BTreeMap<String, Io>,
173    points: VecDeque<HostPoint>,
174}
175
176/// Per interface: (rx, tx) bytes.
177type IfaceCounters = BTreeMap<String, (u64, u64)>;
178
179#[derive(Debug, Default)]
180struct Io {
181    net: Option<(Instant, u64, u64)>,
182    disk: Option<(Instant, u64, u64)>,
183    /// Disk counters are read every [`DISK_EVERY`]th sample: the rate holds
184    /// between reads.
185    disk_rate: (Option<f64>, Option<f64>),
186}
187
188/// Bytes per second between two counter readings; none across a reset.
189fn per_sec(
190    prev: Option<(Instant, u64, u64)>,
191    now: Instant,
192    a: u64,
193    b: u64,
194) -> (Option<f64>, Option<f64>) {
195    match prev {
196        Some((at, pa, pb)) if a >= pa && b >= pb => {
197            let dt = now.duration_since(at).as_secs_f64();
198            if dt > 0.0 {
199                (Some((a - pa) as f64 / dt), Some((b - pb) as f64 / dt))
200            } else {
201                (None, None)
202            }
203        }
204        _ => (None, None),
205    }
206}
207
208/// How often storage pool usage is re-read.
209const POOLS_EVERY: Duration = Duration::from_secs(30);
210
211fn push(h: &mut VecDeque<f32>, v: f32) {
212    if h.len() == HISTORY {
213        h.pop_front();
214    }
215    h.push_back(v);
216}
217
218impl Sampler {
219    pub fn new() -> Self {
220        Self::default()
221    }
222
223    /// Take one sample of the host and of every instance in the client's
224    /// project.
225    pub fn sample(&mut self, client: &Client) -> Result<(HostSample, Vec<InstanceSample>)> {
226        // Every project, so one sample covers every org.
227        let v = client.get("/1.0/instances?recursion=2&all-projects=true")?;
228        let now = Instant::now();
229        let mut out = Vec::new();
230        for i in v.as_array().into_iter().flatten() {
231            out.push(self.instance(i, now));
232        }
233        if self.n % DISK_EVERY == 0 {
234            // Disk I/O is optional: the rest of the sample stands without it.
235            if let Ok(text) = client.get_raw("/1.0/metrics") {
236                let io = parse_disk_metrics(&String::from_utf8_lossy(&text));
237                for i in out.iter_mut().filter(|i| i.running()) {
238                    let (r, w) = io
239                        .get(&(i.project.clone(), i.name.clone()))
240                        .copied()
241                        .unwrap_or((0, 0));
242                    i.disk_read_bytes = Some(r);
243                    i.disk_write_bytes = Some(w);
244                    let e = self
245                        .io
246                        .entry(format!("{}/{}", i.project, i.name))
247                        .or_default();
248                    e.disk_rate = per_sec(e.disk, now, r, w);
249                    e.disk = Some((now, r, w));
250                }
251            }
252        }
253        for i in out.iter_mut().filter(|i| i.running()) {
254            if let Some(e) = self.io.get(&format!("{}/{}", i.project, i.name)) {
255                (i.disk_read_rate, i.disk_write_rate) = e.disk_rate;
256            }
257        }
258        self.n = self.n.wrapping_add(1);
259        out.sort_by(|a, b| (&a.project, &a.name).cmp(&(&b.project, &b.name)));
260        let keys: std::collections::BTreeSet<String> = out
261            .iter()
262            .map(|i| format!("{}/{}", i.project, i.name))
263            .collect();
264        self.cpu.retain(|k, _| keys.contains(k));
265        self.hist.retain(|k, _| keys.contains(k));
266        self.io.retain(|k, _| keys.contains(k));
267        if self
268            .pools
269            .as_ref()
270            .is_none_or(|(at, _)| now.duration_since(*at) >= POOLS_EVERY)
271        {
272            // Storage is a nicety: a pool that cannot be read leaves the
273            // last numbers in place rather than failing the sample.
274            if let Ok(p) = pools(client) {
275                self.pools = Some((now, p));
276            }
277        }
278        Ok((self.host(now), out))
279    }
280
281    /// The host's last hour, oldest first.
282    pub fn host_points(&self) -> Vec<HostPoint> {
283        self.points.iter().copied().collect()
284    }
285
286    fn instance(&mut self, i: &Value, now: Instant) -> InstanceSample {
287        let project = i["project"].as_str().unwrap_or("default").to_string();
288        // Names are unique per project only: key by both.
289        let name = i["name"].as_str().unwrap_or_default().to_string();
290        let key = format!("{project}/{name}");
291        let config = i["config"].as_object();
292        let cfg = |k: &str| {
293            config
294                .and_then(|c| c.get(k))
295                .and_then(Value::as_str)
296                .unwrap_or_default()
297                .to_string()
298        };
299        let labels = config
300            .map(|c| {
301                c.iter()
302                    .filter_map(|(k, v)| {
303                        k.strip_prefix("user.")
304                            .filter(|k| *k != "isb.create-token")
305                            .map(|k| (k.to_string(), v.as_str().unwrap_or_default().to_string()))
306                    })
307                    .collect()
308            })
309            .unwrap_or_default();
310        let kind = if cfg("volatile.container.oci") == "true" {
311            "oci".to_string()
312        } else {
313            i["type"].as_str().unwrap_or_default().to_string()
314        };
315        let status = i["status"].as_str().unwrap_or_default().to_string();
316        let state = &i["state"];
317        let running = status.eq_ignore_ascii_case("running");
318        let usage = state["cpu"]["usage"].as_u64().filter(|_| running);
319        let cpu_pct = match (usage, self.cpu.get(&key)) {
320            (Some(u), Some((prev, at))) if u >= *prev => {
321                let wall = now.duration_since(*at).as_nanos() as f64;
322                (wall > 0.0).then(|| ((u - prev) as f64 / wall * 100.0) as f32)
323            }
324            _ => None,
325        };
326        match usage {
327            Some(u) => {
328                self.cpu.insert(key.clone(), (u, now));
329            }
330            None => {
331                self.cpu.remove(&key);
332            }
333        }
334        let h = self.hist.entry(key.clone()).or_default();
335        if running {
336            push(h, cpu_pct.unwrap_or(0.0));
337        } else {
338            h.clear();
339        }
340        let image = {
341            let d = cfg("image.description");
342            if d.is_empty() { cfg("image.id") } else { d }
343        };
344        let (rx, tx) = net_counters(state);
345        let (net_rx_rate, net_tx_rate) = match (rx, tx) {
346            (Some(r), Some(t)) if running => {
347                let e = self.io.entry(key.clone()).or_default();
348                let v = per_sec(e.net, now, r, t);
349                e.net = Some((now, r, t));
350                v
351            }
352            _ => {
353                self.io.remove(&key);
354                (None, None)
355            }
356        };
357        InstanceSample {
358            net_rx_bytes: rx.filter(|_| running),
359            net_tx_bytes: tx.filter(|_| running),
360            disk_read_bytes: None,
361            disk_write_bytes: None,
362            net_rx_rate,
363            net_tx_rate,
364            disk_read_rate: None,
365            disk_write_rate: None,
366            ip: running.then(|| first_ip(state)).flatten(),
367            cpu_pct,
368            cpu_history: h.iter().copied().collect(),
369            mem_bytes: state["memory"]["usage"].as_u64().filter(|_| running),
370            // -1 (or 0) when the driver cannot tell.
371            disk_bytes: state["disk"]["root"]["usage"]
372                .as_i64()
373                .filter(|u| *u > 0)
374                .map(|u| u as u64),
375            name,
376            status,
377            kind,
378            labels,
379            image,
380            created_at: i["created_at"].as_str().unwrap_or_default().to_string(),
381            project,
382        }
383    }
384
385    fn host(&mut self, now: Instant) -> HostSample {
386        let mut h = HostSample {
387            hostname: sys::hostname(),
388            cpus: std::thread::available_parallelism()
389                .map(|n| n.get() as u32)
390                .unwrap_or(1),
391            ..Default::default()
392        };
393        if let Some((busy, total)) = sys::cpu_ticks() {
394            if let Some((pb, pt)) = self.host_cpu {
395                if total > pt {
396                    let pct = (busy.saturating_sub(pb)) as f32 / (total - pt) as f32 * 100.0;
397                    h.cpu_pct = Some(pct);
398                    push(&mut self.host_hist, pct);
399                }
400            }
401            self.host_cpu = Some((busy, total));
402        }
403        h.cpu_history = self.host_hist.iter().copied().collect();
404        let cores = sys::cpu_cores();
405        if cores.len() == self.host_cores.len() {
406            h.cpu_cores = cores
407                .iter()
408                .zip(&self.host_cores)
409                .map(|((b, t), (pb, pt))| {
410                    if t > pt {
411                        b.saturating_sub(*pb) as f32 / (t - pt) as f32 * 100.0
412                    } else {
413                        0.0
414                    }
415                })
416                .collect();
417        }
418        self.host_cores = cores;
419        if let Some((total, avail)) = sys::memory() {
420            h.mem_total = total;
421            h.mem_used = total.saturating_sub(avail);
422        }
423        if let Some((total, free)) = sys::swap() {
424            h.swap_total = Some(total);
425            h.swap_used = Some(total.saturating_sub(free));
426        }
427        if let Some((_, pools)) = &self.pools {
428            (h.disk_used, h.disk_total) =
429                sum_pools(pools.iter().map(|p| (p.used, p.total)).collect());
430            h.pools = pools.clone();
431        }
432        if let Some([l1, l5, l15]) = sys::loadavg() {
433            h.load1 = l1;
434            h.load5 = Some(l5);
435            h.load15 = Some(l15);
436        }
437        h.uptime_secs = sys::uptime();
438        if let Some((r, w)) = sys::disk_io() {
439            let prev = self.host_disk.map(|(at, (a, b))| (at, a, b));
440            (h.disk_read_rate, h.disk_write_rate) = per_sec(prev, now, r, w);
441            self.host_disk = Some((now, (r, w)));
442        }
443        self.interfaces(&mut h, now);
444        push_point(
445            &mut self.points,
446            HostPoint {
447                t: unix_secs(),
448                cpu: h.cpu_pct,
449                mem_used: (h.mem_total > 0).then_some(h.mem_used),
450                net_rx: h.net_rx_rate,
451                net_tx: h.net_tx_rate,
452            },
453        );
454        h
455    }
456
457    fn interfaces(&mut self, h: &mut HostSample, now: Instant) {
458        let counters: BTreeMap<String, (u64, u64)> = sys::net_dev()
459            .into_iter()
460            .filter(|(name, _, _)| shown_iface(name))
461            .map(|(name, rx, tx)| (name, (rx, tx)))
462            .collect();
463        let addrs = sys::addresses();
464        let (mut rx_sum, mut tx_sum, mut rated) = (0.0, 0.0, false);
465        for (name, (rx, tx)) in &counters {
466            let prev = self
467                .host_net
468                .as_ref()
469                .and_then(|(at, m)| m.get(name).map(|(a, b)| (*at, *a, *b)));
470            let (rx_rate, tx_rate) = per_sec(prev, now, *rx, *tx);
471            if let (Some(r), Some(t)) = (rx_rate, tx_rate) {
472                rx_sum += r;
473                tx_sum += t;
474                rated = true;
475            }
476            h.interfaces.push(IfaceSample {
477                name: name.clone(),
478                addresses: addrs.get(name).cloned().unwrap_or_default(),
479                up: sys::iface_up(name),
480                rx_bytes: *rx,
481                tx_bytes: *tx,
482                rx_rate,
483                tx_rate,
484            });
485        }
486        if rated {
487            h.net_rx_rate = Some(rx_sum);
488            h.net_tx_rate = Some(tx_sum);
489        }
490        self.host_net = Some((now, counters));
491    }
492}
493
494fn push_point(points: &mut VecDeque<HostPoint>, p: HostPoint) {
495    if points.len() == HOST_POINTS {
496        points.pop_front();
497    }
498    points.push_back(p);
499}
500
501fn unix_secs() -> u64 {
502    std::time::SystemTime::now()
503        .duration_since(std::time::UNIX_EPOCH)
504        .map(|d| d.as_secs())
505        .unwrap_or(0)
506}
507
508/// Interfaces the host view lists: loopback and the host ends of instances'
509/// NICs (one per instance, which would drown the rest) are left out.
510fn shown_iface(name: &str) -> bool {
511    name != "lo"
512        && !name.starts_with("veth")
513        && !name.starts_with("tap")
514        && !name.starts_with("macvtap")
515}
516
517/// Points within the last `range` seconds of `now`, averaged into buckets so
518/// at most `max` come back. Returns the bucket step in seconds.
519pub fn downsample(points: &[HostPoint], range: u64, now: u64, max: usize) -> (u64, Vec<HostPoint>) {
520    let max = max.max(1) as u64;
521    let step = range.div_ceil(max).max(2);
522    let from = now.saturating_sub(range);
523    let mut out: Vec<HostPoint> = Vec::new();
524    let mut acc: Option<(u64, [f64; 4], [u32; 4])> = None;
525    let flush = |out: &mut Vec<HostPoint>, a: (u64, [f64; 4], [u32; 4])| {
526        let avg = |i: usize| (a.2[i] > 0).then(|| a.1[i] / f64::from(a.2[i]));
527        out.push(HostPoint {
528            t: a.0,
529            cpu: avg(0).map(|v| v as f32),
530            mem_used: avg(1).map(|v| v as u64),
531            net_rx: avg(2),
532            net_tx: avg(3),
533        });
534    };
535    for p in points.iter().filter(|p| p.t > from && p.t <= now) {
536        let bucket = p.t / step * step;
537        if let Some(a) = acc.filter(|a| a.0 != bucket) {
538            flush(&mut out, a);
539            acc = None;
540        }
541        let a = acc.get_or_insert((bucket, [0.0; 4], [0; 4]));
542        for (i, v) in [
543            p.cpu.map(f64::from),
544            p.mem_used.map(|m| m as f64),
545            p.net_rx,
546            p.net_tx,
547        ]
548        .into_iter()
549        .enumerate()
550        {
551            if let Some(v) = v {
552                a.1[i] += v;
553                a.2[i] += 1;
554            }
555        }
556    }
557    if let Some(a) = acc {
558        flush(&mut out, a);
559    }
560    (step, out)
561}
562
563/// Each storage pool's used and total bytes.
564fn pools(client: &Client) -> Result<Vec<PoolSample>> {
565    let list = client.get("/1.0/storage-pools?recursion=1")?;
566    let mut out = Vec::new();
567    for p in list.as_array().into_iter().flatten() {
568        let name = p["name"].as_str().unwrap_or_default();
569        let r = client.get(&format!("/1.0/storage-pools/{name}/resources"))?;
570        out.push(PoolSample {
571            name: name.to_string(),
572            driver: p["driver"].as_str().unwrap_or_default().to_string(),
573            used: r["space"]["used"].as_u64().unwrap_or(0),
574            total: r["space"]["total"].as_u64().unwrap_or(0),
575        });
576    }
577    Ok(out)
578}
579
580/// Pools that share a filesystem report the same total: count it once (with
581/// the larger used, as their reads are a moment apart).
582fn sum_pools(spaces: Vec<(u64, u64)>) -> (u64, u64) {
583    let mut by_total: BTreeMap<u64, u64> = BTreeMap::new();
584    for (used, total) in spaces.into_iter().filter(|(_, t)| *t > 0) {
585        let u = by_total.entry(total).or_default();
586        *u = (*u).max(used);
587    }
588    (by_total.values().sum(), by_total.keys().sum())
589}
590
591/// Bytes received and sent, summed over every interface but loopback.
592fn net_counters(state: &Value) -> (Option<u64>, Option<u64>) {
593    let Some(ifs) = state["network"].as_object() else {
594        return (None, None);
595    };
596    let (mut rx, mut tx, mut any) = (0u64, 0u64, false);
597    for (name, n) in ifs {
598        if name == "lo" {
599            continue;
600        }
601        let c = &n["counters"];
602        if let (Some(r), Some(t)) = (c["bytes_received"].as_u64(), c["bytes_sent"].as_u64()) {
603            rx = rx.saturating_add(r);
604            tx = tx.saturating_add(t);
605            any = true;
606        }
607    }
608    if any {
609        (Some(rx), Some(tx))
610    } else {
611        (None, None)
612    }
613}
614
615/// Per `(project, instance)`: disk bytes read and written, summed over its
616/// devices, from incus' OpenMetrics text.
617pub fn parse_disk_metrics(text: &str) -> BTreeMap<(String, String), (u64, u64)> {
618    let mut out: BTreeMap<(String, String), (u64, u64)> = BTreeMap::new();
619    for line in text.lines() {
620        let (read, rest) = if let Some(r) = line.strip_prefix("incus_disk_read_bytes_total{") {
621            (true, r)
622        } else if let Some(r) = line.strip_prefix("incus_disk_written_bytes_total{") {
623            (false, r)
624        } else {
625            continue;
626        };
627        let Some((labels, value)) = rest.rsplit_once('}') else {
628            continue;
629        };
630        let label = |k: &str| {
631            labels.split(',').find_map(|kv| {
632                let (a, b) = kv.split_once('=')?;
633                (a.trim() == k).then(|| b.trim().trim_matches('"').to_string())
634            })
635        };
636        let (Some(name), Some(project)) = (label("name"), label("project")) else {
637            continue;
638        };
639        let Ok(v) = value.trim().parse::<f64>() else {
640            continue;
641        };
642        let e = out.entry((project, name)).or_default();
643        let v = v.max(0.0) as u64;
644        if read {
645            e.0 = e.0.saturating_add(v);
646        } else {
647            e.1 = e.1.saturating_add(v);
648        }
649    }
650    out
651}
652
653/// First global address, IPv4 preferred, on any interface but loopback.
654/// The instance's address: `eth0`'s first (its NIC on the org's bridge),
655/// then any other interface's. A bridge made inside the instance, such as
656/// Docker's `docker0`, sorts before `eth0` but reaches nothing.
657fn first_ip(state: &Value) -> Option<String> {
658    let mut v6 = None;
659    let nets = state["network"].as_object()?;
660    let eth0 = nets.get_key_value("eth0");
661    for (ifname, n) in eth0
662        .into_iter()
663        .chain(nets.iter().filter(|(k, _)| *k != "eth0"))
664    {
665        if ifname == "lo" {
666            continue;
667        }
668        for a in n["addresses"].as_array().into_iter().flatten() {
669            if a["scope"] != "global" {
670                continue;
671            }
672            match a["family"].as_str() {
673                Some("inet") => return a["address"].as_str().map(String::from),
674                Some("inet6") if v6.is_none() => v6 = a["address"].as_str().map(String::from),
675                _ => {}
676            }
677        }
678    }
679    v6
680}
681
682#[cfg(test)]
683mod tests {
684    use super::*;
685
686    #[test]
687    fn the_address_is_eth0s_even_with_docker_inside() {
688        let addr = |a: &str| serde_json::json!({"addresses": [{"family": "inet", "scope": "global", "address": a}]});
689        let state = serde_json::json!({"network": {
690            "docker0": addr("172.17.0.1"),
691            "eth0": addr("10.81.189.151"),
692            "lo": addr("127.0.0.1"),
693        }});
694        assert_eq!(first_ip(&state).as_deref(), Some("10.81.189.151"));
695        let other = serde_json::json!({"network": {"enp5s0": addr("10.0.0.9")}});
696        assert_eq!(first_ip(&other).as_deref(), Some("10.0.0.9"));
697    }
698    use serde_json::json;
699
700    #[test]
701    fn shown_interfaces() {
702        assert!(shown_iface("eth0") && shown_iface("incusbr0") && shown_iface("tailscale0"));
703        assert!(!shown_iface("lo") && !shown_iface("veth1a2b") && !shown_iface("tap0"));
704    }
705
706    #[test]
707    fn host_points_downsample() {
708        let p = |t: u64, cpu: f32| HostPoint {
709            t,
710            cpu: Some(cpu),
711            mem_used: Some(100),
712            net_rx: None,
713            net_tx: Some(10.0),
714        };
715        let pts: Vec<HostPoint> = (0..60).map(|i| p(1000 + i * 2, i as f32)).collect();
716        // The last 60 s at most 10 points: 6 s buckets of three samples each.
717        let (step, out) = downsample(&pts, 60, 1118, 10);
718        assert_eq!(step, 6);
719        assert!(out.len() <= 11, "{}", out.len());
720        assert!(out.iter().all(|x| x.t > 1118 - 60 - step));
721        assert_eq!(out.last().unwrap().net_rx, None);
722        assert_eq!(out.last().unwrap().net_tx, Some(10.0));
723        // Short ranges keep the samples themselves.
724        let (step, out) = downsample(&pts, 10, 1118, 300);
725        assert_eq!(step, 2);
726        assert_eq!(out.len(), 5);
727        assert_eq!(out.last().unwrap().cpu, Some(59.0));
728    }
729
730    #[test]
731    fn instance_rates_from_counters() {
732        let mut s = Sampler::new();
733        let inst = |rx: u64| {
734            json!({"name": "a", "status": "Running", "type": "container",
735                   "state": {"network": {"eth0": {"counters": {"bytes_received": rx, "bytes_sent": 0}}}}})
736        };
737        let t0 = Instant::now();
738        assert_eq!(s.instance(&inst(1000), t0).net_rx_rate, None);
739        let b = s.instance(&inst(3000), t0 + Duration::from_secs(2));
740        assert_eq!((b.net_rx_rate, b.net_tx_rate), (Some(1000.0), Some(0.0)));
741        // A counter that went down (a restart) gives no rate.
742        let c = s.instance(&inst(10), t0 + Duration::from_secs(4));
743        assert_eq!(c.net_rx_rate, None);
744    }
745
746    #[test]
747    fn disk_metrics() {
748        let t = "# HELP x\n\
749                 incus_disk_read_bytes_total{device=\"vda\",name=\"web\",project=\"isb-acme\",type=\"container\"} 4096\n\
750                 incus_disk_read_bytes_total{device=\"vdb\",name=\"web\",project=\"isb-acme\",type=\"container\"} 1.5e+03\n\
751                 incus_disk_written_bytes_total{device=\"vda\",name=\"web\",project=\"isb-acme\",type=\"container\"} 10\n\
752                 incus_cpu_seconds_total{cpu=\"0\",mode=\"user\",name=\"web\",project=\"isb-acme\",type=\"container\"} 1\n";
753        let m = parse_disk_metrics(t);
754        assert_eq!(m[&("isb-acme".to_string(), "web".to_string())], (5596, 10));
755        assert_eq!(m.len(), 1);
756    }
757
758    #[test]
759    fn instance_rates_and_kinds() {
760        let mut s = Sampler::new();
761        let inst = |usage: u64| {
762            json!({
763                "name": "a", "status": "Running", "type": "container",
764                "config": {"volatile.container.oci": "true", "user.isb.stack": "app", "user.isb.create-token": "x"},
765                "state": {"cpu": {"usage": usage}, "memory": {"usage": 1024},
766                          "network": {"lo": {"addresses": [{"family": "inet", "address": "127.0.0.1", "scope": "local"}]},
767                                      "eth0": {"counters": {"bytes_received": 100, "bytes_sent": 50}, "addresses": [{"family": "inet6", "address": "fd42::1", "scope": "global"},
768                                                             {"family": "inet", "address": "10.0.0.2", "scope": "global"}]}}}
769            })
770        };
771        let t0 = Instant::now();
772        let a = s.instance(&inst(1_000_000_000), t0);
773        assert_eq!(a.cpu_pct, None);
774        assert_eq!(a.kind, "oci");
775        assert_eq!(a.ip.as_deref(), Some("10.0.0.2"));
776        assert_eq!(a.stack(), Some("app"));
777        assert_eq!((a.net_rx_bytes, a.net_tx_bytes), (Some(100), Some(50)));
778        assert!(!a.labels.contains_key("isb.create-token"));
779        let b = s.instance(&inst(1_500_000_000), t0 + std::time::Duration::from_secs(1));
780        assert!((b.cpu_pct.unwrap() - 50.0).abs() < 0.1, "{b:?}");
781        assert_eq!(b.cpu_history.len(), 2);
782        assert_eq!(b.disk_bytes, None, "no disk state: unknown");
783    }
784
785    #[test]
786    fn disk_usage() {
787        let mut s = Sampler::new();
788        let inst = |usage: i64| {
789            json!({"name": "a", "status": "Stopped", "type": "container",
790                   "state": {"disk": {"root": {"usage": usage, "total": 0}}}})
791        };
792        // Reported for stopped instances too; `dir` pools say -1.
793        assert_eq!(
794            s.instance(&inst(5 << 20), Instant::now()).disk_bytes,
795            Some(5 << 20)
796        );
797        assert_eq!(s.instance(&inst(-1), Instant::now()).disk_bytes, None);
798        // Two `dir` pools on one filesystem count once; a second disk adds.
799        assert_eq!(sum_pools(vec![(60, 100), (61, 100)]), (61, 100));
800        assert_eq!(sum_pools(vec![(60, 100), (5, 50), (0, 0)]), (65, 150));
801    }
802}