HardwareManager: Rework to use a watch channel instead of broadcasting updates (#10194)

This is an attempt to make hardware monitoring somewhat more
level-triggered and somewhat less edge-triggered. In #10187, we had a
case where the two different sources of `HardwareUpdate` notifications
got out of sync, resulting in the polling source failing to send a
`TofinoAvailable` update even though it realized the Tofino was, in
fact, available.

On this branch, we keep the current `HardwareView` in a watch channel;
any changes made to it will result in a `.changed()` notification
firing, removing the possibility of mismatched updates.

Fixes #10187.

Author: jgallagher

sled-agent/config-reconciler/src/raw_disks.rs

@@ -6,7 +6,6 @@
 //! [`RawDisk`]s sled-agent is aware of.
 
 use iddqd::IdOrdMap;
-use omicron_common::disk::DiskIdentity;
 use sled_agent_types::inventory::InventoryDisk;
 use sled_storage::disk::RawDisk;
 use slog::Logger;
@@ -55,10 +54,15 @@ impl RawDisksSender {
             // disks that shouldn't be removed even if they're not present in
             // `new_disks`; check for that first.
             for old_disk in disks.iter() {
-                if !new_disks.contains_key(old_disk.identity())
-                    && !can_remove_disk(old_disk, log)
-                {
-                    new_disks.insert_overwrite(old_disk.clone());
+                if !new_disks.contains_key(old_disk.identity()) {
+                    if can_remove_disk(old_disk, log) {
+                        info!(
+                            log, "Removing disk";
+                            "identity" => ?old_disk.identity(),
+                        );
+                    } else {
+                        new_disks.insert_overwrite(old_disk.clone());
+                    }
                 }
             }
 
@@ -98,31 +102,6 @@ impl RawDisksSender {
         })
     }
 
-    /// Remove a raw disk that is no longer visible to sled-agent.
-    pub fn remove_raw_disk(
-        &self,
-        identity: &DiskIdentity,
-        log: &Logger,
-    ) -> bool {
-        self.0.send_if_modified(|disks| {
-            let Some(disk) = disks.get(identity) else {
-                info!(
-                    log, "Ignoring request to remove nonexistent disk";
-                    "identity" => ?identity,
-                );
-                return false;
-            };
-
-            if !can_remove_disk(disk, log) {
-                return false;
-            }
-
-            info!(log, "Removing disk"; "identity" => ?identity);
-            Arc::make_mut(disks).remove(identity);
-            true
-        })
-    }
-
     pub(crate) fn to_inventory(&self) -> Vec<InventoryDisk> {
         self.0
             .borrow()
@@ -163,6 +142,7 @@ fn can_remove_disk(disk: &RawDisk, log: &Logger) -> bool {
 mod tests {
     use super::*;
     use camino::Utf8PathBuf;
+    use omicron_common::disk::DiskIdentity;
     use omicron_common::disk::DiskVariant;
     use omicron_test_utils::dev;
     use proptest::collection::btree_map;
@@ -252,9 +232,6 @@ mod tests {
         // Change the active firmware slot of the disk at the given index, then
         // pass it to `RawDisksSender::add_or_update_raw_disk`
         Update(Index),
-        // Call `RawDisksSender::remove_raw_disk` with the disk at the given
-        // index
-        Remove(Index),
         // Call `RawDisksSender::set_raw_disks` using the set of disks in the
         // range `[start, start + num)`
         Set { start: Index, num: usize },
@@ -305,19 +282,6 @@ mod tests {
                     states[index].present = true;
                     true
                 }
-                Operation::Remove(index) => {
-                    let index = index.index(states.len());
-                    eprintln!("removing disk {index}");
-                    let was_present = states[index].present;
-                    tx.remove_raw_disk(disks[index].identity(), log);
-                    // Synthetic disks should never be removed
-                    if disks[index].is_synthetic() {
-                        false
-                    } else {
-                        states[index].present = false;
-                        was_present
-                    }
-                }
                 Operation::Set { start, num } => {
                     let start = start.index(states.len());
                     let end =
@@ -381,7 +345,7 @@ mod tests {
             assert_eq!(rx.has_changed().expect("channel open"), expect_changes);
 
             // After this operation, check that the disk is either present or
-            // not (the thing changed by our add/remove/set operations) and has
+            // not (the thing changed by our add or set operations) and has
             // the expected firmware slot (the thing changed by our update
             // operation).
             let current = rx.borrow_and_update();

sled-agent/src/hardware_monitor.rs

@@ -2,20 +2,20 @@
 // License, v. 2.0. If a copy of the MPL was not distributed with this
 // file, You can obtain one at https://mozilla.org/MPL/2.0/.
 
-//! A task that listens for hardware events from the
+//! A task that listens for changes in the [`HardwareView`] from the
 //! [`sled_hardware::HardwareManager`] and dispatches them to other parts
 //! of the bootstrap agent and sled-agent code.
 
 use crate::services::ServiceManager;
 use crate::sled_agent::SledAgent;
 use sled_agent_config_reconciler::RawDisksSender;
 use sled_agent_types::debug::OperatorSwitchZonePolicy;
-use sled_hardware::{HardwareManager, HardwareUpdate};
+use sled_hardware::{HardwareManager, HardwareView};
 use sled_hardware_types::Baseboard;
 use sled_storage::disk::RawDisk;
 use slog::Logger;
-use tokio::sync::broadcast::error::RecvError;
-use tokio::sync::{broadcast, oneshot, watch};
+use tokio::sync::oneshot;
+use tokio::sync::watch;
 
 /// A handle controlling the behavior of a [`HardwareMonitor`]
 #[derive(Debug, Clone)]
@@ -52,15 +52,12 @@ pub struct HardwareMonitor {
     // Receive a onetime notification that the ServiceManager is ready
     service_manager_ready_rx: oneshot::Receiver<ServiceManager>,
 
-    // Receive messages from the [`HardwareManager`]
-    hardware_rx: broadcast::Receiver<HardwareUpdate>,
+    // Receive current view of hardware from the [`HardwareManager`]
+    hardware_view_rx: watch::Receiver<HardwareView>,
 
     // Receive the operator's policy controlling the switch zone
     switch_zone_policy_rx: watch::Receiver<OperatorSwitchZonePolicy>,
 
-    // A reference to the hardware manager
-    hardware_manager: HardwareManager,
-
     // A handle to send raw disk updates to the config-reconciler system.
     raw_disks_tx: RawDisksSender,
 
@@ -96,8 +93,8 @@ impl HardwareMonitor {
         let (sled_agent_started_tx, sled_agent_started_rx) = oneshot::channel();
         let (service_manager_ready_tx, service_manager_ready_rx) =
             oneshot::channel();
-        let baseboard = hardware_manager.baseboard();
-        let hardware_rx = hardware_manager.monitor();
+        let hardware_view_rx = hardware_manager.subscribe();
+        let baseboard = hardware_view_rx.borrow().baseboard();
         let log = log.new(o!("component" => "HardwareMonitor"));
         let (switch_zone_policy_tx, switch_zone_policy_rx) =
             watch::channel(OperatorSwitchZonePolicy::StartIfSwitchPresent);
@@ -106,9 +103,8 @@ impl HardwareMonitor {
             baseboard,
             sled_agent_started_rx,
             service_manager_ready_rx,
-            hardware_rx,
+            hardware_view_rx,
             switch_zone_policy_rx,
-            hardware_manager: hardware_manager.clone(),
             raw_disks_tx,
             sled_agent: None,
             service_manager: None,
@@ -150,14 +146,25 @@ impl HardwareMonitor {
                         policy,
                     ).await;
                 }
-                update = self.hardware_rx.recv() => {
-                    info!(
-                        self.log,
-                        "Received hardware update message";
-                        "update" => ?update,
-                    );
-                    self.handle_hardware_update(update.clone()).await
-            },
+                result = self.hardware_view_rx.changed() => {
+                    match result {
+                        Ok(()) => {
+                            info!(
+                                self.log,
+                                "Received notification hardware \
+                                 view has changed"
+                            );
+                            self.check_latest_hardware_snapshot().await;
+                        }
+                        Err(_recv_error) => {
+                            // The `HardwareManager` monitoring task is an
+                            // infinite loop - the only way for us to get
+                            // an error from `changed()` here is if it panicked,
+                            // so we will propagate such a panic.
+                            panic!("Hardware manager monitor task panicked");
+                        }
+                    }
+                }
                 Ok(()) = self.switch_zone_policy_rx.changed() => {
                     let policy = self.current_switch_zone_policy();
                     info!(
@@ -179,70 +186,6 @@ impl HardwareMonitor {
         *self.switch_zone_policy_rx.borrow_and_update()
     }
 
-    // Handle an update from the [`HardwareMonitor`]
-    async fn handle_hardware_update(
-        &mut self,
-        update: Result<HardwareUpdate, RecvError>,
-    ) {
-        match update {
-            Ok(update) => match update {
-                HardwareUpdate::TofinoAvailable => {
-                    info!(
-                        self.log,
-                        "Hardware monitor got TofinoAvailable message"
-                    );
-                    let policy = self.current_switch_zone_policy();
-                    self.ensure_switch_zone_activated_or_deactivated(
-                        true, policy,
-                    )
-                    .await
-                }
-                HardwareUpdate::TofinoUnavailable => {
-                    info!(
-                        self.log,
-                        "Hardware monitor got TofinoUnavailable message"
-                    );
-                    let policy = self.current_switch_zone_policy();
-                    self.ensure_switch_zone_activated_or_deactivated(
-                        false, policy,
-                    )
-                    .await
-                }
-                HardwareUpdate::TofinoDeviceChange => {
-                    info!(
-                        self.log,
-                        "Hardware monitor got TofinoDeviceChange message"
-                    );
-                    if let Some(sled_agent) = &mut self.sled_agent {
-                        sled_agent.notify_nexus_about_self(&self.log).await;
-                    }
-                }
-                HardwareUpdate::DiskAdded(disk) => {
-                    self.raw_disks_tx
-                        .add_or_update_raw_disk(disk.into(), &self.log);
-                }
-                HardwareUpdate::DiskRemoved(disk) => {
-                    self.raw_disks_tx
-                        .remove_raw_disk(disk.identity(), &self.log);
-                }
-                HardwareUpdate::DiskUpdated(disk) => {
-                    self.raw_disks_tx
-                        .add_or_update_raw_disk(disk.into(), &self.log);
-                }
-            },
-            Err(broadcast::error::RecvError::Lagged(count)) => {
-                warn!(self.log, "Hardware monitor missed {count} messages");
-                self.check_latest_hardware_snapshot().await;
-            }
-            Err(broadcast::error::RecvError::Closed) => {
-                // The `HardwareManager` monitoring task is an infinite loop -
-                // the only way for us to get `Closed` here is if it panicked,
-                // so we will propagate such a panic.
-                panic!("Hardware manager monitor task panicked");
-            }
-        }
-    }
-
     async fn ensure_switch_zone_activated_or_deactivated(
         &mut self,
         is_tofino_available: bool,
@@ -296,29 +239,29 @@ impl HardwareMonitor {
         }
     }
 
-    // Observe the current hardware state manually.
+    // Act on the current hardware snapshot.
     //
-    // We use this when we're monitoring hardware for the first
-    // time, and if we miss notifications.
+    // We use this on startup and any time the snapshot changes.
     async fn check_latest_hardware_snapshot(&mut self) {
         if let Some(sled_agent) = &self.sled_agent {
             sled_agent.notify_nexus_about_self(&self.log).await;
         }
 
+        let snapshot = self.hardware_view_rx.borrow_and_update().clone();
         info!(
             self.log, "Checking current full hardware snapshot";
-            "disks" => ?self.hardware_manager.disks(),
+            "snapshot" => ?snapshot,
         );
 
         let policy = self.current_switch_zone_policy();
         self.ensure_switch_zone_activated_or_deactivated(
-            self.hardware_manager.is_scrimlet_asic_available(),
+            snapshot.is_scrimlet_asic_available(),
             policy,
         )
         .await;
 
         self.raw_disks_tx.set_raw_disks(
-            self.hardware_manager.disks().into_values().map(RawDisk::from),
+            snapshot.into_disks().into_values().map(RawDisk::from),
             &self.log,
         );
     }