Cluster upgrade

docs/index.rst

@@ -52,8 +52,23 @@ welcome—see the section on contributing for how to get involved.
    tutorial/wasm_modules_tutor
    tutorial/lua_modules_tutor
    tutorial/menotify_tutor
-   tutorial/menotify_sensor_dev_tutor
+    tutorial/menotify_sensor_dev_tutor
 
+Source Code
+-----------
+
+The project source code is hosted on GitHub:
+`github.com/tinythings/sysinspect <https://github.com/tinythings/sysinspect>`_.
+
+Related Projects
+----------------
+
+`LogJet <https://github.com/tinythings/logjet>`_ is a sibling project providing
+OTLP telemetry storage, replay, and bridge functionality.  It acts as a durable
+log, metric, and trace buffer that sits between local telemetry sources and
+downstream collectors — optimised for weak hardware, limited RAM, intermittent
+connectivity, and sequential replay.  Together with Sysinspect, it enables
+end-to-end disconnected-capable observability pipelines.
 
 Licence
 -------

examples/demos/infoping/model.cfg

@@ -203,6 +203,12 @@ actions:
         args:
           host: "claim(connectivity.target)"
           count: "claim(connectivity.ping-count)"
+      internet:
+        opts:
+          - ping
+        args:
+          host: google.com
+          count: 2
 
   check-sshd:
     descr: Verify SSH daemon status

libsysinspect/src/console/mod.rs

@@ -119,6 +119,17 @@ pub struct ConsoleMinionProcessSignalRequest {
     pub signal: i32,
 }
 
+/// Request parameters for a minion self-upgrade via the master fileserver.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ConsoleMinionUpgradeSelfRequest {
+    /// Relative subpath of the new binary under the master fileserver root.
+    pub subpath: String,
+    /// Expected SHA-256 checksum of the new binary.
+    pub checksum: String,
+    /// Version label of the new binary.
+    pub version: String,
+}
+
 fn default_top_process_limit() -> usize {
     24
 }
@@ -336,6 +347,32 @@ pub enum ConsolePayload {
         #[serde(default)]
         failures: Vec<String>,
     },
+    /// Current cluster upgrade counts derived from master-side Sled markers.
+    UpgradeStatus {
+        /// Number of minions still marked as requiring upgrade/sync.
+        required: usize,
+        /// Number of marked minions last seen as unreachable during upgrade.
+        unreachable: usize,
+        /// Number of minions pending post-upgrade auto-hopstart.
+        #[serde(default)]
+        pending_post_upgrade: usize,
+    },
+    /// Result summary for one cluster upgrade run.
+    UpgradeSummary {
+        /// Number of offline minions upgraded successfully over SSH.
+        updated: usize,
+        /// Number of online minions that received a self-upgrade command.
+        dispatched: usize,
+        /// Number of minions skipped because they were unavailable or unmanaged.
+        skipped: usize,
+        /// Number of upgrade attempts that failed unexpectedly.
+        failed: usize,
+        /// Number of minions that were offline and had no SSH fallback.
+        offline: usize,
+        /// Human-readable details for skipped or failed minions.
+        #[serde(default)]
+        items: Vec<String>,
+    },
 }
 
 /// One online-minion summary row returned by the master.
@@ -363,6 +400,12 @@ pub struct ConsoleOnlineMinionRow {
     /// Whether the current runtime version is older than the matching repository version.
     #[serde(default)]
     pub outdated: bool,
+    /// Whether this minion is currently marked in the master's upgrade queue.
+    #[serde(default)]
+    pub upgrade_required: bool,
+    /// Whether the latest upgrade attempt marked this minion unreachable.
+    #[serde(default)]
+    pub upgrade_unreachable: bool,
     /// OS distribution reported by minion traits, if present.
     #[serde(default)]
     pub os_distribution: String,

libsysinspect/src/journal_ut.rs

@@ -1,5 +1,19 @@
 use crate::journal::Journal;
 
+fn open_with_retry(dir: &std::path::Path, max_bytes: u64) -> Journal {
+    let mut last_err = None;
+    for _ in 0..20 {
+        match Journal::open(dir, max_bytes) {
+            Ok(journal) => return journal,
+            Err(err) => {
+                last_err = Some(err);
+                std::thread::sleep(std::time::Duration::from_millis(25));
+            }
+        }
+    }
+    panic!("failed to reopen journal: {}", last_err.unwrap());
+}
+
 fn temp_dir() -> std::path::PathBuf {
     let dir = std::env::temp_dir().join(format!(
         "libsysinspect-journal-ut-{}-{}",
@@ -73,7 +87,7 @@ fn state_persists_across_reopen() {
     let j1 = Journal::open(&dir, 0).unwrap();
     j1.append("c1", b"lost").unwrap();
     drop(j1);
-    let j2 = Journal::open(&dir, 0).unwrap();
+    let j2 = open_with_retry(&dir, 0);
     let pending = j2.pending().unwrap();
     assert_eq!(pending.len(), 1);
     assert_eq!(pending[0].0, "c1");
@@ -133,7 +147,7 @@ fn reopen_after_partial_ack_preserves_survivors() {
         j.ack_cycle("c2").unwrap();
     }
     {
-        let j = Journal::open(&dir, 0).unwrap();
+        let j = open_with_retry(&dir, 0);
         let pending = j.pending().unwrap();
         assert_eq!(pending.len(), 1);
         assert_eq!(pending[0].0, "c1");
@@ -149,7 +163,7 @@ fn budget_applies_after_reopen() {
         j.append("c1", b"1234567890").unwrap();
     }
     {
-        let j = Journal::open(&dir, 20).unwrap();
+        let j = open_with_retry(&dir, 20);
         j.append("c1", b"abcdefghij").unwrap();
         j.append("c2", b"overflow!").unwrap();
         let pending = j.pending().unwrap();
@@ -229,7 +243,7 @@ fn completed_cycle_marker_persists_across_reopen_until_ack() {
         assert!(j.is_cycle_locally_complete("c1").unwrap());
     }
     {
-        let j = Journal::open(&dir, 0).unwrap();
+        let j = open_with_retry(&dir, 0);
         assert!(j.is_cycle_locally_complete("c1").unwrap());
         j.ack_cycle("c1").unwrap();
         assert!(!j.is_cycle_locally_complete("c1").unwrap());
@@ -374,7 +388,7 @@ fn reopen_and_continue_appending() {
         j.ack_cycle("c1").unwrap();
     }
     {
-        let j = Journal::open(&dir, 0).unwrap();
+        let j = open_with_retry(&dir, 0);
         j.append("c2", b"c").unwrap();
         j.append("c3", b"d").unwrap();
         let pending = j.pending().unwrap();

libsysproto/src/query.rs

@@ -68,6 +68,18 @@ pub mod commands {
     // Send a Unix signal to one process on one specific minion
     pub const CLUSTER_MINION_PROCESS_SIGNAL: &str = "cluster/minion/process/signal";
 
+    // Mark all selected minions as requiring a cluster upgrade/sync
+    pub const CLUSTER_MARK_UPGRADE_REQUIRED: &str = "cluster/upgrade/mark";
+
+    // Run cluster upgrade/sync across marked minions
+    pub const CLUSTER_UPGRADE_MINIONS: &str = "cluster/upgrade/run";
+
+    // Read current cluster upgrade status counts
+    pub const CLUSTER_UPGRADE_STATUS: &str = "cluster/upgrade/status";
+
+    // Ask a minion to download, verify, and replace its own binary
+    pub const CLUSTER_MINION_UPGRADE_SELF: &str = "cluster/minion/upgrade/self";
+
     // Force all online minions to reconnect (cluster-wide broadcast)
     pub const CLUSTER_RECONNECT: &str = "cluster/reconnect";
 

src/clifmt.rs

@@ -416,5 +416,22 @@ pub fn render_console_payload(payload: &ConsolePayload) -> String {
         ConsolePayload::MasterLogs { snapshot: _ } => String::new(),
         ConsolePayload::MasterModuleIndex { .. } => String::new(),
         ConsolePayload::MasterLibraryIndex { .. } => String::new(),
+        ConsolePayload::UpgradeStatus { required, unreachable, .. } => {
+            format!("Cluster upgrade status: required={required}, unreachable={unreachable}")
+        }
+        ConsolePayload::UpgradeSummary { updated, dispatched, skipped, failed, offline, items } => {
+            let mut out = vec![
+                format!("SSH upgraded:         {updated}"),
+                format!("Dispatched to online: {dispatched}"),
+                format!("Skipped:              {skipped}"),
+                format!("Failed:               {failed}"),
+                format!("Offline:              {offline}"),
+            ];
+            if !items.is_empty() {
+                out.push(String::new());
+                out.extend(items.clone());
+            }
+            out.join("\n")
+        }
     }
 }