feat(orb-health): add orbd daemon, filesizes health

Cargo.toml

@@ -38,8 +38,10 @@ members = [
 	"orb-connd/dbus",
 	"orb-core/agent-iroh",
 	"orb-dogd",
+	"orb-health",
 	"orb-info",
 	"orb-jobs-agent",
+	"orbd",
 	"ota-backend",
 	"prelude",
 	"qr-link",
@@ -179,6 +181,7 @@ tracing-opentelemetry = { version = "0.28", default-features = false }
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 url = "2.5.4"
 uuid = "1.19.0"
+walkdir = "2"
 wiremock = "0.6.4"
 x509-parser = "0.17.0"
 zbus = { version = "4.4.0", default-features = false, features = ["tokio"] }
@@ -209,6 +212,7 @@ orb-const-concat.path = "const-concat"
 orb-dogd.path = "orb-dogd"
 orb-endpoints.path = "endpoints"
 orb-header-parsing.path = "header-parsing"
+orb-health.path = "orb-health"
 orb-info = { path = "orb-info", default-features = false }
 orb-io-utils.path = "io-utils"
 orb-mcu-interface.path = "mcu-interface"

orb-health/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "orb-health"
+version = "0.1.0"
+description = "Orb Health service (orbd service)"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+
+[dependencies]
+color-eyre.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+walkdir.workspace = true

orb-health/src/file_sizes.rs

@@ -0,0 +1,43 @@
+use color_eyre::eyre::Result;
+use std::path::PathBuf;
+use tokio::task;
+use tracing::{error, info};
+use walkdir::WalkDir;
+
+const PERSISTENT: &str = "/usr/persistent";
+
+pub async fn run() -> Result<()> {
+    task::spawn_blocking(|| collect(PERSISTENT)).await?
+}
+
+fn collect(root: &str) -> Result<()> {
+    let root = PathBuf::from(root);
+    let mut entries = Vec::new();
+
+    for entry in WalkDir::new(root).follow_links(false) {
+        let entry = match entry {
+            Ok(e) => e,
+            Err(e) => {
+                error!("failed to read entry: {e}");
+                continue;
+            }
+        };
+
+        if entry.file_type().is_file() {
+            match entry.metadata() {
+                Ok(meta) => entries.push((entry.into_path(), meta.len())),
+                Err(e) => {
+                    error!(path = %entry.path().display(), "failed to read metadata: {e}")
+                }
+            }
+        }
+    }
+
+    entries.sort_unstable_by(|a, b| b.1.cmp(&a.1));
+
+    for (path, size_bytes) in entries {
+        info!("{}: {size_bytes} bytes", path.display());
+    }
+
+    Ok(())
+}

orb-health/src/lib.rs

@@ -0,0 +1,9 @@
+use tracing::warn;
+
+pub mod file_sizes;
+
+pub async fn run() {
+    if let Err(e) = file_sizes::run().await {
+        warn!("orb-health::file_sizes failed: {e}");
+    }
+}

orbd/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "orbd"
+version = "0.1.0"
+description = "Orb daemon (orbd) - monolith deployment unit"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+authors = ["Philipp Popov <pophilpo@users.noreply.github.com>"]
+
+[dependencies]
+clap = { workspace = true, features = ["derive"] }
+color-eyre.workspace = true
+orb-build-info.workspace = true
+orb-health.workspace = true
+orb-telemetry.workspace = true
+tokio = { workspace = true, features = ["full"] }
+tracing.workspace = true
+
+[build-dependencies]
+orb-build-info = { workspace = true, features = ["build-script"] }
+
+[package.metadata.deb]
+maintainer-scripts = "debian/"
+assets = [["target/release/orbd", "/usr/local/bin/", "755"]]
+systemd-units = [{ unit-name = "worldcoin-orbd" }]

orbd/build.rs

@@ -0,0 +1,3 @@
+fn main() {
+    orb_build_info::initialize().unwrap();
+}

orbd/debian/worldcoin-orbd.service

@@ -0,0 +1,25 @@
+[Unit]
+Description=Worldcoin Orb Daemon
+After=zenohd.service
+Wants=zenohd.service
+
+[Service]
+Type=exec
+User=worldcoin
+Environment=RUST_BACKTRACE=1
+SyslogIdentifier=worldcoin-orbd
+WorkingDirectory=/home/worldcoin
+ExecStart=/usr/local/bin/orbd
+Restart=always
+RestartSec=3
+
+# Allow traversal of root-owned directories under /usr/persistent to stat
+# file sizes. CAP_DAC_READ_SEARCH bypasses directory read/execute permission
+# checks. CapabilityBoundingSet restricts to only this capability.
+# This does not allow reading the root-owned files.
+CapabilityBoundingSet=CAP_DAC_READ_SEARCH
+AmbientCapabilities=CAP_DAC_READ_SEARCH
+NoNewPrivileges=yes
+
+[Install]
+WantedBy=multi-user.target

orbd/src/lib.rs

@@ -0,0 +1,149 @@
+#[allow(unused_imports)]
+use std::future::Future;
+
+#[allow(unused_imports)]
+use std::pin::Pin;
+
+#[allow(unused_imports)]
+use color_eyre::eyre::Result;
+#[allow(unused_imports)]
+use tokio::signal::unix::{self, SignalKind};
+#[allow(unused_imports)]
+use tokio::sync::watch;
+#[allow(unused_imports)]
+use tokio::task::JoinSet;
+
+use tracing::{info, warn};
+
+type BoxComponentFuture = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
+
+pub struct Component {
+    name: &'static str,
+    run: Box<dyn FnOnce(ComponentContext) -> BoxComponentFuture + Send + 'static>,
+}
+
+impl Component {
+    pub fn new<F, Fut>(name: &'static str, run: F) -> Self
+    where
+        F: FnOnce(ComponentContext) -> Fut + Send + 'static,
+        Fut: Future<Output = ()> + Send + 'static,
+    {
+        Self {
+            name,
+            run: Box::new(move |ctx| Box::pin(run(ctx))),
+        }
+    }
+
+    pub fn name(&self) -> &'static str {
+        self.name
+    }
+
+    async fn run(self, ctx: ComponentContext) {
+        (self.run)(ctx).await
+    }
+}
+
+#[derive(Clone)]
+pub struct ComponentContext {
+    pub shutdown: Shutdown,
+}
+
+#[derive(Clone)]
+pub struct Shutdown {
+    rx: watch::Receiver<bool>,
+}
+
+impl Shutdown {
+    pub fn is_cancelled(&self) -> bool {
+        *self.rx.borrow()
+    }
+
+    pub async fn cancelled(&mut self) {
+        if self.is_cancelled() {
+            return;
+        }
+
+        loop {
+            if self.rx.changed().await.is_err() || self.is_cancelled() {
+                return;
+            }
+        }
+    }
+}
+
+#[derive(Default)]
+pub struct Program {
+    components: Vec<Component>,
+}
+
+impl Program {
+    pub fn new() -> Self {
+        Self {
+            components: Vec::new(),
+        }
+    }
+
+    pub fn component(&mut self, component: Component) {
+        self.components.push(component);
+    }
+
+    pub async fn run(self) -> Result<()> {
+        let (shudown_tx, shutdown_rx) = watch::channel(false);
+
+        let mut components = JoinSet::new();
+
+        for component in self.components {
+            let name = component.name();
+
+            let ctx = ComponentContext {
+                shutdown: Shutdown {
+                    rx: shutdown_rx.clone(),
+                },
+            };
+
+            info!(component = name, "starting component");
+
+            components.spawn(async move {
+                component.run(ctx).await;
+                name
+            });
+        }
+
+        let mut sigterm = unix::signal(SignalKind::terminate())?;
+        let mut sigint = unix::signal(SignalKind::interrupt())?;
+
+        loop {
+            tokio::select! {
+            _ = sigterm.recv() => {
+                warn!("recieved SIGTERM");
+                let _ = shudown_tx.send(true);
+                break;
+            },
+
+            _ = sigint.recv() => {
+                warn!("received SIGINT");
+                let _ = shudown_tx.send(true);
+                break;
+            }
+
+            Some(result) = components.join_next() => {
+                match result {
+                    Ok(name) => info!(component=name, "component finished"),
+
+                    Err(err) => warn!("component task panicked: {err}"),
+                }
+            }
+
+            }
+        }
+
+        while let Some(result) = components.join_next().await {
+            match result {
+                Ok(name) => info!(component = name, "component stopped"),
+                Err(err) => warn!("component task panicked during shutdown: {err}"),
+            }
+        }
+
+        Ok(())
+    }
+}

orbd/src/main.rs

@@ -0,0 +1,42 @@
+use clap::Parser;
+use color_eyre::eyre::Result;
+use orb_build_info::{make_build_info, BuildInfo};
+use tracing::info;
+
+use orbd::{Component, Program};
+
+const BUILD_INFO: BuildInfo = make_build_info!();
+const SYSLOG_IDENTIFIER: &str = "worldcoin-orbd";
+
+#[derive(Parser, Debug)]
+#[clap(version = BUILD_INFO.version, about)]
+struct Args {}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    color_eyre::install()?;
+
+    let tel_flusher = orb_telemetry::TelemetryConfig::new()
+        .with_journald(SYSLOG_IDENTIFIER)
+        .init();
+
+    Args::parse();
+
+    let result = run().await;
+
+    tel_flusher.flush_blocking();
+
+    result
+}
+
+async fn run() -> Result<()> {
+    info!(version = BUILD_INFO.version, "orbd starting");
+
+    let mut program = Program::new();
+
+    program.component(Component::new("orb-health", |_ctx| async {
+        orb_health::run().await;
+    }));
+
+    program.run().await
+}