monitord/
machines.rs

1use std::collections::HashMap;
2use std::collections::HashSet;
3use std::sync::Arc;
4
5use thiserror::Error;
6use tokio::sync::RwLock;
7use tracing::{debug, error};
8
9use crate::MachineStats;
10use crate::MonitordStats;
11
12#[derive(Error, Debug)]
13pub enum MonitordMachinesError {
14    #[error("Machines D-Bus error: {0}")]
15    ZbusError(#[from] zbus::Error),
16}
17
18pub fn filter_machines(
19    machines: Vec<crate::dbus::zbus_machines::ListedMachine>,
20    allowlist: &HashSet<String>,
21    blocklist: &HashSet<String>,
22) -> Vec<crate::dbus::zbus_machines::ListedMachine> {
23    machines
24        .into_iter()
25        .filter(|c| c.class == "container")
26        .filter(|c| !blocklist.contains(&c.name))
27        .filter(|c| allowlist.is_empty() || allowlist.contains(&c.name))
28        .collect()
29}
30
31pub async fn get_machines(
32    connection: &zbus::Connection,
33    config: &crate::config::Config,
34) -> Result<HashMap<String, u32>, MonitordMachinesError> {
35    let c = crate::dbus::zbus_machines::ManagerProxy::new(connection).await?;
36    let mut results = HashMap::<String, u32>::new();
37
38    let machines = c.list_machines().await?;
39
40    for machine in filter_machines(
41        machines,
42        &config.machines.allowlist,
43        &config.machines.blocklist,
44    ) {
45        let m = c.get_machine(&machine.name).await?;
46        let leader_pid = m.leader().await?;
47        results.insert(machine.name, leader_pid);
48    }
49
50    Ok(results)
51}
52
53pub async fn update_machines_stats(
54    config: Arc<crate::config::Config>,
55    connection: zbus::Connection,
56    locked_monitord_stats: Arc<RwLock<MonitordStats>>,
57) -> anyhow::Result<()> {
58    let locked_machine_stats: Arc<RwLock<MachineStats>> =
59        Arc::new(RwLock::new(MachineStats::default()));
60
61    for (machine, leader_pid) in get_machines(&connection, &config).await?.into_iter() {
62        debug!(
63            "Collecting container: machine: {} leader_pid: {}",
64            machine, leader_pid
65        );
66        let container_address = format!(
67            "unix:path=/proc/{}/root/run/dbus/system_bus_socket",
68            leader_pid
69        );
70        let sdc = zbus::connection::Builder::address(container_address.as_str())?
71            .method_timeout(std::time::Duration::from_secs(config.monitord.dbus_timeout))
72            .build()
73            .await?;
74        let mut join_set = tokio::task::JoinSet::new();
75
76        if config.pid1.enabled {
77            join_set.spawn(crate::pid1::update_pid1_stats(
78                leader_pid as i32,
79                locked_machine_stats.clone(),
80            ));
81        }
82
83        if config.networkd.enabled {
84            join_set.spawn(crate::networkd::update_networkd_stats(
85                config.networkd.link_state_dir.clone(),
86                None,
87                sdc.clone(),
88                locked_machine_stats.clone(),
89            ));
90        }
91
92        if config.system_state.enabled {
93            join_set.spawn(crate::system::update_system_stats(
94                sdc.clone(),
95                locked_machine_stats.clone(),
96            ));
97        }
98
99        join_set.spawn(crate::system::update_version(
100            sdc.clone(),
101            locked_machine_stats.clone(),
102        ));
103
104        if config.units.enabled {
105            join_set.spawn(crate::units::update_unit_stats(
106                Arc::clone(&config),
107                sdc.clone(),
108                locked_machine_stats.clone(),
109            ));
110        }
111
112        if config.dbus_stats.enabled {
113            join_set.spawn(crate::dbus_stats::update_dbus_stats(
114                Arc::clone(&config),
115                sdc.clone(),
116                locked_machine_stats.clone(),
117            ));
118        }
119
120        while let Some(res) = join_set.join_next().await {
121            match res {
122                Ok(r) => match r {
123                    Ok(_) => (),
124                    Err(e) => {
125                        error!(
126                            "Collection specific failure (container {}): {:?}",
127                            machine, e
128                        );
129                    }
130                },
131                Err(e) => {
132                    error!("Join error (container {}): {:?}", machine, e);
133                }
134            }
135        }
136
137        {
138            let mut monitord_stats = locked_monitord_stats.write().await;
139            let machine_stats = locked_machine_stats.read().await;
140            monitord_stats
141                .machines
142                .insert(machine, machine_stats.clone());
143        }
144    }
145
146    Ok(())
147}
148
149#[cfg(test)]
150mod tests {
151    use std::collections::HashSet;
152    use zbus::zvariant::OwnedObjectPath;
153
154    #[test]
155    fn test_filter_machines() {
156        let machines = vec![
157            crate::dbus::zbus_machines::ListedMachine {
158                name: "foo".to_string(),
159                class: "container".to_string(),
160                service: "".to_string(),
161                path: OwnedObjectPath::try_from("/sample/object").unwrap(),
162            },
163            crate::dbus::zbus_machines::ListedMachine {
164                name: "bar".to_string(),
165                class: "container".to_string(),
166                service: "".to_string(),
167                path: OwnedObjectPath::try_from("/sample/object").unwrap(),
168            },
169            crate::dbus::zbus_machines::ListedMachine {
170                name: "baz".to_string(),
171                class: "container".to_string(),
172                service: "".to_string(),
173                path: OwnedObjectPath::try_from("/sample/object").unwrap(),
174            },
175        ];
176        let allowlist = HashSet::from(["foo".to_string(), "baz".to_string()]);
177        let blocklist = HashSet::from(["bar".to_string()]);
178
179        let filtered = super::filter_machines(machines, &allowlist, &blocklist);
180
181        assert_eq!(filtered.len(), 2);
182        assert_eq!(filtered[0].name, "foo");
183        assert_eq!(filtered[1].name, "baz");
184    }
185}
monitord/machines.rs

monitord/
machines.rs
