monitord/
lib.rs

1//! # monitord Crate
2//!
3//! `monitord` is a library to gather statistics about systemd.
4
5use std::sync::Arc;
6
7use std::collections::HashMap;
8use std::time::Duration;
9use std::time::Instant;
10
11use thiserror::Error;
12use tokio::sync::RwLock;
13use tracing::error;
14use tracing::info;
15use tracing::warn;
16
17#[derive(Error, Debug)]
18pub enum MonitordError {
19    #[error("D-Bus connection error: {0}")]
20    ZbusError(#[from] zbus::Error),
21}
22
23pub mod boot;
24pub mod config;
25pub(crate) mod dbus;
26pub mod dbus_stats;
27pub mod json;
28pub mod logging;
29pub mod machines;
30pub mod networkd;
31pub mod pid1;
32pub mod system;
33pub mod timer;
34pub mod unit_constants;
35pub mod units;
36pub mod varlink;
37pub mod varlink_networkd;
38pub mod varlink_units;
39pub mod verify;
40
41pub const DEFAULT_DBUS_ADDRESS: &str = "unix:path=/run/dbus/system_bus_socket";
42
43/// Stats collected for a single systemd-nspawn container or VM managed by systemd-machined
44#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default, PartialEq)]
45pub struct MachineStats {
46    /// systemd-networkd interface states inside the container
47    pub networkd: networkd::NetworkdState,
48    /// PID 1 process stats from procfs (using the container's leader PID)
49    pub pid1: Option<pid1::Pid1Stats>,
50    /// Overall systemd system state (e.g. running, degraded) inside the container
51    pub system_state: system::SystemdSystemState,
52    /// Aggregated systemd unit counts and per-service/timer stats inside the container
53    pub units: units::SystemdUnitStats,
54    /// systemd version running inside the container
55    pub version: system::SystemdVersion,
56    /// D-Bus daemon/broker statistics inside the container
57    pub dbus_stats: Option<dbus_stats::DBusStats>,
58    /// Boot blame statistics: slowest units at boot with activation times in seconds
59    #[serde(skip_serializing_if = "Option::is_none")]
60    pub boot_blame: Option<boot::BootBlameStats>,
61    /// Unit verification error statistics
62    pub verify_stats: Option<verify::VerifyStats>,
63}
64
65/// Root struct containing all enabled monitord metrics for the host system and containers
66#[derive(serde::Serialize, serde::Deserialize, Debug, Default, PartialEq)]
67pub struct MonitordStats {
68    /// systemd-networkd interface states and managed interface count
69    pub networkd: networkd::NetworkdState,
70    /// PID 1 (systemd) process stats from procfs: CPU, memory, FDs, tasks
71    pub pid1: Option<pid1::Pid1Stats>,
72    /// Overall systemd manager state (e.g. running, degraded, initializing)
73    pub system_state: system::SystemdSystemState,
74    /// Aggregated systemd unit counts by type/state and per-service/timer detailed metrics
75    pub units: units::SystemdUnitStats,
76    /// Installed systemd version (major.minor.revision.os)
77    pub version: system::SystemdVersion,
78    /// D-Bus daemon/broker statistics (connections, bus names, match rules, per-peer accounting)
79    pub dbus_stats: Option<dbus_stats::DBusStats>,
80    /// Per-container stats keyed by machine name, collected via systemd-machined
81    pub machines: HashMap<String, MachineStats>,
82    /// Boot blame statistics: slowest units at boot with activation times in seconds
83    #[serde(skip_serializing_if = "Option::is_none")]
84    pub boot_blame: Option<boot::BootBlameStats>,
85    /// Unit verification error statistics
86    pub verify_stats: Option<verify::VerifyStats>,
87    /// End-to-end duration of the last stat collection run in milliseconds.
88    pub stat_collection_run_time_ms: f64,
89}
90
91/// Print statistics in the format set in configuration
92pub fn print_stats(
93    key_prefix: &str,
94    output_format: &config::MonitordOutputFormat,
95    stats: &MonitordStats,
96) {
97    match output_format {
98        config::MonitordOutputFormat::Json => println!(
99            "{}",
100            serde_json::to_string(&stats).expect("Invalid JSON serialization")
101        ),
102        config::MonitordOutputFormat::JsonFlat => println!(
103            "{}",
104            json::flatten(stats, key_prefix).expect("Invalid JSON serialization")
105        ),
106        config::MonitordOutputFormat::JsonPretty => println!(
107            "{}",
108            serde_json::to_string_pretty(&stats).expect("Invalid JSON serialization")
109        ),
110    }
111}
112
113fn set_stat_collection_run_time(stats: &mut MonitordStats, elapsed_runtime: Duration) {
114    stats.stat_collection_run_time_ms = elapsed_runtime.as_secs_f64() * 1000.0;
115}
116
117/// Reuse an existing D-Bus connection or create a new system bus connection.
118async fn get_or_create_dbus_connection(
119    config: &config::Config,
120    maybe_connection: Option<zbus::Connection>,
121) -> Result<zbus::Connection, MonitordError> {
122    match maybe_connection {
123        Some(conn) => Ok(conn),
124        None => Ok(zbus::connection::Builder::system()?
125            .method_timeout(std::time::Duration::from_secs(config.monitord.dbus_timeout))
126            .build()
127            .await?),
128    }
129}
130
131/// Main statistic collection function running what's required by configuration in parallel
132/// Takes an optional locked stats struct to update and to output stats to STDOUT or not.
133/// Takes an optional D-Bus connection. Returns `Some(connection)` if the
134/// collection cycle completed without errors (meaning the connection is reusable),
135/// `None` if errors occurred.
136pub async fn stat_collector(
137    config: config::Config,
138    maybe_locked_stats: Option<Arc<RwLock<MonitordStats>>>,
139    output_stats: bool,
140    maybe_connection: Option<zbus::Connection>,
141) -> Result<Option<zbus::Connection>, MonitordError> {
142    let mut collect_interval_ms: u128 = 0;
143    if config.monitord.daemon {
144        collect_interval_ms = (config.monitord.daemon_stats_refresh_secs * 1000).into();
145    }
146
147    let config = Arc::new(config);
148    let locked_monitord_stats: Arc<RwLock<MonitordStats>> =
149        maybe_locked_stats.unwrap_or(Arc::new(RwLock::new(MonitordStats::default())));
150    let locked_machine_stats: Arc<RwLock<MachineStats>> =
151        Arc::new(RwLock::new(MachineStats::default()));
152    let cached_machine_connections: Arc<tokio::sync::Mutex<machines::MachineConnections>> =
153        Arc::new(tokio::sync::Mutex::new(HashMap::new()));
154    std::env::set_var("DBUS_SYSTEM_BUS_ADDRESS", &config.monitord.dbus_address);
155    let sdc = get_or_create_dbus_connection(&config, maybe_connection).await?;
156    let mut join_set = tokio::task::JoinSet::new();
157    let mut had_error;
158
159    loop {
160        let collect_start_time = Instant::now();
161        info!("Starting stat collection run");
162
163        // Always collect systemd version
164
165        join_set.spawn(crate::system::update_version(
166            sdc.clone(),
167            locked_machine_stats.clone(),
168        ));
169
170        // Collect pid1 procfs stats
171        if config.pid1.enabled {
172            join_set.spawn(crate::pid1::update_pid1_stats(
173                1,
174                locked_machine_stats.clone(),
175            ));
176        }
177
178        // Run networkd collector if enabled
179        if config.networkd.enabled {
180            let config_clone = Arc::clone(&config);
181            let sdc_clone = sdc.clone();
182            let stats_clone = locked_machine_stats.clone();
183            join_set.spawn(async move {
184                if config_clone.varlink.enabled {
185                    let socket_path = crate::varlink_networkd::NETWORK_SOCKET_PATH.to_string();
186                    match crate::varlink_networkd::get_networkd_state(&socket_path).await {
187                        Ok(networkd_stats) => {
188                            let mut machine_stats = stats_clone.write().await;
189                            machine_stats.networkd = networkd_stats;
190                            return Ok(());
191                        }
192                        Err(err) => {
193                            warn!(
194                                "Varlink networkd stats failed, falling back to file-based: {:?}",
195                                err
196                            );
197                        }
198                    }
199                }
200                crate::networkd::update_networkd_stats(
201                    config_clone.networkd.link_state_dir.clone(),
202                    None,
203                    sdc_clone,
204                    stats_clone,
205                )
206                .await
207            });
208        }
209
210        // Run system running (SystemState) state collector
211        if config.system_state.enabled {
212            join_set.spawn(crate::system::update_system_stats(
213                sdc.clone(),
214                locked_machine_stats.clone(),
215            ));
216        }
217
218        // Run service collectors if there are services listed in config
219        if config.units.enabled {
220            let config_clone = Arc::clone(&config);
221            let sdc_clone = sdc.clone();
222            let stats_clone = locked_machine_stats.clone();
223            join_set.spawn(async move {
224                if config_clone.varlink.enabled {
225                    let socket_path = crate::varlink_units::METRICS_SOCKET_PATH.to_string();
226                    match crate::varlink_units::update_unit_stats(
227                        Arc::clone(&config_clone),
228                        stats_clone.clone(),
229                        socket_path,
230                    )
231                    .await
232                    {
233                        Ok(()) => {
234                            // Timer properties are not yet exposed via varlink; collect via D-Bus.
235                            match crate::timer::collect_all_timers_dbus(&sdc_clone, &config_clone)
236                                .await
237                            {
238                                Ok(timer_stats) => {
239                                    let mut ms = stats_clone.write().await;
240                                    ms.units.timer_stats = timer_stats.timer_stats;
241                                    ms.units.timer_persistent_units =
242                                        timer_stats.timer_persistent_units;
243                                    ms.units.timer_remain_after_elapse =
244                                        timer_stats.timer_remain_after_elapse;
245                                }
246                                Err(err) => {
247                                    warn!("Varlink timer stats (D-Bus fallback) failed: {:?}", err);
248                                }
249                            }
250                            return Ok(());
251                        }
252                        Err(err) => {
253                            warn!(
254                                "Varlink units stats failed, falling back to D-Bus: {:?}",
255                                err
256                            );
257                        }
258                    }
259                }
260                crate::units::update_unit_stats(config_clone, sdc_clone, stats_clone).await
261            });
262        }
263
264        if config.machines.enabled {
265            join_set.spawn(crate::machines::update_machines_stats(
266                Arc::clone(&config),
267                sdc.clone(),
268                locked_monitord_stats.clone(),
269                cached_machine_connections.clone(),
270            ));
271        }
272
273        if config.dbus_stats.enabled {
274            join_set.spawn(crate::dbus_stats::update_dbus_stats(
275                Arc::clone(&config),
276                sdc.clone(),
277                locked_machine_stats.clone(),
278            ));
279        }
280
281        if config.boot_blame.enabled {
282            join_set.spawn(crate::boot::update_boot_blame_stats(
283                Arc::clone(&config),
284                sdc.clone(),
285                locked_machine_stats.clone(),
286            ));
287        }
288
289        if config.verify.enabled {
290            join_set.spawn(crate::verify::update_verify_stats(
291                sdc.clone(),
292                locked_machine_stats.clone(),
293                config.verify.allowlist.clone(),
294                config.verify.blocklist.clone(),
295            ));
296        }
297
298        if join_set.len() == 1 {
299            warn!("No collectors except systemd version scheduled to run. Exiting");
300        }
301
302        // Check all collection for errors and log if one fails
303        had_error = false;
304        while let Some(res) = join_set.join_next().await {
305            match res {
306                Ok(r) => match r {
307                    Ok(_) => (),
308                    Err(e) => {
309                        had_error = true;
310                        error!("Collection specific failure: {:?}", e);
311                    }
312                },
313                Err(e) => {
314                    had_error = true;
315                    error!("Join error: {:?}", e);
316                }
317            }
318        }
319
320        let elapsed_runtime = collect_start_time.elapsed();
321        let elapsed_runtime_ms = elapsed_runtime.as_millis();
322
323        {
324            // Update monitord stats with machine stats
325            let mut monitord_stats = locked_monitord_stats.write().await;
326            let machine_stats = locked_machine_stats.read().await;
327            monitord_stats.pid1 = machine_stats.pid1.clone();
328            monitord_stats.networkd = machine_stats.networkd.clone();
329            monitord_stats.system_state = machine_stats.system_state;
330            monitord_stats.version = machine_stats.version.clone();
331            monitord_stats.units = machine_stats.units.clone();
332            monitord_stats.dbus_stats = machine_stats.dbus_stats.clone();
333            monitord_stats.boot_blame = machine_stats.boot_blame.clone();
334            monitord_stats.verify_stats = machine_stats.verify_stats.clone();
335            set_stat_collection_run_time(&mut monitord_stats, elapsed_runtime);
336        }
337
338        info!("stat collection run took {}ms", elapsed_runtime_ms);
339        if output_stats {
340            let monitord_stats = locked_monitord_stats.read().await;
341            print_stats(
342                &config.monitord.key_prefix,
343                &config.monitord.output_format,
344                &monitord_stats,
345            );
346        }
347        if !config.monitord.daemon {
348            break;
349        }
350        let sleep_time_ms = collect_interval_ms - elapsed_runtime_ms;
351        info!("stat collection sleeping for {}s 😴", sleep_time_ms / 1000);
352        tokio::time::sleep(Duration::from_millis(
353            sleep_time_ms
354                .try_into()
355                .expect("Sleep time does not fit into a u64 :O"),
356        ))
357        .await;
358    }
359    Ok(if had_error { None } else { Some(sdc) })
360}
361
362#[cfg(test)]
363mod tests {
364    use super::*;
365
366    #[test]
367    fn test_stat_collection_run_time_ms_conversion() {
368        let mut stats = MonitordStats::default();
369        set_stat_collection_run_time(&mut stats, Duration::from_millis(5));
370        assert_eq!(stats.stat_collection_run_time_ms, 5.0);
371
372        set_stat_collection_run_time(&mut stats, Duration::from_micros(500));
373        assert!((stats.stat_collection_run_time_ms - 0.5).abs() < f64::EPSILON);
374    }
375}
monitord/lib.rs

monitord/
lib.rs
