Add disks to zettacache live with zcache add command (openzfs#392)

Currently, disks are added to the zettacache implicitly, by specifying
new devices on the command line with the `-c PATH` argument.

This commit adds a way to add disks to the zettacache without restarting
the agent, by running `zcache add PATH`.

Additionally, a `zcache sync [--merge]` subcommand is added, to sync a
checkpoint (and optionally request an immediate merge of the index).

cmd/zfs_object_agent/util/src/message.rs

@@ -1,8 +1,11 @@
+use std::fmt::Debug;
 use std::mem::size_of;
 use std::ptr;
 use std::slice;
 
 use safer_ffi::prelude::*;
+use serde::Deserialize;
+use serde::Serialize;
 use tokio::io;
 use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
@@ -132,3 +135,11 @@ pub const TYPE_REPORT_HITS: &str = "report hits";
 pub const TYPE_LIST_DEVICES: &str = "list devices";
 pub const TYPE_ZCACHE_IOSTAT: &str = "zcache iostat";
 pub const TYPE_ZCACHE_STATS: &str = "zcache stats";
+pub const TYPE_ADD_DISK: &str = "add disk";
+pub const TYPE_SYNC_CHECKPOINT: &str = "sync checkpoint";
+pub const TYPE_INITIATE_MERGE: &str = "initiate merge";
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct AddDiskRequest {
+    pub path: String,
+}

cmd/zfs_object_agent/zcache/src/add.rs

@@ -0,0 +1,42 @@
+//! `zcache add` subcommand
+
+use anyhow::anyhow;
+use anyhow::Result;
+use async_trait::async_trait;
+use clap::Parser;
+use util::message::AddDiskRequest;
+use util::message::TYPE_ADD_DISK;
+use util::writeln_stdout;
+
+use crate::remote_channel::RemoteChannel;
+use crate::remote_channel::RemoteError;
+use crate::subcommand::ZcacheSubCommand;
+
+#[derive(Parser)]
+#[clap(about = "Add a disk to the ZettaCache.")]
+pub struct Add {
+    path: String,
+}
+
+#[async_trait]
+impl ZcacheSubCommand for Add {
+    async fn invoke(&self) -> Result<()> {
+        let mut remote = RemoteChannel::new(true).await?;
+
+        let request = AddDiskRequest {
+            path: self.path.clone(),
+        };
+
+        match remote
+            .call(TYPE_ADD_DISK, Some(nvpair::to_nvlist(&request).unwrap()))
+            .await
+        {
+            Ok(_) => {
+                writeln_stdout!("Disk {} added", self.path);
+            }
+            Err(RemoteError::ResultError(_)) => return Err(anyhow!("No cache found")),
+            Err(RemoteError::Other(e)) => return Err(e),
+        }
+        Ok(())
+    }
+}

cmd/zfs_object_agent/zcache/src/hits.rs

@@ -4,6 +4,7 @@ use std::time::Duration;
 use std::time::SystemTime;
 use std::time::UNIX_EPOCH;
 
+use anyhow::anyhow;
 use anyhow::Result;
 use async_trait::async_trait;
 use chrono::DateTime;
@@ -184,7 +185,9 @@ impl ZcacheSubCommand for Hits {
                     writeln_stdout!("Hits-by-size data cleared");
                 }
                 Err(RemoteError::ResultError(_)) => {
-                    writeln_stdout!("No cache found, so no hits-by-size data present");
+                    return Err(anyhow!(
+                        "No cache found, so no hits-by-size data is available"
+                    ))
                 }
                 Err(RemoteError::Other(e)) => return Err(e),
             }
@@ -207,7 +210,9 @@ impl ZcacheSubCommand for Hits {
                 hits_by_size.print(quantiles, cumulative, ghost);
             }
             Err(RemoteError::ResultError(_)) => {
-                writeln_stdout!("No cache found, so no hits-by-size data is available");
+                return Err(anyhow!(
+                    "No cache found, so no hits-by-size data is available"
+                ));
             }
             Err(RemoteError::Other(e)) => return Err(e),
         }

cmd/zfs_object_agent/zcache/src/main.rs

@@ -4,22 +4,21 @@
 #![warn(clippy::cast_sign_loss)]
 #![deny(clippy::print_stdout)]
 #![deny(clippy::print_stderr)]
+
+mod add;
 mod hits;
 mod iostat;
 mod list;
 mod remote_channel;
 mod stats;
 mod subcommand;
+mod sync;
 
 use anyhow::Result;
 use clap::Parser;
 use clap::Subcommand;
 use hits::ClearHitData;
-use hits::Hits;
-use iostat::Iostat;
-use list::List;
 use log::*;
-use stats::Stats;
 use subcommand::ZcacheSubCommand;
 
 fn main() -> Result<()> {
@@ -52,10 +51,12 @@ struct Cli {
 /// 2. Add an entry to the enum here where it will be parsed and instantiated automatically.
 /// 3. Add a match entry to the match block in `async_main()`.
 enum Commands {
-    Hits(Hits),
-    Iostat(Iostat),
-    List(List),
-    Stats(Stats),
+    Hits(hits::Hits),
+    Iostat(iostat::Iostat),
+    List(list::List),
+    Stats(stats::Stats),
+    Add(add::Add),
+    Sync(sync::Sync),
 
     // clear_hit_data is deprecated/hidden
     #[clap(rename_all = "snake_case")]
@@ -75,6 +76,8 @@ async fn async_main() -> Result<()> {
         Commands::Iostat(subcommand) => subcommand.invoke().await?,
         Commands::List(subcommand) => subcommand.invoke().await?,
         Commands::Stats(subcommand) => subcommand.invoke().await?,
+        Commands::Add(subcommand) => subcommand.invoke().await?,
+        Commands::Sync(subcommand) => subcommand.invoke().await?,
     }
 
     Ok(())

cmd/zfs_object_agent/zcache/src/sync.rs

@@ -0,0 +1,40 @@
+//! `zcache sync` subcommand
+
+use anyhow::anyhow;
+use anyhow::Result;
+use async_trait::async_trait;
+use clap::Parser;
+use util::message::TYPE_INITIATE_MERGE;
+use util::message::TYPE_SYNC_CHECKPOINT;
+
+use crate::remote_channel::RemoteChannel;
+use crate::remote_channel::RemoteError;
+use crate::subcommand::ZcacheSubCommand;
+
+#[derive(Parser)]
+#[clap(about = "Wait for changes to be persisted to Zettacache.")]
+pub struct Sync {
+    /// Request index merge.  If a merge is already in progress, a new merge will be started as
+    /// soon as this one completes.
+    #[clap(long)]
+    merge: bool,
+}
+
+#[async_trait]
+impl ZcacheSubCommand for Sync {
+    async fn invoke(&self) -> Result<()> {
+        let mut remote = RemoteChannel::new(true).await?;
+
+        let result = if self.merge {
+            remote.call(TYPE_INITIATE_MERGE, None).await
+        } else {
+            remote.call(TYPE_SYNC_CHECKPOINT, None).await
+        };
+        match result {
+            Ok(_) => {}
+            Err(RemoteError::ResultError(_)) => return Err(anyhow!("No cache found")),
+            Err(RemoteError::Other(e)) => return Err(e),
+        }
+        Ok(())
+    }
+}

cmd/zfs_object_agent/zettacache/src/base_types.rs

@@ -62,7 +62,7 @@ impl DiskId {
         assert_le!(value, Self::MAX_VALUE);
         DiskId(u16::try_from(value).unwrap())
     }
-    pub fn get(self) -> usize {
+    pub fn index(self) -> usize {
         self.0 as usize
     }
 }