1use std::collections::HashMap;
7use std::str::FromStr;
8use std::sync::Arc;
9use std::time::SystemTime;
10use std::time::UNIX_EPOCH;
11
12use int_enum::IntEnum;
13use serde_repr::*;
14use struct_field_names_as_array::FieldNamesAsArray;
15use strum_macros::EnumIter;
16use strum_macros::EnumString;
17use thiserror::Error;
18use tokio::sync::RwLock;
19use tracing::debug;
20use tracing::error;
21use zbus::zvariant::ObjectPath;
22use zbus::zvariant::OwnedObjectPath;
23
24#[derive(Error, Debug)]
25pub enum MonitordUnitsError {
26 #[error("Units D-Bus error: {0}")]
27 ZbusError(#[from] zbus::Error),
28 #[error("Integer conversion error: {0}")]
29 IntConversion(#[from] std::num::TryFromIntError),
30 #[error("System time error: {0}")]
31 SystemTimeError(#[from] std::time::SystemTimeError),
32}
33
34use crate::timer::TimerStats;
35use crate::MachineStats;
36
37#[derive(
38 serde::Serialize, serde::Deserialize, Clone, Debug, Default, Eq, FieldNamesAsArray, PartialEq,
39)]
40
41pub struct SystemdUnitStats {
44 pub activating_units: u64,
46 pub active_units: u64,
48 pub automount_units: u64,
50 pub device_units: u64,
52 pub failed_units: u64,
54 pub inactive_units: u64,
56 pub jobs_queued: u64,
58 pub loaded_units: u64,
60 pub masked_units: u64,
62 pub mount_units: u64,
64 pub not_found_units: u64,
66 pub path_units: u64,
68 pub scope_units: u64,
70 pub service_units: u64,
72 pub slice_units: u64,
74 pub socket_units: u64,
76 pub target_units: u64,
78 pub timer_units: u64,
80 pub timer_persistent_units: u64,
82 pub timer_remain_after_elapse: u64,
84 pub total_units: u64,
86 pub service_stats: HashMap<String, ServiceStats>,
88 pub timer_stats: HashMap<String, TimerStats>,
90 pub unit_states: HashMap<String, UnitStates>,
92}
93
94#[derive(
97 serde::Serialize, serde::Deserialize, Clone, Debug, Default, Eq, FieldNamesAsArray, PartialEq,
98)]
99pub struct ServiceStats {
100 pub active_enter_timestamp: u64,
102 pub active_exit_timestamp: u64,
104 pub cpuusage_nsec: u64,
106 pub inactive_exit_timestamp: u64,
108 pub ioread_bytes: u64,
110 pub ioread_operations: u64,
112 pub memory_available: u64,
114 pub memory_current: u64,
116 pub nrestarts: u32,
118 pub processes: u32,
120 pub restart_usec: u64,
122 pub state_change_timestamp: u64,
124 pub status_errno: i32,
126 pub tasks_current: u64,
128 pub timeout_clean_usec: u64,
130 pub watchdog_usec: u64,
132}
133
134#[derive(
137 serde::Serialize, serde::Deserialize, Clone, Debug, Default, Eq, FieldNamesAsArray, PartialEq,
138)]
139pub struct UnitStates {
140 pub active_state: SystemdUnitActiveState,
142 pub load_state: SystemdUnitLoadState,
144 pub unhealthy: bool,
147 pub time_in_state_usecs: Option<u64>,
150}
151
152#[allow(non_camel_case_types)]
159#[derive(
160 Serialize_repr,
161 Deserialize_repr,
162 Clone,
163 Copy,
164 Debug,
165 Default,
166 Eq,
167 PartialEq,
168 EnumIter,
169 EnumString,
170 IntEnum,
171 strum_macros::Display,
172)]
173#[repr(u8)]
174pub enum SystemdUnitActiveState {
175 #[default]
177 unknown = 0,
178 active = 1,
180 reloading = 2,
182 inactive = 3,
184 failed = 4,
186 activating = 5,
188 deactivating = 6,
190}
191
192#[allow(non_camel_case_types)]
195#[derive(
196 Serialize_repr,
197 Deserialize_repr,
198 Clone,
199 Copy,
200 Debug,
201 Default,
202 Eq,
203 PartialEq,
204 EnumIter,
205 EnumString,
206 IntEnum,
207 strum_macros::Display,
208)]
209#[repr(u8)]
210pub enum SystemdUnitLoadState {
211 #[default]
213 unknown = 0,
214 loaded = 1,
216 error = 2,
218 masked = 3,
220 not_found = 4,
222}
223
224#[derive(Debug)]
226pub struct ListedUnit {
227 pub name: String, pub description: String, pub load_state: String, pub active_state: String, pub sub_state: String, pub follow_unit: String, pub unit_object_path: OwnedObjectPath, pub job_id: u32, pub job_type: String, pub job_object_path: OwnedObjectPath, }
238impl
239 From<(
240 String,
241 String,
242 String,
243 String,
244 String,
245 String,
246 OwnedObjectPath,
247 u32,
248 String,
249 OwnedObjectPath,
250 )> for ListedUnit
251{
252 fn from(
253 tuple: (
254 String,
255 String,
256 String,
257 String,
258 String,
259 String,
260 OwnedObjectPath,
261 u32,
262 String,
263 OwnedObjectPath,
264 ),
265 ) -> Self {
266 ListedUnit {
267 name: tuple.0,
268 description: tuple.1,
269 load_state: tuple.2,
270 active_state: tuple.3,
271 sub_state: tuple.4,
272 follow_unit: tuple.5,
273 unit_object_path: tuple.6,
274 job_id: tuple.7,
275 job_type: tuple.8,
276 job_object_path: tuple.9,
277 }
278 }
279}
280
281pub const SERVICE_FIELD_NAMES: &[&str] = &ServiceStats::FIELD_NAMES_AS_ARRAY;
282pub const UNIT_FIELD_NAMES: &[&str] = &SystemdUnitStats::FIELD_NAMES_AS_ARRAY;
283pub const UNIT_STATES_FIELD_NAMES: &[&str] = &UnitStates::FIELD_NAMES_AS_ARRAY;
284
285async fn parse_service(
287 connection: &zbus::Connection,
288 name: &str,
289 object_path: &OwnedObjectPath,
290) -> Result<ServiceStats, MonitordUnitsError> {
291 debug!("Parsing service {} stats", name);
292
293 let sp = crate::dbus::zbus_service::ServiceProxy::builder(connection)
294 .path(object_path.clone())?
295 .build()
296 .await?;
297 let up = crate::dbus::zbus_unit::UnitProxy::builder(connection)
298 .path(object_path.clone())?
299 .build()
300 .await?;
301
302 let (
305 active_enter_timestamp,
306 active_exit_timestamp,
307 cpuusage_nsec,
308 inactive_exit_timestamp,
309 ioread_bytes,
310 ioread_operations,
311 memory_current,
312 memory_available,
313 nrestarts,
314 processes,
315 restart_usec,
316 state_change_timestamp,
317 status_errno,
318 tasks_current,
319 timeout_clean_usec,
320 watchdog_usec,
321 ) = tokio::join!(
322 up.active_enter_timestamp(),
323 up.active_exit_timestamp(),
324 sp.cpuusage_nsec(),
325 up.inactive_exit_timestamp(),
326 sp.ioread_bytes(),
327 sp.ioread_operations(),
328 sp.memory_current(),
329 sp.memory_available(),
330 sp.nrestarts(),
331 sp.get_processes(),
332 sp.restart_usec(),
333 up.state_change_timestamp(),
334 sp.status_errno(),
335 sp.tasks_current(),
336 sp.timeout_clean_usec(),
337 sp.watchdog_usec(),
338 );
339
340 Ok(ServiceStats {
341 active_enter_timestamp: active_enter_timestamp?,
342 active_exit_timestamp: active_exit_timestamp?,
343 cpuusage_nsec: cpuusage_nsec?,
344 inactive_exit_timestamp: inactive_exit_timestamp?,
345 ioread_bytes: ioread_bytes?,
346 ioread_operations: ioread_operations?,
347 memory_current: memory_current?,
348 memory_available: memory_available?,
349 nrestarts: nrestarts?,
350 processes: processes?.len().try_into()?,
351 restart_usec: restart_usec?,
352 state_change_timestamp: state_change_timestamp?,
353 status_errno: status_errno?,
354 tasks_current: tasks_current?,
355 timeout_clean_usec: timeout_clean_usec?,
356 watchdog_usec: watchdog_usec?,
357 })
358}
359
360pub fn is_unit_unhealthy(
364 active_state: SystemdUnitActiveState,
365 load_state: SystemdUnitLoadState,
366) -> bool {
367 match load_state {
368 SystemdUnitLoadState::loaded => !matches!(active_state, SystemdUnitActiveState::active),
370 SystemdUnitLoadState::masked => false,
373 _ => true,
375 }
376}
377
378async fn get_time_in_state(
379 connection: Option<&zbus::Connection>,
380 unit: &ListedUnit,
381) -> Result<Option<u64>, MonitordUnitsError> {
382 match connection {
383 Some(c) => {
384 let up = crate::dbus::zbus_unit::UnitProxy::builder(c)
385 .path(ObjectPath::from(unit.unit_object_path.clone()))?
386 .build()
387 .await?;
388 let now: u64 = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs() * 1_000_000;
389 let state_change_timestamp = match up.state_change_timestamp().await {
390 Ok(sct) => sct,
391 Err(err) => {
392 error!(
393 "Unable to get state_change_timestamp for {} - Setting to 0: {:?}",
394 &unit.name, err,
395 );
396 0
397 }
398 };
399 Ok(Some(now - state_change_timestamp))
400 }
401 None => {
402 error!("No zbus connection passed, but time_in_state_usecs enabled");
403 Ok(None)
404 }
405 }
406}
407
408pub async fn parse_state(
410 stats: &mut SystemdUnitStats,
411 unit: &ListedUnit,
412 config: &crate::config::UnitsConfig,
413 connection: Option<&zbus::Connection>,
414) -> Result<(), MonitordUnitsError> {
415 if config.state_stats_blocklist.contains(&unit.name) {
416 debug!("Skipping state stats for {} due to blocklist", &unit.name);
417 return Ok(());
418 }
419 if !config.state_stats_allowlist.is_empty()
420 && !config.state_stats_allowlist.contains(&unit.name)
421 {
422 return Ok(());
423 }
424 let active_state = SystemdUnitActiveState::from_str(&unit.active_state)
425 .unwrap_or(SystemdUnitActiveState::unknown);
426 let load_state = SystemdUnitLoadState::from_str(&unit.load_state.replace('-', "_"))
427 .unwrap_or(SystemdUnitLoadState::unknown);
428
429 let mut time_in_state_usecs: Option<u64> = None;
431 if config.state_stats_time_in_state {
432 time_in_state_usecs = get_time_in_state(connection, unit).await?;
433 }
434
435 stats.unit_states.insert(
436 unit.name.clone(),
437 UnitStates {
438 active_state,
439 load_state,
440 unhealthy: is_unit_unhealthy(active_state, load_state),
441 time_in_state_usecs,
442 },
443 );
444 Ok(())
445}
446
447fn parse_unit(stats: &mut SystemdUnitStats, unit: &ListedUnit) {
449 match unit.name.rsplit('.').next() {
451 Some("automount") => stats.automount_units += 1,
452 Some("device") => stats.device_units += 1,
453 Some("mount") => stats.mount_units += 1,
454 Some("path") => stats.path_units += 1,
455 Some("scope") => stats.scope_units += 1,
456 Some("service") => stats.service_units += 1,
457 Some("slice") => stats.slice_units += 1,
458 Some("socket") => stats.socket_units += 1,
459 Some("target") => stats.target_units += 1,
460 Some("timer") => stats.timer_units += 1,
461 unknown => debug!("Found unhandled '{:?}' unit type", unknown),
462 };
463 match unit.load_state.as_str() {
465 "loaded" => stats.loaded_units += 1,
466 "masked" => stats.masked_units += 1,
467 "not-found" => stats.not_found_units += 1,
468 _ => debug!("{} is not loaded. It's {}", unit.name, unit.load_state),
469 };
470 match unit.active_state.as_str() {
472 "activating" => stats.activating_units += 1,
473 "active" => stats.active_units += 1,
474 "failed" => stats.failed_units += 1,
475 "inactive" => stats.inactive_units += 1,
476 unknown => debug!("Found unhandled '{}' unit state", unknown),
477 };
478 if unit.job_id != 0 {
480 stats.jobs_queued += 1;
481 }
482}
483
484pub async fn parse_unit_state(
486 config: &crate::config::Config,
487 connection: &zbus::Connection,
488) -> Result<SystemdUnitStats, MonitordUnitsError> {
489 if !config.units.state_stats_allowlist.is_empty() {
490 debug!(
491 "Using unit state allowlist: {:?}",
492 config.units.state_stats_allowlist
493 );
494 }
495
496 if !config.units.state_stats_blocklist.is_empty() {
497 debug!(
498 "Using unit state blocklist: {:?}",
499 config.units.state_stats_blocklist,
500 );
501 }
502
503 let mut stats = SystemdUnitStats::default();
504 let p = crate::dbus::zbus_systemd::ManagerProxy::new(connection).await?;
505 let units = p.list_units().await?;
506
507 stats.total_units = units.len() as u64;
508 for unit_raw in units {
509 let unit: ListedUnit = unit_raw.into();
510 parse_unit(&mut stats, &unit);
512
513 if config.units.state_stats {
516 parse_state(&mut stats, &unit, &config.units, Some(connection)).await?;
517 }
518
519 if config.services.contains(&unit.name) {
521 debug!("Collecting service stats for {:?}", &unit);
522 match parse_service(connection, &unit.name, &unit.unit_object_path).await {
523 Ok(service_stats) => {
524 stats.service_stats.insert(unit.name.clone(), service_stats);
525 }
526 Err(err) => error!(
527 "Unable to get service stats for {} {}: {:#?}",
528 &unit.name, &unit.unit_object_path, err
529 ),
530 }
531 }
532
533 if config.timers.enabled && unit.name.contains(".timer") {
535 if config.timers.blocklist.contains(&unit.name) {
536 debug!("Skipping timer stats for {} due to blocklist", &unit.name);
537 continue;
538 }
539 if !config.timers.allowlist.is_empty() && !config.timers.allowlist.contains(&unit.name)
540 {
541 continue;
542 }
543 let timer_stats: Option<TimerStats> =
544 match crate::timer::collect_timer_stats(connection, &mut stats, &unit).await {
545 Ok(ts) => Some(ts),
546 Err(err) => {
547 error!("Failed to get {} stats: {:#?}", &unit.name, err);
548 None
549 }
550 };
551 if let Some(ts) = timer_stats {
552 stats.timer_stats.insert(unit.name.clone(), ts);
553 }
554 }
555 }
556 debug!("unit stats: {:?}", stats);
557 Ok(stats)
558}
559
560pub async fn update_unit_stats(
562 config: Arc<crate::config::Config>,
563 connection: zbus::Connection,
564 locked_machine_stats: Arc<RwLock<MachineStats>>,
565) -> anyhow::Result<()> {
566 let mut machine_stats = locked_machine_stats.write().await;
567 match parse_unit_state(&config, &connection).await {
568 Ok(units_stats) => machine_stats.units = units_stats,
569 Err(err) => error!("units stats failed: {:?}", err),
570 }
571 Ok(())
572}
573
574#[cfg(test)]
575mod tests {
576 use super::*;
577 use std::collections::HashSet;
578 use strum::IntoEnumIterator;
579
580 fn get_unit_file() -> ListedUnit {
581 ListedUnit {
582 name: String::from("apport-autoreport.timer"),
583 description: String::from(
584 "Process error reports when automatic reporting is enabled (timer based)",
585 ),
586 load_state: String::from("loaded"),
587 active_state: String::from("inactive"),
588 sub_state: String::from("dead"),
589 follow_unit: String::from(""),
590 unit_object_path: ObjectPath::try_from(
591 "/org/freedesktop/systemd1/unit/apport_2dautoreport_2etimer",
592 )
593 .expect("Unable to make an object path")
594 .into(),
595 job_id: 0,
596 job_type: String::from(""),
597 job_object_path: ObjectPath::try_from("/").unwrap().into(),
598 }
599 }
600
601 #[test]
602 fn test_is_unit_healthy() {
603 assert!(!is_unit_unhealthy(
605 SystemdUnitActiveState::active,
606 SystemdUnitLoadState::loaded
607 ));
608 assert!(is_unit_unhealthy(
610 SystemdUnitActiveState::activating,
611 SystemdUnitLoadState::loaded
612 ));
613 assert!(!is_unit_unhealthy(
615 SystemdUnitActiveState::activating,
616 SystemdUnitLoadState::masked
617 ));
618 assert!(is_unit_unhealthy(
620 SystemdUnitActiveState::deactivating,
621 SystemdUnitLoadState::not_found
622 ));
623 assert!(is_unit_unhealthy(
624 SystemdUnitActiveState::active,
626 SystemdUnitLoadState::error,
627 ));
628 }
629
630 #[tokio::test]
631 async fn test_state_parse() -> Result<(), MonitordUnitsError> {
632 let test_unit_name = String::from("apport-autoreport.timer");
633 let expected_stats = SystemdUnitStats {
634 activating_units: 0,
635 active_units: 0,
636 automount_units: 0,
637 device_units: 0,
638 failed_units: 0,
639 inactive_units: 0,
640 jobs_queued: 0,
641 loaded_units: 0,
642 masked_units: 0,
643 mount_units: 0,
644 not_found_units: 0,
645 path_units: 0,
646 scope_units: 0,
647 service_units: 0,
648 slice_units: 0,
649 socket_units: 0,
650 target_units: 0,
651 timer_units: 0,
652 timer_persistent_units: 0,
653 timer_remain_after_elapse: 0,
654 total_units: 0,
655 service_stats: HashMap::new(),
656 timer_stats: HashMap::new(),
657 unit_states: HashMap::from([(
658 test_unit_name.clone(),
659 UnitStates {
660 active_state: SystemdUnitActiveState::inactive,
661 load_state: SystemdUnitLoadState::loaded,
662 unhealthy: true,
663 time_in_state_usecs: None,
664 },
665 )]),
666 };
667 let mut stats = SystemdUnitStats::default();
668 let systemd_unit = get_unit_file();
669 let mut config = crate::config::UnitsConfig::default();
670
671 parse_state(&mut stats, &systemd_unit, &config, None).await?;
673 assert_eq!(expected_stats, stats);
674
675 config.state_stats_allowlist = HashSet::from([test_unit_name.clone()]);
677
678 let mut allowlist_stats = SystemdUnitStats::default();
680 parse_state(&mut allowlist_stats, &systemd_unit, &config, None).await?;
681 assert_eq!(expected_stats, allowlist_stats);
682
683 config.state_stats_blocklist = HashSet::from([test_unit_name]);
685
686 let mut blocklist_stats = SystemdUnitStats::default();
688 let expected_blocklist_stats = SystemdUnitStats::default();
689 parse_state(&mut blocklist_stats, &systemd_unit, &config, None).await?;
690 assert_eq!(expected_blocklist_stats, blocklist_stats);
691 Ok(())
692 }
693
694 #[test]
695 fn test_unit_parse() {
696 let expected_stats = SystemdUnitStats {
697 activating_units: 0,
698 active_units: 0,
699 automount_units: 0,
700 device_units: 0,
701 failed_units: 0,
702 inactive_units: 1,
703 jobs_queued: 0,
704 loaded_units: 1,
705 masked_units: 0,
706 mount_units: 0,
707 not_found_units: 0,
708 path_units: 0,
709 scope_units: 0,
710 service_units: 0,
711 slice_units: 0,
712 socket_units: 0,
713 target_units: 0,
714 timer_units: 1,
715 timer_persistent_units: 0,
716 timer_remain_after_elapse: 0,
717 total_units: 0,
718 service_stats: HashMap::new(),
719 timer_stats: HashMap::new(),
720 unit_states: HashMap::new(),
721 };
722 let mut stats = SystemdUnitStats::default();
723 let systemd_unit = get_unit_file();
724 parse_unit(&mut stats, &systemd_unit);
725 assert_eq!(expected_stats, stats);
726 }
727
728 #[test]
729 fn test_unit_parse_activating() {
730 let mut activating_unit = get_unit_file();
731 activating_unit.active_state = String::from("activating");
732 let mut stats = SystemdUnitStats::default();
733 parse_unit(&mut stats, &activating_unit);
734 assert_eq!(stats.activating_units, 1);
735 assert_eq!(stats.active_units, 0);
736 assert_eq!(stats.inactive_units, 0);
737 }
738
739 #[test]
740 fn test_iterators() {
741 assert!(SystemdUnitActiveState::iter().collect::<Vec<_>>().len() > 0);
742 assert!(SystemdUnitLoadState::iter().collect::<Vec<_>>().len() > 0);
743 }
744}