Call metrics API for units related stats

Cargo.toml

@@ -16,6 +16,7 @@ categories = ["network-programming", "os::linux-apis"]
 anyhow = "1.0"
 clap = { version = "4.5", features = ["derive", "std", "help"], default-features = false }
 configparser = { version = "3.1.0", features = ["indexmap"] }
+futures-util = "0.3"
 indexmap = "2.9"
 int-enum = "1.1"
 serde = { version = "1.0", features = ["derive"] }
@@ -35,6 +36,7 @@ uzers = "0.12"
 zbus = { version = "5.11.0", features = ["tokio"], default-features = false }
 zvariant = "5.7.0"
 zvariant_derive = "5.7.0"
+zlink = "0.4.0"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 procfs = "0.18"

README.md

@@ -439,7 +439,35 @@ always be running / calling all these DBus calls per run.
 
 ## Varlink
 
-None yet :(.
+monitord supports collecting unit statistics via systemd's [Varlink metrics API](https://github.com/systemd/systemd/pull/39202),
+available in systemd v260+. When enabled, monitord connects to the `io.systemd.Metrics` interface
+at `/run/systemd/report/io.systemd.Manager` to collect unit counts, active/load states, and restart counts.
+
+### Enabling Varlink
+
+Set `enabled = true` in the `[varlink]` section of `monitord.conf`:
+
+```ini
+[varlink]
+enabled = true
+```
+
+When varlink is enabled, monitord will attempt to collect unit stats via the metrics API first.
+If the varlink socket is unavailable (e.g., systemd < v260), it automatically falls back to D-Bus collection.
+
+### Metrics collected via Varlink
+
+- Unit counts by type (service, mount, socket, target, device, automount, timer, path, slice, scope)
+- Unit counts by state (active, failed, inactive)
+- Per-unit active state and load state (with allowlist/blocklist filtering)
+- Per-unit health status (computed from active + load state)
+- Per-service restart counts (`nrestarts`)
+
+### Containers
+
+For systemd-nspawn containers, monitord connects to the container's varlink socket via
+`/proc/<leader_pid>/root/run/systemd/report/io.systemd.Manager`, similar to how D-Bus uses
+the container-scoped bus socket.
 
 ### varlink 101
 

monitord.conf

@@ -51,6 +51,9 @@ sshd.service
 [machines]
 enabled = true
 
+[varlink]
+enabled = false
+
 [machines.allowlist]
 fedora38
 

src/config.rs

@@ -192,6 +192,16 @@
     pub blocklist: HashSet<String>,
 }
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct VarlinkConfig {
+    pub enabled: bool,
+}
+impl Default for VarlinkConfig {
+    fn default() -> Self {
+        VarlinkConfig { enabled: false }
+    }
+}
+
 /// Config struct
 /// Each section represents an ini file section
 #[derive(Clone, Debug, Default, Eq, PartialEq)]
@@ -207,6 +217,7 @@
     pub dbus_stats: DBusStatsConfig,
     pub boot_blame: BootBlameConfig,
     pub verify: VerifyConfig,
+    pub varlink: VarlinkConfig,
 }
 
 impl TryFrom<Ini> for Config {
@@ -325,6 +336,9 @@
             config.verify.blocklist = verify_blocklist.keys().map(|s| s.to_string()).collect();
         }
 
+        // [varlink] section
+        config.varlink.enabled = read_config_bool(&ini_config, "varlink", "enabled")?;
+
         Ok(config)
     }
 }
@@ -427,6 +441,9 @@
 
 [boot.blocklist]
 bar.service
+
+[varlink]
+enabled = true
 "###;
 
     const MINIMAL_CONFIG: &str = r###"
@@ -513,6 +530,7 @@
                 allowlist: HashSet::new(),
                 blocklist: HashSet::new(),
             },
+            varlink: VarlinkConfig { enabled: true },
         };
 
         let mut monitord_config = NamedTempFile::new().expect("Unable to make named tempfile");

src/lib.rs

@@ -31,7 +31,10 @@ pub mod networkd;
 pub mod pid1;
 pub mod system;
 pub mod timer;
+pub mod unit_constants;
 pub mod units;
+pub mod varlink;
+pub mod varlink_units;
 pub mod verify;
 
 pub const DEFAULT_DBUS_ADDRESS: &str = "unix:path=/run/dbus/system_bus_socket";
@@ -167,11 +170,20 @@ pub async fn stat_collector(
 
         // Run service collectors if there are services listed in config
         if config.units.enabled {
-            join_set.spawn(crate::units::update_unit_stats(
-                Arc::clone(&config),
-                sdc.clone(),
-                locked_machine_stats.clone(),
-            ));
+            if config.varlink.enabled {
+                join_set.spawn(crate::varlink_units::update_unit_stats(
+                    Arc::clone(&config),
+                    sdc.clone(),
+                    locked_machine_stats.clone(),
+                    crate::varlink_units::METRICS_SOCKET_PATH.to_string(),
+                ));
+            } else {
+                join_set.spawn(crate::units::update_unit_stats(
+                    Arc::clone(&config),
+                    sdc.clone(),
+                    locked_machine_stats.clone(),
+                ));
+            }
         }
 
         if config.machines.enabled {

src/machines.rs

@@ -102,11 +102,25 @@ pub async fn update_machines_stats(
         ));
 
         if config.units.enabled {
-            join_set.spawn(crate::units::update_unit_stats(
-                Arc::clone(&config),
-                sdc.clone(),
-                locked_machine_stats.clone(),
-            ));
+            if config.varlink.enabled {
+                let container_socket_path = format!(
+                    "/proc/{}/root{}",
+                    leader_pid,
+                    crate::varlink_units::METRICS_SOCKET_PATH
+                );
+                join_set.spawn(crate::varlink_units::update_unit_stats(
+                    Arc::clone(&config),
+                    sdc.clone(),
+                    locked_machine_stats.clone(),
+                    container_socket_path,
+                ));
+            } else {
+                join_set.spawn(crate::units::update_unit_stats(
+                    Arc::clone(&config),
+                    sdc.clone(),
+                    locked_machine_stats.clone(),
+                ));
+            }
         }
 
         if config.dbus_stats.enabled {