Extract ZfsTestHarness from StorageManagerTestHarness

Cargo.toml

@@ -138,6 +138,7 @@ members = [
     "sled-hardware",
     "sled-hardware/types",
     "sled-storage",
+    "sled-storage/zfs-test-harness",
     "sp-sim",
     "test-utils",
     "trust-quorum",
@@ -299,6 +300,7 @@ default-members = [
     "sled-hardware",
     "sled-hardware/types",
     "sled-storage",
+    "sled-storage/zfs-test-harness",
     "sp-sim",
     "trust-quorum",
     "trust-quorum/gfss",
@@ -783,6 +785,7 @@ wicketd-client = { path = "clients/wicketd-client" }
 xshell = "0.2.7"
 zerocopy = "0.8.26"
 zeroize = { version = "1.8.1", features = ["zeroize_derive", "std"] }
+zfs-test-harness = { path = "sled-storage/zfs-test-harness" }
 zip = { version = "4.2.0", default-features = false, features = ["deflate","bzip2"] }
 zone = { version = "0.3.1", default-features = false, features = ["async"] }
 

sled-agent/Cargo.toml

@@ -139,6 +139,7 @@ slog-term.workspace = true
 tempfile.workspace = true
 tokio-stream.workspace = true
 tokio-util.workspace = true
+zfs-test-harness.workspace = true
 
 illumos-utils = { workspace = true, features = ["testing"] }
 sled-agent-config-reconciler = { workspace = true, features = ["testing"] }

sled-agent/src/long_running_tasks.rs

@@ -5,8 +5,8 @@
 //! This module is responsible for spawning, starting, and managing long running
 //! tasks and task driven subsystems. These tasks run for the remainder of the
 //! sled-agent process from the moment they begin. Primarily they include the
-//! "managers", like `StorageManager`, `InstanceManager`, etc..., and are used
-//! by both the bootstrap agent and the sled-agent.
+//! "managers", like `KeyManager`, `ServiceManager`, etc., and are used by both
+//! the bootstrap agent and the sled-agent.
 //!
 //! We don't bother keeping track of the spawned tasks handles because we know
 //! these tasks are supposed to run forever, and they can shutdown if their

sled-agent/src/services.rs

@@ -4,9 +4,6 @@
 
 //! Sled-local service management.
 //!
-//! For controlling zone-based storage services, refer to
-//! [sled_storage::manager::StorageManager].
-//!
 //! For controlling virtual machine instances, refer to
 //! [crate::instance_manager::InstanceManager].
 //!

sled-agent/src/sim/storage.rs

@@ -1229,6 +1229,13 @@ impl Zpool {
 /// Represents a nested dataset
 pub struct NestedDatasetStorage {
     config: NestedDatasetConfig,
+    // In-memory flag for whether this dataset pretends to be mounted; defaults
+    // to true.
+    //
+    // Nothing in the simulated storage implementation acts on this value; it is
+    // merely a sticky bool that remembers the most recent value passed to
+    // `nested_dataset_set_mounted()`.
+    mounted: bool,
     // We intentionally store the children before the mountpoint,
     // so they are deleted first.
     children: BTreeMap<String, NestedDatasetStorage>,
@@ -1262,6 +1269,7 @@ impl NestedDatasetStorage {
 
         Self {
             config: NestedDatasetConfig { name, inner: shared_config },
+            mounted: true,
             children: BTreeMap::new(),
             mountpoint,
         }
@@ -1380,7 +1388,14 @@ impl StorageInner {
                 return Ok(DatasetProperties {
                     id: Some(*id),
                     name: dataset_name.to_string(),
-                    mounted: true,
+                    // We should have an entry in `self.nested_datasets` for
+                    // every entry in `config.datasets` (`datasets_ensure()`
+                    // keeps these in sync), but we only keeping track of a
+                    // `mounted` property on nested datasets. Look that up here.
+                    mounted: self
+                        .nested_datasets
+                        .get(&dataset.name)
+                        .map_or(true, |d| d.mounted),
                     avail: ByteCount::from_kibibytes_u32(1024),
                     used: ByteCount::from_kibibytes_u32(1024),
                     quota: dataset.inner.quota,
@@ -1399,7 +1414,7 @@ impl StorageInner {
                 return Ok(DatasetProperties {
                     id: None,
                     name: dataset_name.to_string(),
-                    mounted: true,
+                    mounted: nested_dataset_storage.mounted,
                     avail: ByteCount::from_kibibytes_u32(1024),
                     used: ByteCount::from_kibibytes_u32(1024),
                     quota: config.quota,
@@ -1517,6 +1532,64 @@ impl StorageInner {
         }
     }
 
+    #[cfg(test)]
+    pub fn nested_dataset_is_mounted(
+        &self,
+        dataset: &NestedDatasetLocation,
+    ) -> Result<bool, HttpError> {
+        let Some(mut nested_dataset) = self.nested_datasets.get(&dataset.root)
+        else {
+            return Err(HttpError::for_not_found(
+                None,
+                "Dataset not found".to_string(),
+            ));
+        };
+        for component in dataset.path.split('/') {
+            if component.is_empty() {
+                continue;
+            }
+            nested_dataset =
+                nested_dataset.children.get(component).ok_or_else(|| {
+                    HttpError::for_not_found(
+                        None,
+                        "Dataset not found".to_string(),
+                    )
+                })?;
+        }
+        Ok(nested_dataset.mounted)
+    }
+
+    pub fn nested_dataset_set_mounted(
+        &mut self,
+        dataset: &NestedDatasetLocation,
+        mounted: bool,
+    ) -> Result<(), HttpError> {
+        let Some(mut nested_dataset) =
+            self.nested_datasets.get_mut(&dataset.root)
+        else {
+            return Err(HttpError::for_not_found(
+                None,
+                "Dataset not found".to_string(),
+            ));
+        };
+        for component in dataset.path.split('/') {
+            if component.is_empty() {
+                continue;
+            }
+            nested_dataset = nested_dataset
+                .children
+                .get_mut(component)
+                .ok_or_else(|| {
+                    HttpError::for_not_found(
+                        None,
+                        "Dataset not found".to_string(),
+                    )
+                })?;
+        }
+        nested_dataset.mounted = mounted;
+        Ok(())
+    }
+
     pub fn nested_dataset_ensure(
         &mut self,
         config: NestedDatasetConfig,