DLPX-80711 zcache device removal (openzfs#419)

cmd/zfs_object_agent/util/src/message.rs

@@ -139,6 +139,10 @@ pub const TYPE_ZCACHE_STATS: &str = "zcache stats";
 pub const TYPE_ZCACHE_STATUS: &str = "zcache status";
 pub const TYPE_ADD_DISK: &str = "add disk";
 pub const TYPE_EXPAND_DISK: &str = "expand disk";
+pub const TYPE_REMOVE_DISK: &str = "remove disk";
+pub const TYPE_CANCEL_DISK_REMOVAL: &str = "cancel disk removal";
+pub const TYPE_PAUSE_DISK_REMOVALS: &str = "pause disk removals";
+pub const TYPE_RESUME_DISK_REMOVALS: &str = "resume disk removals";
 pub const TYPE_SYNC_CHECKPOINT: &str = "sync checkpoint";
 pub const TYPE_INITIATE_MERGE: &str = "initiate merge";
 
@@ -152,6 +156,11 @@ pub struct ExpandDiskRequest {
     pub path: PathBuf,
 }
 
+#[derive(Serialize, Deserialize, Debug)]
+pub struct RemoveDiskRequest {
+    pub path: PathBuf,
+}
+
 #[derive(Serialize, Deserialize, Debug)]
 pub struct ExpandDiskResponse {
     pub new_size: u64,

cmd/zfs_object_agent/util/src/vec_ext.rs

@@ -243,7 +243,7 @@ pub struct VecMap<K, V> {
 
 impl<K, V> VecMap<K, V>
 where
-    K: Into<usize> + Copy,
+    K: From<usize> + Into<usize> + Copy,
 {
     /// Returns old value (or None if not present)
     pub fn insert(&mut self, key: K, value: V) -> Option<V> {
@@ -285,6 +285,10 @@ where
         self.vec[index].take()
     }
 
+    pub fn contains_key(&self, key: &K) -> bool {
+        self.get(*key).is_some()
+    }
+
     pub fn values(&self) -> impl Iterator<Item = &V> {
         self.vec.iter().filter_map(|v| v.as_ref())
     }
@@ -313,6 +317,38 @@ where
         self.vec[range].iter_mut().filter_map(|v| v.as_mut())
     }
 
+    pub fn iter(&self) -> impl DoubleEndedIterator<Item = (K, &V)> {
+        self.vec
+            .iter()
+            .enumerate()
+            .flat_map(|(k, v)| v.as_ref().map(|v| (k.into(), v)))
+    }
+
+    pub fn iter_mut(&mut self) -> impl DoubleEndedIterator<Item = (K, &mut V)> {
+        self.vec
+            .iter_mut()
+            .enumerate()
+            .flat_map(|(k, v)| v.as_mut().map(|v| (k.into(), v)))
+    }
+
+    pub fn keys(&self) -> impl DoubleEndedIterator<Item = K> + '_ {
+        self.iter().map(|(k, _)| k)
+    }
+
+    pub fn retain<F>(&mut self, mut f: F)
+    where
+        F: FnMut(&K, &mut V) -> bool,
+    {
+        for k in 0..self.vec.len() {
+            let k = k.into();
+            if let Some(v) = self.get_mut(k) {
+                if !f(&k, v) {
+                    self.remove(k);
+                }
+            }
+        }
+    }
+
     /// Returns the number of elements in the map.
     pub fn len(&self) -> usize {
         self.num_entries
@@ -322,3 +358,31 @@ where
         self.len() == 0
     }
 }
+
+impl<K, V> VecMap<K, V>
+where
+    K: From<usize> + Into<usize> + Copy,
+    V: Default,
+{
+    pub fn get_mut_or_default(&mut self, key: K) -> &mut V {
+        if self.contains_key(&key) {
+            return self.get_mut(key).unwrap();
+        }
+        let old = self.insert(key, Default::default());
+        assert!(old.is_none());
+        self.get_mut(key).unwrap()
+    }
+}
+
+impl<K, V> FromIterator<(K, V)> for VecMap<K, V>
+where
+    K: From<usize> + Into<usize> + Copy,
+{
+    fn from_iter<T: IntoIterator<Item = (K, V)>>(iter: T) -> VecMap<K, V> {
+        let mut map = VecMap::default();
+        for (k, v) in iter.into_iter() {
+            map.insert(k, v);
+        }
+        map
+    }
+}

cmd/zfs_object_agent/util/src/zettacache_stats.rs

@@ -386,8 +386,8 @@ pub enum CacheStatCounter {
     SpeculativeBufferBytesAvailable,
     SlabCapacity,
     AvailableSpace,
-    AvailableBlocksSize,
-    AvailableSlabsSize,
+    FreeBlocksSize,
+    FreeSlabsSize,
 }
 
 impl Display for CacheStatCounter {
@@ -447,8 +447,8 @@ impl Sub<&Self> for &CacheStats {
                 | CacheStatCounter::SpeculativeBufferBytesAvailable
                 | CacheStatCounter::SlabCapacity
                 | CacheStatCounter::AvailableSpace
-                | CacheStatCounter::AvailableSlabsSize
-                | CacheStatCounter::AvailableBlocksSize => {
+                | CacheStatCounter::FreeSlabsSize
+                | CacheStatCounter::FreeBlocksSize => {
                     *diff_stat = self_stat.clone();
                 }
                 // Everything else should be subtracted

cmd/zfs_object_agent/zcache/Cargo.toml

@@ -13,6 +13,7 @@ chrono = "0.4"
 async-trait = "0.1.51"
 clap={version="3.1.6",features=["derive"]}
 exitcode = "1.1.2"
+git-version = "0.3.5"
 humantime = "2.1.0"
 log = "0.4"
 more-asserts = "0.2.1"

cmd/zfs_object_agent/zcache/src/main.rs

@@ -12,6 +12,7 @@ mod iostat;
 mod labelclear;
 mod list;
 mod remote_channel;
+mod remove;
 mod stats;
 mod status;
 mod subcommand;
@@ -22,17 +23,24 @@ use std::path::PathBuf;
 use anyhow::Result;
 use clap::Parser;
 use clap::Subcommand;
+use git_version::git_version;
 use hits::ClearHitData;
 use log::*;
 use subcommand::ZcacheSubCommand;
 
+static GIT_VERSION: &str = git_version!(
+    fallback = match option_env!("CARGO_ZOA_GITREV") {
+        Some(value) => value,
+        None => "unknown",
+    }
+);
+
 fn main() -> Result<()> {
     async_main()
 }
 
 #[derive(Parser)]
-// XXX other commands use a git derived version here
-#[clap(version = "1.1")]
+#[clap(version=GIT_VERSION)]
 #[clap(name = "zcache")]
 #[clap(about = "ZFS ZettaCache Command")]
 #[clap(propagate_version = true)]
@@ -62,6 +70,7 @@ enum Commands {
     Stats(stats::Stats),
     Add(add::Add),
     Expand(expand::Expand),
+    Remove(remove::Remove),
     Sync(sync::Sync),
     Labelclear(labelclear::Labelclear),
     Status(status::Status),
@@ -80,6 +89,7 @@ impl Commands {
             Commands::Stats(stats) => stats,
             Commands::Add(add) => add,
             Commands::Expand(expand) => expand,
+            Commands::Remove(remove) => remove,
             Commands::Sync(sync) => sync,
             Commands::Labelclear(labelclear) => labelclear,
             Commands::ClearHitData(clear_hit_data) => clear_hit_data,
@@ -104,8 +114,26 @@ mod test_clap {
 
     use super::*;
 
+    fn neg(s: &str) {
+        assert!(Cli::try_parse_from(s.split_whitespace()).is_err());
+    }
+
     #[test]
     fn test_debug_asserts() {
         Cli::command().debug_assert();
     }
+
+    #[test]
+    fn test_zcache_remove_conflicts() {
+        neg("zcache remove --pause --resume");
+        neg("zcache remove --pause --cancel");
+        neg("zcache remove --pause -s");
+        neg("zcache remove --pause disk0");
+        neg("zcache remove --resume --cancel");
+        neg("zcache remove --resume -s");
+        neg("zcache remove --resume -s");
+        neg("zcache remove --resume disk0");
+        neg("zcache remove --cancel");
+        neg("zcache remove");
+    }
 }

cmd/zfs_object_agent/zcache/src/remove.rs

@@ -0,0 +1,74 @@
+//! `zcache remove` subcommand
+
+use std::path::PathBuf;
+
+use anyhow::Result;
+use async_trait::async_trait;
+use clap::Parser;
+use util::message::RemoveDiskRequest;
+use util::message::TYPE_CANCEL_DISK_REMOVAL;
+use util::message::TYPE_PAUSE_DISK_REMOVALS;
+use util::message::TYPE_REMOVE_DISK;
+use util::message::TYPE_RESUME_DISK_REMOVALS;
+use util::writeln_stdout;
+
+use crate::remote_channel::RemoteChannel;
+use crate::subcommand::ZcacheSubCommand;
+
+#[derive(Parser)]
+#[clap(about = "Remove disks from the ZettaCache.")]
+pub struct Remove {
+    /// Pause ongoing removals
+    #[clap(
+        long,
+        conflicts_with_all = &["resume", "cancel", "path"]
+    )]
+    pause: bool,
+
+    /// Resume paused removals
+    #[clap(
+        long,
+        conflicts_with_all = &["cancel", "path"]
+    )]
+    resume: bool,
+
+    /// Stop and cancel in-progress removal
+    #[clap(short = 's', long)]
+    cancel: bool,
+
+    /// Supplied disk for the operation issued
+    #[clap(required_unless_present_any = &["pause", "resume"])]
+    path: Option<PathBuf>,
+}
+
+#[async_trait]
+impl ZcacheSubCommand for Remove {
+    async fn invoke(&self) -> Result<()> {
+        let mut remote = RemoteChannel::new(true).await?;
+        if self.pause {
+            remote.call(TYPE_PAUSE_DISK_REMOVALS, None).await?;
+            writeln_stdout!("Removals paused");
+        } else if self.resume {
+            remote.call(TYPE_RESUME_DISK_REMOVALS, None).await?;
+            writeln_stdout!("Resuming removals");
+        } else if self.cancel {
+            let path = self.path.clone().unwrap();
+            let request = RemoveDiskRequest { path: path.clone() };
+            remote
+                .call(
+                    TYPE_CANCEL_DISK_REMOVAL,
+                    Some(nvpair::to_nvlist(&request).unwrap()),
+                )
+                .await?;
+            writeln_stdout!("Cancelled the removal of {path:?}");
+        } else {
+            let path = self.path.clone().unwrap();
+            let request = RemoveDiskRequest { path: path.clone() };
+            remote
+                .call(TYPE_REMOVE_DISK, Some(nvpair::to_nvlist(&request).unwrap()))
+                .await?;
+            writeln_stdout!("Removing {path:?}");
+        }
+        Ok(())
+    }
+}

cmd/zfs_object_agent/zcache/src/stats.rs

@@ -256,11 +256,11 @@ impl StatsDisplay {
         if self.show_block_allocator {
             let available_space = values.value(AvailableSpace);
             let slab_capacity = values.value(SlabCapacity);
-            let free_blocks_size = values.value(AvailableBlocksSize);
-            let free_slabs_size = values.value(AvailableSlabsSize);
-            let block_allocator_allocated = slab_capacity - free_blocks_size - free_slabs_size;
+            let free_blocks_size = values.value(FreeBlocksSize);
+            let free_slabs_size = values.value(FreeSlabsSize);
+            let allocated_space = slab_capacity - free_blocks_size - free_slabs_size;
 
-            self.display_bytes(block_allocator_allocated as f64);
+            self.display_bytes(allocated_space as f64);
             self.display_bytes(available_space as f64);
             self.display_percent(free_blocks_size as f64, slab_capacity as f64);
             self.display_percent(free_slabs_size as f64, slab_capacity as f64);

cmd/zfs_object_agent/zettacache/src/base_types.rs

@@ -91,6 +91,19 @@ impl DiskId {
     pub fn index(self) -> usize {
         self.0 as usize
     }
+    pub fn next(&self) -> DiskId {
+        DiskId(self.0 + 1)
+    }
+}
+impl From<DiskId> for usize {
+    fn from(val: DiskId) -> Self {
+        val.index()
+    }
+}
+impl From<usize> for DiskId {
+    fn from(val: usize) -> Self {
+        DiskId::new(val)
+    }
 }
 
 #[derive(Serialize, Deserialize, Copy, Clone, PartialEq, Eq, Ord, PartialOrd)]