Use measurements with sprockets (both bootstrap server and TQ)

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

@@ -12,14 +12,15 @@ use sled_agent_types::inventory::ConfigReconcilerInventoryStatus;
 use sled_agent_types::inventory::InventoryDataset;
 use sled_agent_types::inventory::InventoryDisk;
 use sled_agent_types::inventory::InventoryZpool;
+use sled_agent_types::inventory::OmicronSingleMeasurement;
 use sled_agent_types::inventory::OmicronSledConfig;
 use sled_storage::config::MountConfig;
 use sled_storage::disk::Disk;
 use sled_storage::nested_dataset::NestedDatasetConfig;
 use sled_storage::nested_dataset::NestedDatasetListOptions;
 use sled_storage::nested_dataset::NestedDatasetLocation;
 use slog::Logger;
-use std::collections::HashSet;
+use std::collections::{BTreeSet, HashSet};
 use std::sync::Arc;
 use std::sync::OnceLock;
 use tokio::sync::watch;
@@ -63,10 +64,15 @@ use crate::reconciler_task::CurrentlyManagedZpools;
 use crate::reconciler_task::CurrentlyManagedZpoolsReceiver;
 use crate::reconciler_task::ReconcilerResult;
 
+use crate::InternalDisks;
+use sled_agent_types::resolvable_files::ResolverStatus;
+
 #[derive(Debug, thiserror::Error)]
 pub enum InventoryError {
     #[error("ledger contents not yet available")]
     LedgerContentsNotAvailable,
+    #[error("waiting for ledger task to run")]
+    WaitingOnLedger,
     #[error("could not contact dataset task")]
     DatasetTaskError(#[from] DatasetTaskError),
     #[error("could not list dataset properties")]
@@ -92,6 +98,7 @@ pub struct ConfigReconcilerSpawnToken {
     raw_disks_rx: RawDisksReceiver,
     ledger_task_log: Logger,
     reconciler_task_log: Logger,
+    ledger_rx: Option<watch::Receiver<CurrentSledConfig>>,
 }
 
 #[derive(Debug)]
@@ -180,11 +187,16 @@ impl ConfigReconcilerHandle {
                     .new(slog::o!("component" => "SledConfigLedgerTask")),
                 reconciler_task_log: base_log
                     .new(slog::o!("component" => "ConfigReconcilerTask")),
+                ledger_rx: None,
             },
         )
     }
 
-    /// Spawn the primary config reconciliation task.
+    /// Pre-spawn the ledger task
+    ///
+    /// This is the first half of spawning the reconciliation task. We need to
+    /// spawn the ledger task early to allow for access to the ledger for
+    /// early measurement reconcilaition
     ///
     /// This method can effectively only be called once, because the caller must
     /// supply the `token` returned by `new()` when this handle was created.
@@ -193,15 +205,11 @@ impl ConfigReconcilerHandle {
     ///
     /// Panics if called multiple times, which is statically impossible outside
     /// shenanigans to get a second [`ConfigReconcilerSpawnToken`].
-    pub fn spawn_reconciliation_task<
-        T: SledAgentFacilities,
-        U: SledAgentArtifactStore + Clone,
-    >(
+    pub fn pre_spawn_reconciliation_task<U: SledAgentArtifactStore + Clone>(
         &self,
-        sled_agent_facilities: T,
         sled_agent_artifact_store: U,
         token: ConfigReconcilerSpawnToken,
-    ) {
+    ) -> ConfigReconcilerSpawnToken {
         let ConfigReconcilerSpawnToken {
             key_requester,
             time_sync_config,
@@ -212,14 +220,15 @@ impl ConfigReconcilerHandle {
             raw_disks_rx,
             ledger_task_log,
             reconciler_task_log,
+            ledger_rx: _,
         } = token;
 
         // Spawn the task that manages our config ledger.
         let (ledger_task, current_config_rx) =
             LedgerTaskHandle::spawn_ledger_task(
                 self.internal_disks_rx.clone(),
                 sled_agent_artifact_store.clone(),
-                ledger_task_log,
+                ledger_task_log.clone(),
             );
         match self.ledger_task.set(ledger_task) {
             Ok(()) => (),
@@ -233,12 +242,58 @@ impl ConfigReconcilerHandle {
             }
         }
 
+        ConfigReconcilerSpawnToken {
+            key_requester,
+            time_sync_config,
+            reconciler_result_tx,
+            currently_managed_zpools_tx,
+            external_disks_tx,
+            former_zone_root_archiver,
+            raw_disks_rx,
+            ledger_task_log,
+            reconciler_task_log,
+            ledger_rx: Some(current_config_rx),
+        }
+    }
+
+    /// Spawn the primary config reconciliation task.
+    ///
+    /// This method can effectively only be called once, because the caller must
+    /// supply the `token` returned by `new()` when this handle was created.
+    ///
+    /// # Panics
+    ///
+    /// Panics if called multiple times or if we haven't called the ledger setup
+    pub fn spawn_reconciliation_task<
+        T: SledAgentFacilities,
+        U: SledAgentArtifactStore + Clone,
+    >(
+        &self,
+        sled_agent_facilities: T,
+        sled_agent_artifact_store: U,
+        token: ConfigReconcilerSpawnToken,
+    ) {
+        let ConfigReconcilerSpawnToken {
+            key_requester,
+            time_sync_config,
+            reconciler_result_tx,
+            currently_managed_zpools_tx,
+            external_disks_tx,
+            former_zone_root_archiver,
+            raw_disks_rx,
+            ledger_task_log: _,
+            reconciler_task_log,
+            ledger_rx,
+        } = token;
+
+        let ledger_rx = ledger_rx.expect("Failed to call pre_spawn");
+
         reconciler_task::spawn(
             Arc::clone(self.internal_disks_rx.mount_config()),
             self.dataset_task.clone(),
             key_requester,
             time_sync_config,
-            current_config_rx,
+            ledger_rx,
             reconciler_result_tx,
             currently_managed_zpools_tx,
             self.internal_disks_rx.clone(),
@@ -351,6 +406,34 @@ impl ConfigReconcilerHandle {
             .await
     }
 
+    /// Run a first reconciliation of reference measurements on cold boot
+    pub async fn bootstrap_measurement_reconciler(
+        &self,
+        resolver_status: &ResolverStatus,
+        internal_disks: &InternalDisks,
+        desired: &BTreeSet<OmicronSingleMeasurement>,
+        log: &Logger,
+    ) -> Vec<Utf8PathBuf> {
+        let resolved = crate::measurements::reconcile_measurements(
+            resolver_status,
+            internal_disks,
+            desired,
+            log,
+        )
+        .await;
+        resolved.iter().filter_map(|entry| match &entry.result {
+            sled_agent_types::inventory::ConfigReconcilerInventoryResult::Ok => Some(entry.path.clone()),
+            sled_agent_types::inventory::ConfigReconcilerInventoryResult::Err { .. } => None,
+        }).collect()
+    }
+    /// Watch for changes to measurements
+    pub async fn measurement_corpus_rx(
+        &self,
+        pre_boot: Vec<Utf8PathBuf>,
+    ) -> MeasurementsReceiver {
+        MeasurementsReceiver::new(self.reconciler_result_rx.clone(), pre_boot)
+    }
+
     /// Return the currently-ledgered [`OmicronSledConfig`].
     ///
     /// # Errors
@@ -370,8 +453,9 @@ impl ConfigReconcilerHandle {
             // This shouldn't happen in practice: sled-agent should both wait
             // for the boot disk and spawn the reconciler task before starting
             // the dropshot server that allows Nexus to collect inventory.
-            None | Some(CurrentSledConfig::WaitingForInternalDisks) => {
-                Err(InventoryError::LedgerContentsNotAvailable)
+            None => Err(InventoryError::LedgerContentsNotAvailable),
+            Some(CurrentSledConfig::WaitingForInternalDisks) => {
+                Err(InventoryError::WaitingOnLedger)
             }
             Some(CurrentSledConfig::WaitingForInitialConfig) => Ok(None),
             Some(CurrentSledConfig::Ledgered(config)) => Ok(Some(*config)),
@@ -544,3 +628,40 @@ enum AvailableDatasetsReceiverInner {
     #[cfg(feature = "testing")]
     FakeStatic(Vec<(ZpoolName, Utf8PathBuf)>),
 }
+
+#[derive(Debug, Clone)]
+enum MeasurementsReceiverInner {
+    Real { rx: watch::Receiver<ReconcilerResult>, pre_boot: Vec<Utf8PathBuf> },
+    Fake(Vec<Utf8PathBuf>),
+}
+
+#[derive(Debug, Clone)]
+pub struct MeasurementsReceiver {
+    inner: MeasurementsReceiverInner,
+}
+
+impl MeasurementsReceiver {
+    pub fn new(
+        rx: watch::Receiver<ReconcilerResult>,
+        pre_boot: Vec<Utf8PathBuf>,
+    ) -> Self {
+        MeasurementsReceiver {
+            inner: MeasurementsReceiverInner::Real { rx, pre_boot },
+        }
+    }
+
+    pub fn new_fake(paths: Vec<Utf8PathBuf>) -> Self {
+        MeasurementsReceiver { inner: MeasurementsReceiverInner::Fake(paths) }
+    }
+    pub fn latest_measurements(&self) -> Vec<Utf8PathBuf> {
+        match &self.inner {
+            MeasurementsReceiverInner::Real { rx, pre_boot } => {
+                match rx.borrow().all_current_measurements() {
+                    either::Either::Left(_) => pre_boot.clone(),
+                    either::Either::Right(s) => s.collect(),
+                }
+            }
+            MeasurementsReceiverInner::Fake(paths) => paths.clone(),
+        }
+    }
+}

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

@@ -52,6 +52,7 @@ mod handle;
 mod host_phase_2;
 mod internal_disks;
 mod ledger;
+mod measurements;
 mod mupdate_override;
 mod raw_disks;
 mod reconciler_task;
@@ -66,6 +67,7 @@ pub use handle::AvailableDatasetsReceiver;
 pub use handle::ConfigReconcilerHandle;
 pub use handle::ConfigReconcilerSpawnToken;
 pub use handle::InventoryError;
+pub use handle::MeasurementsReceiver;
 pub use handle::ReconcilerInventory;
 pub use handle::TimeSyncConfig;
 pub use host_phase_2::HostPhase2PreparedContents;

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

@@ -0,0 +1,77 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// 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/.
+
+//! Measurements
+use crate::InternalDisks;
+use crate::mupdate_override::ResolverStatusExt;
+use camino::Utf8PathBuf;
+use iddqd::IdOrdMap;
+use sled_agent_types::inventory::ConfigReconcilerInventoryResult;
+use sled_agent_types::inventory::OmicronSingleMeasurement;
+use sled_agent_types::inventory::ReconciledSingleMeasurement;
+use sled_agent_types::resolvable_files::OmicronResolvableFileSource;
+use sled_agent_types::resolvable_files::ResolverStatus;
+use slog::Logger;
+use std::collections::BTreeSet;
+
+pub struct PreparedOmicronMeasurements {
+    pub sources: Vec<OmicronResolvableFileSource>,
+}
+
+impl PreparedOmicronMeasurements {
+    pub fn new(sources: Vec<OmicronResolvableFileSource>) -> Self {
+        Self { sources }
+    }
+
+    pub fn file_sources(&self) -> &Vec<OmicronResolvableFileSource> {
+        &self.sources
+    }
+}
+
+pub(crate) async fn reconcile_measurements(
+    resolver_status: &ResolverStatus,
+    internal_disks: &InternalDisks,
+    desired: &BTreeSet<OmicronSingleMeasurement>,
+    log: &Logger,
+) -> IdOrdMap<ReconciledSingleMeasurement> {
+    let set =
+        resolver_status.prepare_all_measurements(log, desired, internal_disks);
+
+    let mut unique = IdOrdMap::new();
+
+    for entry in set.sources {
+        let mut found = false;
+        for path in entry.file_source.search_paths {
+            let full_path = path.join(entry.file_source.file_name.clone());
+            if let Ok(exists) = tokio::fs::try_exists(&full_path).await {
+                if exists {
+                    unique
+                        .insert_unique(ReconciledSingleMeasurement {
+                            file_name: entry.file_source.file_name.clone(),
+                            path: full_path,
+                            result: ConfigReconcilerInventoryResult::Ok,
+                        })
+                        .expect("file names should be unique");
+                    found = true;
+                    break;
+                }
+            }
+        }
+        if !found {
+            unique
+                .insert_unique(ReconciledSingleMeasurement {
+                    file_name: entry.file_source.file_name.clone(),
+                    path: Utf8PathBuf::new(),
+                    result: ConfigReconcilerInventoryResult::Err {
+                        message:
+                            "The measurement file does not exist in any path"
+                                .to_string(),
+                    },
+                })
+                .expect("file names should be unique");
+        }
+    }
+
+    unique
+}