Add ZFS storage engine: configurable pool, dataset hierarchy, extensible trait

Decouple storage from the ZoneRuntime trait into a dedicated StorageEngine
trait with ZfsStorageEngine (illumos) and MockStorageEngine (testing)
implementations. Replace the per-zone ZfsConfig with a global
StoragePoolConfig that derives dataset hierarchy from a single --storage-pool
flag, with optional per-dataset overrides. This enables persistent volumes,
auto-created base datasets on startup, and a clean extension point for
future storage backends.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

crates/reddwarf-runtime/src/brand/lx.rs

@@ -30,11 +30,7 @@ mod tests {
                 gateway: "10.0.0.1".to_string(),
                 prefix_len: 16,
             }),
-            zfs: ZfsConfig {
-                parent_dataset: "rpool/zones".to_string(),
-                clone_from: None,
-                quota: None,
-            },
+            storage: ZoneStorageOpts::default(),
             lx_image_path: image_path,
             processes: vec![],
             cpu_cap: None,

crates/reddwarf-runtime/src/controller.rs

@@ -19,8 +19,6 @@ pub struct PodControllerConfig {
     pub api_url: String,
     /// Prefix for zone root paths (e.g., "/zones")
     pub zonepath_prefix: String,
-    /// Parent ZFS dataset (e.g., "rpool/zones")
-    pub zfs_parent_dataset: String,
     /// Default zone brand
     pub default_brand: ZoneBrand,
     /// Name of the etherstub for pod networking
@@ -444,11 +442,7 @@ impl PodController {
             brand: self.config.default_brand.clone(),
             zonepath,
             network,
-            zfs: ZfsConfig {
-                parent_dataset: self.config.zfs_parent_dataset.clone(),
-                clone_from: None,
-                quota: None,
-            },
+            storage: ZoneStorageOpts::default(),
             lx_image_path: None,
             processes,
             cpu_cap: None,
@@ -504,7 +498,10 @@ mod tests {
         let storage = Arc::new(RedbBackend::new(&db_path).unwrap());
         let ipam = Ipam::new(storage, "10.88.0.0/16").unwrap();
 
-        let runtime = Arc::new(crate::mock::MockRuntime::new());
+        let mock_storage = Arc::new(crate::storage::MockStorageEngine::new(
+            crate::types::StoragePoolConfig::from_pool("rpool"),
+        ));
+        let runtime = Arc::new(crate::mock::MockRuntime::new(mock_storage));
         let api_client = Arc::new(ApiClient::new("http://127.0.0.1:6443"));
         let (event_tx, _) = broadcast::channel(16);
 
@@ -512,7 +509,6 @@ mod tests {
             node_name: "node1".to_string(),
             api_url: "http://127.0.0.1:6443".to_string(),
             zonepath_prefix: "/zones".to_string(),
-            zfs_parent_dataset: "rpool/zones".to_string(),
             default_brand: ZoneBrand::Reddwarf,
             etherstub_name: "reddwarf0".to_string(),
             pod_cidr: "10.88.0.0/16".to_string(),
@@ -580,7 +576,6 @@ mod tests {
         assert_eq!(zone_config.processes[1].name, "sidecar");
         assert_eq!(zone_config.processes[1].command, vec!["/bin/sh", "-c"]);
         assert_eq!(zone_config.brand, ZoneBrand::Reddwarf);
-        assert_eq!(zone_config.zfs.parent_dataset, "rpool/zones");
 
         // Verify per-pod networking
         match &zone_config.network {

crates/reddwarf-runtime/src/error.rs

@@ -131,6 +131,19 @@ pub enum RuntimeError {
         cidr: String,
     },
 
+    /// Storage initialization failed
+    #[error("Storage initialization failed: {message}")]
+    #[diagnostic(
+        code(reddwarf::runtime::storage_init_failed),
+        help("Verify the ZFS pool '{pool}' exists and you have permission to create datasets. Run: zpool list")
+    )]
+    StorageInitFailed {
+        #[allow(unused)]
+        pool: String,
+        #[allow(unused)]
+        message: String,
+    },
+
     /// Internal error
     #[error("Internal runtime error: {message}")]
     #[diagnostic(

crates/reddwarf-runtime/src/illumos.rs

@@ -1,28 +1,26 @@
 use crate::brand::lx::lx_install_args;
 use crate::command::exec;
-use crate::error::{Result, RuntimeError};
+use crate::error::Result;
+use crate::storage::StorageEngine;
 use crate::traits::ZoneRuntime;
 use crate::types::*;
-use crate::zfs;
 use crate::zone::config::generate_zonecfg;
 use crate::zone::state::parse_zoneadm_line;
 use async_trait::async_trait;
+use std::sync::Arc;
 use tracing::info;
 
 /// illumos zone runtime implementation
 ///
-/// Manages real zones via zonecfg/zoneadm, dladm for networking, and zfs for storage.
-pub struct IllumosRuntime;
+/// Manages real zones via zonecfg/zoneadm, dladm for networking.
+/// Storage (ZFS datasets) is delegated to the injected `StorageEngine`.
+pub struct IllumosRuntime {
+    storage: Arc<dyn StorageEngine>,
+}
 
 impl IllumosRuntime {
-    pub fn new() -> Self {
-        Self
-    }
-}
-
-impl Default for IllumosRuntime {
-    fn default() -> Self {
-        Self::new()
+    pub fn new(storage: Arc<dyn StorageEngine>) -> Self {
+        Self { storage }
     }
 }
 
@@ -37,7 +35,9 @@ impl ZoneRuntime for IllumosRuntime {
         let tmp_path = format!("/tmp/zonecfg-{}.cmd", config.zone_name);
         tokio::fs::write(&tmp_path, &zonecfg_content)
             .await
-            .map_err(|e| RuntimeError::zone_operation_failed(&config.zone_name, e.to_string()))?;
+            .map_err(|e| {
+                crate::error::RuntimeError::zone_operation_failed(&config.zone_name, e.to_string())
+            })?;
 
         let result = exec("zonecfg", &["-z", &config.zone_name, "-f", &tmp_path]).await;
 
@@ -166,47 +166,12 @@ impl ZoneRuntime for IllumosRuntime {
         Ok(())
     }
 
-    async fn create_zfs_dataset(&self, zone_name: &str, config: &ZoneConfig) -> Result<()> {
-        info!("Creating ZFS dataset for zone: {}", zone_name);
-
-        let dataset = zfs::dataset_path(&config.zfs, zone_name);
-
-        if let Some(ref clone_from) = config.zfs.clone_from {
-            // Fast clone path
-            exec("zfs", &["clone", clone_from, &dataset]).await?;
-        } else {
-            exec("zfs", &["create", &dataset]).await?;
-        }
-
-        if let Some(ref quota) = config.zfs.quota {
-            exec("zfs", &["set", &format!("quota={}", quota), &dataset]).await?;
-        }
-
-        info!("ZFS dataset created: {}", dataset);
-        Ok(())
-    }
-
-    async fn destroy_zfs_dataset(&self, zone_name: &str, config: &ZoneConfig) -> Result<()> {
-        info!("Destroying ZFS dataset for zone: {}", zone_name);
-
-        let dataset = zfs::dataset_path(&config.zfs, zone_name);
-        exec("zfs", &["destroy", "-r", &dataset]).await?;
-
-        info!("ZFS dataset destroyed: {}", dataset);
-        Ok(())
-    }
-
-    async fn create_snapshot(&self, dataset: &str, snapshot_name: &str) -> Result<()> {
-        let snap = format!("{}@{}", dataset, snapshot_name);
-        exec("zfs", &["snapshot", &snap]).await?;
-        info!("ZFS snapshot created: {}", snap);
-        Ok(())
-    }
-
     async fn provision(&self, config: &ZoneConfig) -> Result<()> {
         info!("Provisioning zone: {}", config.zone_name);
 
-        self.create_zfs_dataset(&config.zone_name, config).await?;
+        self.storage
+            .create_zone_dataset(&config.zone_name, &config.storage)
+            .await?;
         self.setup_network(&config.zone_name, &config.network)
             .await?;
         self.create_zone(config).await?;
@@ -237,7 +202,7 @@ impl ZoneRuntime for IllumosRuntime {
         self.delete_zone(&config.zone_name).await?;
         self.teardown_network(&config.zone_name, &config.network)
             .await?;
-        self.destroy_zfs_dataset(&config.zone_name, config).await?;
+        self.storage.destroy_zone_dataset(&config.zone_name).await?;
 
         info!("Zone deprovisioned: {}", config.zone_name);
         Ok(())

crates/reddwarf-runtime/src/lib.rs

@@ -11,9 +11,9 @@ pub mod illumos;
 pub mod mock;
 pub mod network;
 pub mod node_agent;
+pub mod storage;
 pub mod traits;
 pub mod types;
-pub mod zfs;
 pub mod zone;
 
 // Re-export primary types
@@ -22,10 +22,15 @@ pub use mock::MockRuntime;
 pub use network::{CidrConfig, IpAllocation, Ipam};
 pub use traits::ZoneRuntime;
 pub use types::{
-    ContainerProcess, DirectNicConfig, EtherstubConfig, FsMount, NetworkMode, ZfsConfig, ZoneBrand,
-    ZoneConfig, ZoneInfo, ZoneState,
+    ContainerProcess, DirectNicConfig, EtherstubConfig, FsMount, NetworkMode, StoragePoolConfig,
+    ZoneBrand, ZoneConfig, ZoneInfo, ZoneState, ZoneStorageOpts,
 };
 
+// Re-export storage types
+#[cfg(target_os = "illumos")]
+pub use storage::ZfsStorageEngine;
+pub use storage::{MockStorageEngine, StorageEngine, VolumeInfo};
+
 // Re-export controller and agent types
 pub use api_client::ApiClient;
 pub use controller::{PodController, PodControllerConfig};

crates/reddwarf-runtime/src/mock.rs

@@ -1,4 +1,5 @@
 use crate::error::{Result, RuntimeError};
+use crate::storage::StorageEngine;
 use crate::traits::ZoneRuntime;
 use crate::types::*;
 use async_trait::async_trait;
@@ -18,27 +19,24 @@ struct MockZone {
 /// Mock runtime for testing on non-illumos platforms
 ///
 /// Maintains an in-memory zone registry and simulates state transitions.
-/// All network/ZFS operations are no-ops.
+/// All network operations are no-ops. Storage operations are delegated to
+/// the injected `StorageEngine`.
 pub struct MockRuntime {
     zones: Arc<RwLock<HashMap<String, MockZone>>>,
     next_id: Arc<RwLock<i32>>,
+    storage: Arc<dyn StorageEngine>,
 }
 
 impl MockRuntime {
-    pub fn new() -> Self {
+    pub fn new(storage: Arc<dyn StorageEngine>) -> Self {
         Self {
             zones: Arc::new(RwLock::new(HashMap::new())),
             next_id: Arc::new(RwLock::new(1)),
+            storage,
         }
     }
 }
 
-impl Default for MockRuntime {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 #[async_trait]
 impl ZoneRuntime for MockRuntime {
     async fn create_zone(&self, config: &ZoneConfig) -> Result<()> {
@@ -235,23 +233,10 @@ impl ZoneRuntime for MockRuntime {
         Ok(())
     }
 
-    async fn create_zfs_dataset(&self, zone_name: &str, _config: &ZoneConfig) -> Result<()> {
-        debug!("Mock: ZFS dataset created for zone: {}", zone_name);
-        Ok(())
-    }
-
-    async fn destroy_zfs_dataset(&self, zone_name: &str, _config: &ZoneConfig) -> Result<()> {
-        debug!("Mock: ZFS dataset destroyed for zone: {}", zone_name);
-        Ok(())
-    }
-
-    async fn create_snapshot(&self, dataset: &str, snapshot_name: &str) -> Result<()> {
-        debug!("Mock: ZFS snapshot created: {}@{}", dataset, snapshot_name);
-        Ok(())
-    }
-
     async fn provision(&self, config: &ZoneConfig) -> Result<()> {
-        self.create_zfs_dataset(&config.zone_name, config).await?;
+        self.storage
+            .create_zone_dataset(&config.zone_name, &config.storage)
+            .await?;
         self.setup_network(&config.zone_name, &config.network)
             .await?;
         self.create_zone(config).await?;
@@ -293,7 +278,7 @@ impl ZoneRuntime for MockRuntime {
 
         self.teardown_network(&config.zone_name, &config.network)
             .await?;
-        self.destroy_zfs_dataset(&config.zone_name, config).await?;
+        self.storage.destroy_zone_dataset(&config.zone_name).await?;
         Ok(())
     }
 }
@@ -301,6 +286,14 @@ impl ZoneRuntime for MockRuntime {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::storage::MockStorageEngine;
+    use crate::types::StoragePoolConfig;
+
+    fn make_test_storage() -> Arc<dyn StorageEngine> {
+        Arc::new(MockStorageEngine::new(StoragePoolConfig::from_pool(
+            "rpool",
+        )))
+    }
 
     fn make_test_config(name: &str) -> ZoneConfig {
         ZoneConfig {
@@ -314,11 +307,7 @@ mod tests {
                 gateway: "10.0.0.1".to_string(),
                 prefix_len: 16,
             }),
-            zfs: ZfsConfig {
-                parent_dataset: "rpool/zones".to_string(),
-                clone_from: None,
-                quota: None,
-            },
+            storage: ZoneStorageOpts::default(),
             lx_image_path: None,
             processes: vec![],
             cpu_cap: None,
@@ -329,7 +318,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_provision_transitions_to_running() {
-        let rt = MockRuntime::new();
+        let rt = MockRuntime::new(make_test_storage());
         let config = make_test_config("test-zone");
 
         rt.provision(&config).await.unwrap();
@@ -340,7 +329,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_deprovision_removes_zone() {
-        let rt = MockRuntime::new();
+        let rt = MockRuntime::new(make_test_storage());
         let config = make_test_config("test-zone");
 
         rt.provision(&config).await.unwrap();
@@ -352,7 +341,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_duplicate_create_zone_returns_error() {
-        let rt = MockRuntime::new();
+        let rt = MockRuntime::new(make_test_storage());
         let config = make_test_config("test-zone");
 
         rt.create_zone(&config).await.unwrap();
@@ -365,7 +354,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_ops_on_missing_zone_return_not_found() {
-        let rt = MockRuntime::new();
+        let rt = MockRuntime::new(make_test_storage());
 
         assert!(matches!(
             rt.get_zone_state("nonexistent").await.unwrap_err(),
@@ -383,7 +372,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_list_zones_returns_all_provisioned() {
-        let rt = MockRuntime::new();
+        let rt = MockRuntime::new(make_test_storage());
 
         for i in 0..3 {
             let config = make_test_config(&format!("zone-{}", i));
@@ -396,7 +385,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_zone_info() {
-        let rt = MockRuntime::new();
+        let rt = MockRuntime::new(make_test_storage());
         let config = make_test_config("info-zone");
 
         rt.provision(&config).await.unwrap();

crates/reddwarf-runtime/src/storage/mock.rs

@@ -0,0 +1,151 @@
+use crate::error::Result;
+use crate::storage::{StorageEngine, VolumeInfo};
+use crate::types::{StoragePoolConfig, ZoneStorageOpts};
+use async_trait::async_trait;
+use std::collections::HashSet;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+use tracing::debug;
+
+/// In-memory storage engine for testing on non-illumos platforms
+///
+/// Tracks dataset names in memory so tests can assert which datasets
+/// were created/destroyed without touching a real ZFS pool.
+pub struct MockStorageEngine {
+    config: StoragePoolConfig,
+    datasets: Arc<RwLock<HashSet<String>>>,
+}
+
+impl MockStorageEngine {
+    pub fn new(config: StoragePoolConfig) -> Self {
+        Self {
+            config,
+            datasets: Arc::new(RwLock::new(HashSet::new())),
+        }
+    }
+}
+
+#[async_trait]
+impl StorageEngine for MockStorageEngine {
+    async fn initialize(&self) -> Result<()> {
+        let mut ds = self.datasets.write().await;
+        ds.insert(self.config.zones_dataset.clone());
+        ds.insert(self.config.images_dataset.clone());
+        ds.insert(self.config.volumes_dataset.clone());
+        debug!("Mock: initialized storage pool '{}'", self.config.pool);
+        Ok(())
+    }
+
+    async fn create_zone_dataset(&self, zone_name: &str, _opts: &ZoneStorageOpts) -> Result<()> {
+        let dataset = self.config.zone_dataset(zone_name);
+        self.datasets.write().await.insert(dataset.clone());
+        debug!("Mock: created zone dataset {}", dataset);
+        Ok(())
+    }
+
+    async fn destroy_zone_dataset(&self, zone_name: &str) -> Result<()> {
+        let dataset = self.config.zone_dataset(zone_name);
+        self.datasets.write().await.remove(&dataset);
+        debug!("Mock: destroyed zone dataset {}", dataset);
+        Ok(())
+    }
+
+    async fn create_snapshot(&self, dataset: &str, snapshot_name: &str) -> Result<()> {
+        let snap = format!("{}@{}", dataset, snapshot_name);
+        debug!("Mock: created snapshot {}", snap);
+        Ok(())
+    }
+
+    async fn create_volume(&self, name: &str, _quota: Option<&str>) -> Result<()> {
+        let dataset = self.config.volume_dataset(name);
+        self.datasets.write().await.insert(dataset.clone());
+        debug!("Mock: created volume {}", dataset);
+        Ok(())
+    }
+
+    async fn destroy_volume(&self, name: &str) -> Result<()> {
+        let dataset = self.config.volume_dataset(name);
+        self.datasets.write().await.remove(&dataset);
+        debug!("Mock: destroyed volume {}", dataset);
+        Ok(())
+    }
+
+    async fn list_volumes(&self) -> Result<Vec<VolumeInfo>> {
+        let ds = self.datasets.read().await;
+        let prefix = format!("{}/", self.config.volumes_dataset);
+        let volumes = ds
+            .iter()
+            .filter(|d| d.starts_with(&prefix))
+            .map(|d| {
+                let name = d.strip_prefix(&prefix).unwrap_or(d).to_string();
+                VolumeInfo {
+                    name,
+                    dataset: d.clone(),
+                    quota: None,
+                }
+            })
+            .collect();
+        Ok(volumes)
+    }
+
+    fn pool_config(&self) -> &StoragePoolConfig {
+        &self.config
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_mock_initialize_creates_base_datasets() {
+        let config = StoragePoolConfig::from_pool("testpool");
+        let engine = MockStorageEngine::new(config);
+
+        engine.initialize().await.unwrap();
+
+        let ds = engine.datasets.read().await;
+        assert!(ds.contains("testpool/zones"));
+        assert!(ds.contains("testpool/images"));
+        assert!(ds.contains("testpool/volumes"));
+    }
+
+    #[tokio::test]
+    async fn test_mock_zone_dataset_lifecycle() {
+        let config = StoragePoolConfig::from_pool("testpool");
+        let engine = MockStorageEngine::new(config);
+
+        engine
+            .create_zone_dataset("myzone", &ZoneStorageOpts::default())
+            .await
+            .unwrap();
+        assert!(engine
+            .datasets
+            .read()
+            .await
+            .contains("testpool/zones/myzone"));
+
+        engine.destroy_zone_dataset("myzone").await.unwrap();
+        assert!(!engine
+            .datasets
+            .read()
+            .await
+            .contains("testpool/zones/myzone"));
+    }
+
+    #[tokio::test]
+    async fn test_mock_volume_lifecycle() {
+        let config = StoragePoolConfig::from_pool("testpool");
+        let engine = MockStorageEngine::new(config);
+        engine.initialize().await.unwrap();
+
+        engine.create_volume("data-vol", None).await.unwrap();
+        let vols = engine.list_volumes().await.unwrap();
+        assert_eq!(vols.len(), 1);
+        assert_eq!(vols[0].name, "data-vol");
+
+        engine.destroy_volume("data-vol").await.unwrap();
+        let vols = engine.list_volumes().await.unwrap();
+        assert!(vols.is_empty());
+    }
+}

crates/reddwarf-runtime/src/storage/mod.rs

@@ -0,0 +1,52 @@
+mod mock;
+#[cfg(target_os = "illumos")]
+mod zfs;
+
+pub use mock::MockStorageEngine;
+#[cfg(target_os = "illumos")]
+pub use zfs::ZfsStorageEngine;
+
+use crate::error::Result;
+use crate::types::{StoragePoolConfig, ZoneStorageOpts};
+use async_trait::async_trait;
+
+/// Information about a persistent volume
+#[derive(Debug, Clone)]
+pub struct VolumeInfo {
+    pub name: String,
+    pub dataset: String,
+    pub quota: Option<String>,
+}
+
+/// Trait for pluggable storage backends
+///
+/// The default (and currently only real) implementation is `ZfsStorageEngine`,
+/// which manages ZFS datasets for zone root filesystems, images, and
+/// persistent volumes. `MockStorageEngine` provides an in-memory backend
+/// for testing on non-illumos platforms.
+#[async_trait]
+pub trait StorageEngine: Send + Sync {
+    /// Ensure all base datasets exist. Called once at startup.
+    async fn initialize(&self) -> Result<()>;
+
+    /// Create a dataset for a zone, applying per-zone options (clone_from, quota).
+    async fn create_zone_dataset(&self, zone_name: &str, opts: &ZoneStorageOpts) -> Result<()>;
+
+    /// Destroy a zone's dataset (recursive).
+    async fn destroy_zone_dataset(&self, zone_name: &str) -> Result<()>;
+
+    /// Create a ZFS snapshot.
+    async fn create_snapshot(&self, dataset: &str, snapshot_name: &str) -> Result<()>;
+
+    /// Create a persistent volume (ZFS dataset under volumes_dataset).
+    async fn create_volume(&self, name: &str, quota: Option<&str>) -> Result<()>;
+
+    /// Destroy a persistent volume.
+    async fn destroy_volume(&self, name: &str) -> Result<()>;
+
+    /// List all persistent volumes.
+    async fn list_volumes(&self) -> Result<Vec<VolumeInfo>>;
+
+    /// Get the pool configuration.
+    fn pool_config(&self) -> &StoragePoolConfig;
+}

crates/reddwarf-runtime/src/storage/zfs.rs

@@ -0,0 +1,161 @@
+use crate::command::{exec, exec_unchecked};
+use crate::error::{Result, RuntimeError};
+use crate::storage::{StorageEngine, VolumeInfo};
+use crate::types::{StoragePoolConfig, ZoneStorageOpts};
+use async_trait::async_trait;
+use tracing::info;
+
+/// ZFS-backed storage engine for illumos
+///
+/// Manages zone root filesystems, container images, and persistent volumes
+/// as ZFS datasets under the configured pool hierarchy.
+pub struct ZfsStorageEngine {
+    config: StoragePoolConfig,
+}
+
+impl ZfsStorageEngine {
+    pub fn new(config: StoragePoolConfig) -> Self {
+        Self { config }
+    }
+}
+
+#[async_trait]
+impl StorageEngine for ZfsStorageEngine {
+    async fn initialize(&self) -> Result<()> {
+        info!("Initializing ZFS storage pool '{}'", self.config.pool);
+
+        // Create base datasets (ignore already-exists errors via exec_unchecked)
+        for dataset in [
+            &self.config.zones_dataset,
+            &self.config.images_dataset,
+            &self.config.volumes_dataset,
+        ] {
+            let output = exec_unchecked("zfs", &["create", "-p", dataset]).await?;
+            if output.exit_code != 0 && !output.stderr.contains("dataset already exists") {
+                return Err(RuntimeError::StorageInitFailed {
+                    pool: self.config.pool.clone(),
+                    message: format!(
+                        "Failed to create dataset '{}': {}",
+                        dataset,
+                        output.stderr.trim()
+                    ),
+                });
+            }
+        }
+
+        info!(
+            "ZFS storage initialized: zones={}, images={}, volumes={}",
+            self.config.zones_dataset, self.config.images_dataset, self.config.volumes_dataset
+        );
+        Ok(())
+    }
+
+    async fn create_zone_dataset(&self, zone_name: &str, opts: &ZoneStorageOpts) -> Result<()> {
+        let dataset = self.config.zone_dataset(zone_name);
+        info!("Creating ZFS dataset for zone: {}", dataset);
+
+        if let Some(ref clone_from) = opts.clone_from {
+            exec("zfs", &["clone", clone_from, &dataset]).await?;
+        } else {
+            exec("zfs", &["create", &dataset]).await?;
+        }
+
+        if let Some(ref quota) = opts.quota {
+            exec("zfs", &["set", &format!("quota={}", quota), &dataset]).await?;
+        }
+
+        info!("ZFS dataset created: {}", dataset);
+        Ok(())
+    }
+
+    async fn destroy_zone_dataset(&self, zone_name: &str) -> Result<()> {
+        let dataset = self.config.zone_dataset(zone_name);
+        info!("Destroying ZFS dataset: {}", dataset);
+        exec("zfs", &["destroy", "-r", &dataset]).await?;
+        info!("ZFS dataset destroyed: {}", dataset);
+        Ok(())
+    }
+
+    async fn create_snapshot(&self, dataset: &str, snapshot_name: &str) -> Result<()> {
+        let snap = format!("{}@{}", dataset, snapshot_name);
+        exec("zfs", &["snapshot", &snap]).await?;
+        info!("ZFS snapshot created: {}", snap);
+        Ok(())
+    }
+
+    async fn create_volume(&self, name: &str, quota: Option<&str>) -> Result<()> {
+        let dataset = self.config.volume_dataset(name);
+        info!("Creating persistent volume: {}", dataset);
+        exec("zfs", &["create", &dataset]).await?;
+
+        if let Some(q) = quota {
+            exec("zfs", &["set", &format!("quota={}", q), &dataset]).await?;
+        }
+
+        info!("Persistent volume created: {}", dataset);
+        Ok(())
+    }
+
+    async fn destroy_volume(&self, name: &str) -> Result<()> {
+        let dataset = self.config.volume_dataset(name);
+        info!("Destroying persistent volume: {}", dataset);
+        exec("zfs", &["destroy", "-r", &dataset]).await?;
+        info!("Persistent volume destroyed: {}", dataset);
+        Ok(())
+    }
+
+    async fn list_volumes(&self) -> Result<Vec<VolumeInfo>> {
+        let output = exec(
+            "zfs",
+            &[
+                "list",
+                "-r",
+                "-H",
+                "-o",
+                "name,quota",
+                &self.config.volumes_dataset,
+            ],
+        )
+        .await?;
+
+        let mut volumes = Vec::new();
+        for line in output.stdout.lines() {
+            let line = line.trim();
+            if line.is_empty() {
+                continue;
+            }
+            let parts: Vec<&str> = line.split('\t').collect();
+            let dataset = parts[0];
+
+            // Skip the parent dataset itself
+            if dataset == self.config.volumes_dataset {
+                continue;
+            }
+
+            let name = dataset
+                .strip_prefix(&format!("{}/", self.config.volumes_dataset))
+                .unwrap_or(dataset)
+                .to_string();
+
+            let quota = parts.get(1).and_then(|q| {
+                if *q == "none" {
+                    None
+                } else {
+                    Some(q.to_string())
+                }
+            });
+
+            volumes.push(VolumeInfo {
+                name,
+                dataset: dataset.to_string(),
+                quota,
+            });
+        }
+
+        Ok(volumes)
+    }
+
+    fn pool_config(&self) -> &StoragePoolConfig {
+        &self.config
+    }
+}

crates/reddwarf-runtime/src/traits.rs

@@ -4,8 +4,12 @@ use async_trait::async_trait;
 
 /// Trait for zone runtime implementations
 ///
-/// This trait abstracts over the illumos zone lifecycle, networking, and ZFS
+/// This trait abstracts over the illumos zone lifecycle and networking
 /// operations. It enables testing via `MockRuntime` on non-illumos platforms.
+///
+/// Storage operations (ZFS dataset create/destroy, snapshots, volumes) are
+/// handled by the separate `StorageEngine` trait, which is injected into
+/// runtime implementations.
 #[async_trait]
 pub trait ZoneRuntime: Send + Sync {
     // --- Zone lifecycle ---
@@ -50,17 +54,6 @@ pub trait ZoneRuntime: Send + Sync {
     /// Tear down network for a zone
     async fn teardown_network(&self, zone_name: &str, network: &NetworkMode) -> Result<()>;
 
-    // --- ZFS ---
-
-    /// Create a ZFS dataset for a zone
-    async fn create_zfs_dataset(&self, zone_name: &str, config: &ZoneConfig) -> Result<()>;
-
-    /// Destroy a ZFS dataset for a zone
-    async fn destroy_zfs_dataset(&self, zone_name: &str, config: &ZoneConfig) -> Result<()>;
-
-    /// Create a ZFS snapshot
-    async fn create_snapshot(&self, dataset: &str, snapshot_name: &str) -> Result<()>;
-
     // --- High-level lifecycle ---
 
     /// Full provisioning: create dataset -> setup network -> create zone -> install -> boot

crates/reddwarf-runtime/src/types.rs

@@ -123,14 +123,69 @@ pub struct DirectNicConfig {
     pub prefix_len: u8,
 }
 
-/// ZFS dataset configuration for zone storage
+/// Global storage pool configuration
+///
+/// Derived from a single `--storage-pool` flag (e.g., "rpool"), with optional
+/// per-dataset overrides via `--zones-dataset`, `--images-dataset`, `--volumes-dataset`.
 #[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct ZfsConfig {
-    /// Parent dataset (e.g., "rpool/zones")
-    pub parent_dataset: String,
+pub struct StoragePoolConfig {
+    /// Base pool name (e.g., "rpool" or "datapool")
+    pub pool: String,
+    /// Dataset for zone root filesystems (default: "{pool}/zones")
+    pub zones_dataset: String,
+    /// Dataset for container images (default: "{pool}/images")
+    pub images_dataset: String,
+    /// Dataset for persistent volumes (default: "{pool}/volumes")
+    pub volumes_dataset: String,
+}
+
+impl StoragePoolConfig {
+    /// Create config from a pool name, auto-deriving child datasets
+    pub fn from_pool(pool: &str) -> Self {
+        Self {
+            pool: pool.to_string(),
+            zones_dataset: format!("{}/zones", pool),
+            images_dataset: format!("{}/images", pool),
+            volumes_dataset: format!("{}/volumes", pool),
+        }
+    }
+
+    /// Apply optional overrides for individual datasets
+    pub fn with_overrides(
+        mut self,
+        zones: Option<&str>,
+        images: Option<&str>,
+        volumes: Option<&str>,
+    ) -> Self {
+        if let Some(z) = zones {
+            self.zones_dataset = z.to_string();
+        }
+        if let Some(i) = images {
+            self.images_dataset = i.to_string();
+        }
+        if let Some(v) = volumes {
+            self.volumes_dataset = v.to_string();
+        }
+        self
+    }
+
+    /// Derive the full dataset path for a zone
+    pub fn zone_dataset(&self, zone_name: &str) -> String {
+        format!("{}/{}", self.zones_dataset, zone_name)
+    }
+
+    /// Derive the full dataset path for a volume
+    pub fn volume_dataset(&self, volume_name: &str) -> String {
+        format!("{}/{}", self.volumes_dataset, volume_name)
+    }
+}
+
+/// Per-zone storage options (replaces the old ZfsConfig on ZoneConfig)
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct ZoneStorageOpts {
     /// Optional snapshot to clone from (fast provisioning)
     pub clone_from: Option<String>,
-    /// Optional quota
+    /// Optional quota (e.g., "10G")
     pub quota: Option<String>,
 }
 
@@ -171,8 +226,8 @@ pub struct ZoneConfig {
     pub zonepath: String,
     /// Network configuration
     pub network: NetworkMode,
-    /// ZFS dataset configuration
-    pub zfs: ZfsConfig,
+    /// Per-zone storage options (clone source, quota)
+    pub storage: ZoneStorageOpts,
     /// LX brand image path (only for Lx brand)
     pub lx_image_path: Option<String>,
     /// Supervised processes (for reddwarf brand)

crates/reddwarf-runtime/src/zfs/mod.rs

@@ -1,21 +0,0 @@
-pub use crate::types::ZfsConfig;
-
-/// Derive the full dataset path for a zone
-pub fn dataset_path(config: &ZfsConfig, zone_name: &str) -> String {
-    format!("{}/{}", config.parent_dataset, zone_name)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_dataset_path() {
-        let config = ZfsConfig {
-            parent_dataset: "rpool/zones".to_string(),
-            clone_from: None,
-            quota: None,
-        };
-        assert_eq!(dataset_path(&config, "myzone"), "rpool/zones/myzone");
-    }
-}

crates/reddwarf-runtime/src/zone/config.rs

@@ -81,8 +81,7 @@ mod tests {
                 gateway: "10.0.0.1".to_string(),
                 prefix_len: 16,
             }),
-            zfs: ZfsConfig {
-                parent_dataset: "rpool/zones".to_string(),
+            storage: ZoneStorageOpts {
                 clone_from: None,
                 quota: Some("10G".to_string()),
             },
@@ -119,8 +118,7 @@ mod tests {
                 gateway: "192.168.1.1".to_string(),
                 prefix_len: 24,
             }),
-            zfs: ZfsConfig {
-                parent_dataset: "rpool/zones".to_string(),
+            storage: ZoneStorageOpts {
                 clone_from: Some("rpool/zones/template@base".to_string()),
                 quota: None,
             },

crates/reddwarf/src/main.rs

@@ -2,8 +2,8 @@ use clap::{Parser, Subcommand};
 use reddwarf_apiserver::{ApiError, ApiServer, AppState, Config as ApiConfig};
 use reddwarf_core::Namespace;
 use reddwarf_runtime::{
-    ApiClient, Ipam, MockRuntime, NodeAgent, NodeAgentConfig, PodController, PodControllerConfig,
-    ZoneBrand,
+    ApiClient, Ipam, MockRuntime, MockStorageEngine, NodeAgent, NodeAgentConfig, PodController,
+    PodControllerConfig, StorageEngine, StoragePoolConfig, ZoneBrand,
 };
 use reddwarf_scheduler::scheduler::SchedulerConfig;
 use reddwarf_scheduler::Scheduler;
@@ -42,12 +42,21 @@ enum Commands {
         /// Path to the redb database file
         #[arg(long, default_value = "./reddwarf.redb")]
         data_dir: String,
-        /// Prefix for zone root paths
-        #[arg(long, default_value = "/zones")]
-        zonepath_prefix: String,
-        /// Parent ZFS dataset for zone storage
-        #[arg(long, default_value = "rpool/zones")]
-        zfs_parent: String,
+        /// Base ZFS storage pool name (auto-derives {pool}/zones, {pool}/images, {pool}/volumes)
+        #[arg(long, default_value = "rpool")]
+        storage_pool: String,
+        /// Override the zones dataset (default: {storage_pool}/zones)
+        #[arg(long)]
+        zones_dataset: Option<String>,
+        /// Override the images dataset (default: {storage_pool}/images)
+        #[arg(long)]
+        images_dataset: Option<String>,
+        /// Override the volumes dataset (default: {storage_pool}/volumes)
+        #[arg(long)]
+        volumes_dataset: Option<String>,
+        /// Prefix for zone root paths (default: derived from storage pool as "/{pool}/zones")
+        #[arg(long)]
+        zonepath_prefix: Option<String>,
         /// Pod network CIDR for IPAM allocation
         #[arg(long, default_value = "10.88.0.0/16")]
         pod_cidr: String,
@@ -75,8 +84,11 @@ async fn main() -> miette::Result<()> {
             node_name,
             bind,
             data_dir,
+            storage_pool,
+            zones_dataset,
+            images_dataset,
+            volumes_dataset,
             zonepath_prefix,
-            zfs_parent,
             pod_cidr,
             etherstub_name,
         } => {
@@ -84,8 +96,11 @@ async fn main() -> miette::Result<()> {
                 &node_name,
                 &bind,
                 &data_dir,
-                &zonepath_prefix,
-                &zfs_parent,
+                &storage_pool,
+                zones_dataset.as_deref(),
+                images_dataset.as_deref(),
+                volumes_dataset.as_deref(),
+                zonepath_prefix.as_deref(),
                 &pod_cidr,
                 &etherstub_name,
             )
@@ -118,12 +133,16 @@ async fn run_serve(bind: &str, data_dir: &str) -> miette::Result<()> {
 }
 
 /// Run the full agent: API server + scheduler + pod controller + node agent
+#[allow(clippy::too_many_arguments)]
 async fn run_agent(
     node_name: &str,
     bind: &str,
     data_dir: &str,
-    zonepath_prefix: &str,
-    zfs_parent: &str,
+    storage_pool: &str,
+    zones_dataset: Option<&str>,
+    images_dataset: Option<&str>,
+    volumes_dataset: Option<&str>,
+    zonepath_prefix: Option<&str>,
     pod_cidr: &str,
     etherstub_name: &str,
 ) -> miette::Result<()> {
@@ -137,6 +156,24 @@ async fn run_agent(
         .parse()
         .map_err(|e| miette::miette!("Invalid bind address '{}': {}", bind, e))?;
 
+    // Build storage pool configuration
+    let pool_config = StoragePoolConfig::from_pool(storage_pool).with_overrides(
+        zones_dataset,
+        images_dataset,
+        volumes_dataset,
+    );
+
+    // Derive zonepath prefix from pool or use explicit override
+    let zonepath_prefix = zonepath_prefix
+        .unwrap_or_else(|| Box::leak(format!("/{}", pool_config.zones_dataset).into_boxed_str()));
+
+    // Create and initialize storage engine
+    let storage_engine: Arc<dyn StorageEngine> = create_storage_engine(pool_config);
+    storage_engine
+        .initialize()
+        .await
+        .map_err(|e| miette::miette!("Failed to initialize storage: {}", e))?;
+
     // Determine the API URL for internal components to connect to
     let api_url = format!("http://127.0.0.1:{}", listen_addr.port());
 
@@ -176,8 +213,8 @@ async fn run_agent(
         }
     });
 
-    // 3. Create runtime (MockRuntime on non-illumos, IllumosRuntime on illumos)
-    let runtime: Arc<dyn reddwarf_runtime::ZoneRuntime> = create_runtime();
+    // 3. Create runtime with injected storage engine
+    let runtime: Arc<dyn reddwarf_runtime::ZoneRuntime> = create_runtime(storage_engine);
 
     // 4. Create IPAM for per-pod IP allocation
     let ipam = Ipam::new(state.storage.clone(), pod_cidr).map_err(|e| {
@@ -190,7 +227,6 @@ async fn run_agent(
         node_name: node_name.to_string(),
         api_url: api_url.clone(),
         zonepath_prefix: zonepath_prefix.to_string(),
-        zfs_parent_dataset: zfs_parent.to_string(),
         default_brand: ZoneBrand::Reddwarf,
         etherstub_name: etherstub_name.to_string(),
         pod_cidr: pod_cidr.to_string(),
@@ -287,16 +323,30 @@ fn create_app_state(data_dir: &str) -> miette::Result<Arc<AppState>> {
     Ok(Arc::new(AppState::new(storage, version_store)))
 }
 
+/// Create the appropriate storage engine for this platform
+fn create_storage_engine(config: StoragePoolConfig) -> Arc<dyn StorageEngine> {
+    #[cfg(target_os = "illumos")]
+    {
+        info!("Using ZfsStorageEngine (native ZFS support)");
+        Arc::new(reddwarf_runtime::ZfsStorageEngine::new(config))
+    }
+    #[cfg(not(target_os = "illumos"))]
+    {
+        info!("Using MockStorageEngine (in-memory storage for development)");
+        Arc::new(MockStorageEngine::new(config))
+    }
+}
+
 /// Create the appropriate zone runtime for this platform
-fn create_runtime() -> Arc<dyn reddwarf_runtime::ZoneRuntime> {
+fn create_runtime(storage: Arc<dyn StorageEngine>) -> Arc<dyn reddwarf_runtime::ZoneRuntime> {
     #[cfg(target_os = "illumos")]
     {
         info!("Using IllumosRuntime (native zone support)");
-        Arc::new(reddwarf_runtime::IllumosRuntime::new())
+        Arc::new(reddwarf_runtime::IllumosRuntime::new(storage))
     }
     #[cfg(not(target_os = "illumos"))]
     {
         info!("Using MockRuntime (illumos zone emulation for development)");
-        Arc::new(MockRuntime::new())
+        Arc::new(MockRuntime::new(storage))
     }
 }