This repository has been archived on 2026-07-13. You can view files and clone it, but you cannot make any changes to its state, such as pushing and creating new issues, pull requests or comments.
nova-shell/plugin/src/systemd_service.rs

626 lines
20 KiB
Rust

// In-process systemd state for nova-shell.
//
// The machine tree is exposed to QML as a JSON-encoded QString (`machinesJson`)
// rather than QList<QVariantMap>. cxx-qt 0.8.1 does not implement QVariantValue
// for QVariantMap/QVariantList, and cxx-qt main on git regressed qt-build-utils
// to require QuickControls2.prl files that nixpkgs strips. Switch to
// QList<QVariantMap> when a release ships with both fixes.
//
// Local host data uses native zbus over the system + session buses. Everything
// else (local nspawn containers, configured remote hosts, and remote nspawn
// containers) is fetched by spawning `busctl` with the right combination of
// `--host=<ssh-target>` and `--machine=<container>` flags. busctl uses
// systemd-machined's OpenMachine() to tunnel sd-bus into a container, and ssh
// to reach a remote host. Remote fetches only happen while the applet is open
// (toggled via `setAppletOpen`).
use crate::modules_service;
use core::pin::Pin;
use cxx_qt::CxxQtType;
use cxx_qt_lib::QString;
use serde::Serialize;
use std::collections::HashMap;
use std::sync::OnceLock;
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::runtime::Runtime;
use zbus::{Connection, proxy};
#[cxx_qt::bridge]
pub mod qobject {
unsafe extern "C++" {
include!("cxx-qt-lib/qstring.h");
type QString = cxx_qt_lib::QString;
}
extern "RustQt" {
#[qobject]
#[qml_element]
#[qml_singleton]
#[qproperty(QString, hostname)]
// Local failed unit count (drives the bar module label).
#[qproperty(i32, failed_count, cxx_name = "failedCount")]
// Recursive JSON tree. Top level is [local, remote1, remote2, ...].
// Each entry:
// { name, isLocal, marker, systemState, runningCount, totalCount,
// failedUnits: [{name, description, subState, scope, machine}],
// runningUnits: [...],
// errorKind: ""|"transient"|"permanent",
// errorReason: string,
// lastSeen: 0 | unix_seconds,
// containers: [<MachineJson>, ...] }
#[qproperty(QString, machines_json, cxx_name = "machinesJson")]
type SystemdService = super::SystemdServiceRust;
#[qinvokable]
fn poll(self: Pin<&mut Self>);
#[qinvokable]
#[cxx_name = "setAppletOpen"]
fn set_applet_open(self: Pin<&mut Self>, open: bool);
#[qinvokable]
#[cxx_name = "restartUnit"]
fn restart_unit(self: Pin<&mut Self>, name: QString, scope: QString, machine: QString);
}
impl cxx_qt::Initialize for SystemdService {}
}
// systemd1.Manager.ListUnits returns a(ssssssouso): name, description,
// load_state, active_state, sub_state, follower, unit_path, job_id, job_type, job_path.
type UnitTuple = (
String,
String,
String,
String,
String,
String,
zbus::zvariant::OwnedObjectPath,
u32,
String,
zbus::zvariant::OwnedObjectPath,
);
#[proxy(
interface = "org.freedesktop.systemd1.Manager",
default_service = "org.freedesktop.systemd1",
default_path = "/org/freedesktop/systemd1"
)]
trait SystemdManager {
#[zbus(property)]
fn system_state(&self) -> zbus::Result<String>;
fn list_units(&self) -> zbus::Result<Vec<UnitTuple>>;
fn restart_unit(&self, name: &str, mode: &str)
-> zbus::Result<zbus::zvariant::OwnedObjectPath>;
}
#[proxy(
interface = "org.freedesktop.machine1.Manager",
default_service = "org.freedesktop.machine1",
default_path = "/org/freedesktop/machine1"
)]
trait Machined {
fn list_machines(
&self,
) -> zbus::Result<Vec<(String, String, String, zbus::zvariant::OwnedObjectPath)>>;
}
#[derive(Serialize, Clone)]
struct UnitJson {
name: String,
description: String,
#[serde(rename = "subState")]
sub_state: String,
scope: String,
machine: String,
}
#[derive(Serialize, Clone)]
struct MachineJson {
name: String,
#[serde(rename = "isLocal")]
is_local: bool,
marker: String,
#[serde(rename = "systemState")]
system_state: String,
#[serde(rename = "runningCount")]
running_count: i32,
#[serde(rename = "totalCount")]
total_count: i32,
#[serde(rename = "failedUnits")]
failed_units: Vec<UnitJson>,
#[serde(rename = "runningUnits")]
running_units: Vec<UnitJson>,
#[serde(rename = "errorKind")]
error_kind: String,
#[serde(rename = "errorReason")]
error_reason: String,
#[serde(rename = "lastSeen")]
last_seen: u64,
#[serde(default)]
containers: Vec<MachineJson>,
}
impl MachineJson {
fn placeholder(name: String, is_local: bool) -> Self {
Self {
name,
is_local,
marker: String::new(),
system_state: "pending".into(),
running_count: 0,
total_count: 0,
failed_units: Vec::new(),
running_units: Vec::new(),
error_kind: String::new(),
error_reason: String::new(),
last_seen: 0,
containers: Vec::new(),
}
}
fn errored(name: String, kind: &'static str, reason: String, prev_last_seen: u64) -> Self {
Self {
name,
is_local: false,
marker: String::new(),
system_state: "unreachable".into(),
running_count: 0,
total_count: 0,
failed_units: Vec::new(),
running_units: Vec::new(),
error_kind: kind.to_string(),
error_reason: reason,
last_seen: prev_last_seen,
containers: Vec::new(),
}
}
}
pub struct SystemdServiceRust {
hostname: QString,
failed_count: i32,
machines_json: QString,
applet_open: bool,
// Last successful or attempted remote fetch per ssh target. Persists across
// polls so the UI can show stale data with a "last seen" timestamp when
// the host goes unreachable.
remote_cache: HashMap<String, MachineJson>,
}
impl Default for SystemdServiceRust {
fn default() -> Self {
Self {
hostname: QString::from(read_hostname()),
failed_count: 0,
machines_json: QString::from("[]"),
applet_open: false,
remote_cache: HashMap::new(),
}
}
}
fn read_hostname() -> String {
std::fs::read_to_string("/etc/hostname")
.map(|s| s.trim().to_string())
.unwrap_or_else(|_| "localhost".to_string())
}
fn now_unix() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
fn rt() -> &'static Runtime {
static RT: OnceLock<Runtime> = OnceLock::new();
RT.get_or_init(|| {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("tokio runtime")
})
}
async fn fetch_units_zbus(bus: &Connection) -> (String, Vec<UnitTuple>) {
let mut state = String::from("unknown");
let mut units = Vec::new();
if let Ok(mgr) = SystemdManagerProxy::new(bus).await {
if let Ok(s) = mgr.system_state().await {
state = s;
}
if let Ok(u) = mgr.list_units().await {
units = u;
}
}
(state, units)
}
// Partition a unit list by active_state. (failed, running, total)
fn partition_units(
units: Vec<UnitTuple>,
scope: &str,
machine: &str,
) -> (Vec<UnitJson>, Vec<UnitJson>, i32) {
let total = units.len() as i32;
let mut failed = Vec::new();
let mut running = Vec::new();
for u in units {
let entry = UnitJson {
name: u.0,
description: u.1,
sub_state: u.4,
scope: scope.to_string(),
machine: machine.to_string(),
};
match u.3.as_str() {
"failed" => failed.push(entry),
"active" => running.push(entry),
_ => {}
}
}
(failed, running, total)
}
async fn poll_local_zbus() -> (
String,
Vec<UnitTuple>,
Vec<UnitTuple>,
Vec<(String, String, String)>,
) {
let mut sys_state = String::from("unknown");
let mut sys_units = Vec::new();
let mut user_units = Vec::new();
let mut machines = Vec::new();
if let Ok(c) = Connection::system().await {
let (s, u) = fetch_units_zbus(&c).await;
sys_state = s;
sys_units = u;
if let Ok(m) = MachinedProxy::new(&c).await {
if let Ok(list) = m.list_machines().await {
machines = list
.into_iter()
.filter(|(name, _, _, _)| name != ".host")
.map(|(n, c, s, _)| (n, c, s))
.collect();
}
}
}
if let Ok(c) = Connection::session().await {
let (_, u) = fetch_units_zbus(&c).await;
user_units = u;
}
(sys_state, sys_units, user_units, machines)
}
// Run busctl with the given prefix args plus the trailing call args. Returns
// a parsed `data` value or an error classification.
fn busctl_call(
prefix: &[&str],
call: &[&str],
) -> Result<serde_json::Value, (&'static str, String)> {
let mut args: Vec<&str> = prefix.to_vec();
args.extend_from_slice(call);
let out = std::process::Command::new("busctl")
.args(&args)
.output()
.map_err(|e| ("transient", format!("busctl spawn failed: {e}")))?;
if !out.status.success() {
let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
return Err((classify_error(&stderr), stderr));
}
serde_json::from_slice(&out.stdout)
.map_err(|e| ("transient", format!("invalid busctl json: {e}")))
}
// Fetch SystemState + ListUnits via busctl with the given prefix flags.
// `prefix` examples:
// [] - local system bus (unused; we use zbus locally)
// ["--host", "muede-pc2"] - remote host
// ["--machine", "damocles-lab"] - local container
// ["--host", "pc2", "--machine", "x"] - container on a remote host
fn fetch_via_busctl(
name: &str,
prefix: &[&str],
is_local: bool,
marker: &str,
) -> Result<MachineJson, (&'static str, String)> {
let state_v = busctl_call(
prefix,
&[
"--json=short",
"get-property",
"org.freedesktop.systemd1",
"/org/freedesktop/systemd1",
"org.freedesktop.systemd1.Manager",
"SystemState",
],
)?;
let system_state = state_v
.get("data")
.and_then(|d| d.as_str())
.unwrap_or("unknown")
.to_string();
let units_v = busctl_call(
prefix,
&[
"--json=short",
"call",
"org.freedesktop.systemd1",
"/org/freedesktop/systemd1",
"org.freedesktop.systemd1.Manager",
"ListUnits",
],
)?;
let arr = units_v
.get("data")
.and_then(|d| d.get(0))
.and_then(|d| d.as_array())
.ok_or(("transient", "missing ListUnits data".to_string()))?;
let mut total = 0i32;
let mut failed = Vec::new();
let mut running = Vec::new();
for u in arr {
let arr = match u.as_array() {
Some(a) if a.len() >= 5 => a,
_ => continue,
};
total += 1;
let entry = UnitJson {
name: arr[0].as_str().unwrap_or("").to_string(),
description: arr[1].as_str().unwrap_or("").to_string(),
sub_state: arr[4].as_str().unwrap_or("").to_string(),
scope: "system".to_string(),
machine: name.to_string(),
};
match arr[3].as_str() {
Some("failed") => failed.push(entry),
Some("active") => running.push(entry),
_ => {}
}
}
Ok(MachineJson {
name: name.to_string(),
is_local,
marker: marker.to_string(),
system_state,
running_count: running.len() as i32,
total_count: total,
failed_units: failed,
running_units: running,
error_kind: String::new(),
error_reason: String::new(),
last_seen: now_unix(),
containers: Vec::new(),
})
}
// List nspawn containers under the given busctl prefix (empty = local).
fn list_containers_busctl(prefix: &[&str]) -> Vec<String> {
let v = match busctl_call(
prefix,
&[
"--json=short",
"call",
"org.freedesktop.machine1",
"/org/freedesktop/machine1",
"org.freedesktop.machine1.Manager",
"ListMachines",
],
) {
Ok(v) => v,
Err(_) => return Vec::new(),
};
let arr = match v
.get("data")
.and_then(|d| d.get(0))
.and_then(|d| d.as_array())
{
Some(a) => a,
None => return Vec::new(),
};
arr.iter()
.filter_map(|m| m.as_array())
.filter_map(|a| a.first()?.as_str())
.filter(|n| *n != ".host")
.map(String::from)
.collect()
}
// Classify an ssh/busctl stderr blob into transient vs permanent so the UI
// can pick a color and decide whether to retry aggressively.
fn classify_error(stderr: &str) -> &'static str {
let s = stderr.to_lowercase();
if s.contains("permission denied")
|| s.contains("publickey")
|| s.contains("authentication failed")
|| s.contains("host key verification failed")
|| s.contains("could not resolve hostname")
{
"permanent"
} else {
"transient"
}
}
impl cxx_qt::Initialize for qobject::SystemdService {
fn initialize(self: Pin<&mut Self>) {
self.poll();
}
}
impl qobject::SystemdService {
fn poll(mut self: Pin<&mut Self>) {
let (sys_state, sys_units, user_units, local_container_names) =
rt().block_on(poll_local_zbus());
let (sys_failed, sys_running, sys_total) = partition_units(sys_units, "system", "");
let (user_failed, user_running, user_total) = partition_units(user_units, "user", "");
let mut failed: Vec<UnitJson> = Vec::with_capacity(sys_failed.len() + user_failed.len());
failed.extend(sys_failed);
failed.extend(user_failed);
let mut running: Vec<UnitJson> = Vec::with_capacity(sys_running.len() + user_running.len());
running.extend(sys_running);
running.extend(user_running);
let local_failed_count = failed.len() as i32;
let local_running_count = running.len() as i32;
let local_total_count = sys_total + user_total;
let local_containers: Vec<MachineJson> = if self.as_ref().rust().applet_open {
local_container_names
.iter()
.map(|(name, _, _)| {
fetch_via_busctl(name, &["--machine", name], false, "")
.unwrap_or_else(|(k, r)| MachineJson::errored(name.clone(), k, r, 0))
})
.collect()
} else {
local_container_names
.iter()
.map(|(n, _, _)| MachineJson::placeholder(n.clone(), false))
.collect()
};
let local = MachineJson {
name: read_hostname(),
is_local: true,
marker: "this machine".into(),
system_state: sys_state,
running_count: local_running_count,
total_count: local_total_count,
failed_units: failed,
running_units: running,
error_kind: String::new(),
error_reason: String::new(),
last_seen: now_unix(),
containers: local_containers,
};
let mut all_machines = Vec::new();
all_machines.push(local);
// Configured remote machines. Dedup local: drop entries matching the
// local hostname or `localhost`, with or without a `user@` prefix.
let host = read_hostname();
let cfg_machines = modules_service::load_systemd_machines();
let mut want_targets: Vec<String> = Vec::new();
for target in cfg_machines {
let host_part = target.rsplit_once('@').map_or(target.as_str(), |(_, h)| h);
if host_part == host || host_part == "localhost" {
continue;
}
want_targets.push(target);
}
// Fetch only while the applet is open. Keep last-known data in the
// cache so a re-open shows something instantly.
let applet_open = self.as_ref().rust().applet_open;
if applet_open {
for target in &want_targets {
let prev_last_seen = self
.as_ref()
.rust()
.remote_cache
.get(target)
.map(|m| m.last_seen)
.unwrap_or(0);
let entry = match fetch_via_busctl(target, &["--host", target], false, "") {
Ok(mut m) => {
// Enumerate + fetch remote nspawn containers via the
// remote host's machined.
let names = list_containers_busctl(&["--host", target]);
m.containers = names
.iter()
.map(|cn| {
fetch_via_busctl(
cn,
&["--host", target, "--machine", cn],
false,
"",
)
.unwrap_or_else(|(k, r)| MachineJson::errored(cn.clone(), k, r, 0))
})
.collect();
m
}
Err((kind, reason)) => {
MachineJson::errored(target.clone(), kind, reason, prev_last_seen)
}
};
self.as_mut()
.rust_mut()
.remote_cache
.insert(target.clone(), entry);
}
}
// Drop cache entries that are no longer configured.
let want_set: std::collections::HashSet<&str> =
want_targets.iter().map(String::as_str).collect();
self.as_mut()
.rust_mut()
.remote_cache
.retain(|k: &String, _: &mut MachineJson| want_set.contains(k.as_str()));
// Emit configured remotes from cache (or placeholder if never fetched).
for target in &want_targets {
let entry = self
.as_ref()
.rust()
.remote_cache
.get(target)
.cloned()
.unwrap_or_else(|| MachineJson::placeholder(target.clone(), false));
all_machines.push(entry);
}
let machines_json = serde_json::to_string(&all_machines).unwrap_or_else(|_| "[]".into());
self.as_mut().set_failed_count(local_failed_count);
self.as_mut()
.set_machines_json(QString::from(machines_json));
}
fn set_applet_open(mut self: Pin<&mut Self>, open: bool) {
let was_open = self.as_ref().rust().applet_open;
self.as_mut().rust_mut().applet_open = open;
if open && !was_open {
self.poll();
}
}
fn restart_unit(self: Pin<&mut Self>, name: QString, scope: QString, machine: QString) {
let name = name.to_string();
let scope = scope.to_string();
let machine = machine.to_string();
let _ = self;
rt().block_on(async move {
// Local-only restart for now. Container/remote restart is TODO.
if !machine.is_empty() {
tracing::warn!(target: "nova_plugin", machine = %machine, "container/remote restart not yet implemented");
return;
}
let conn = match scope.as_str() {
"user" => Connection::session().await,
_ => Connection::system().await,
};
let Ok(conn) = conn else { return };
let Ok(mgr) = SystemdManagerProxy::new(&conn).await else {
return;
};
if let Err(e) = mgr.restart_unit(&name, "replace").await {
tracing::warn!(target: "nova_plugin", unit = %name, error = %e, "restart_unit failed");
}
});
}
}