Merge #1949

1949: SPDK DiskError and Hot-removal patches r=tiagolobocastro a=tiagolobocastro

     test: add lvs hot-removal
    
    The bdev aio and uring modules now hot-remove the bdev when they
    detect the backing device is hot-removed.
    This happens when they complete an io with -ENODEV, -EIO or positive error.
    
    Add a test which leverages ublk to achieve this by removing the ublk device.
    
---

    feat(ublk): add ublksrv nix package

---

    test: add disk errors test
    
    Makes uses of an LVM lvol allowing us to set it up to yield errors
    without detaching the device.
    
    We "import" the gRPC pool types, allowing us to test we're returning
    the correct states, alerts and errors.

---

    fix(lvm): cleanup stable dm entries when creating a vg
    
---

    test: don't panic twice on environment drop
    
    Also improve some testing asserts to output more information.
    
---

    build: pulls latest spdk-rs to include hotremoval fixes
    
---

    feat(lvm): add device-mapper utilities
    
    Adds methods for suspending, resuming, retrieving the table and
    updating the table.
    This can be useful for testing for example.
    
---

    feat(lvm): support commands with inputs
    
---

    fix(lvm): close fd above stderr
    
    On invocation, lvm requires that only the standard file descriptors stdin,
    stdout and stderr are available. If others are found, they get closed
    and messages are issued warning about the leak.
    
    Previously we were silencing the errors only, now we close the range from
    3 to 1024.
    todo: pick higher range?

---

    refactor: implement default for cli args
    
    Makes it less awkward to use.
    Also we should consider modifying the MayastorEnvironment.
    
---

    refactor: expose lvm vg create lvol
    
    This will allow us to use concrete types for testing.
    
    As a bonus we also avoid looking up the VG twice.
    Though something seems slightly off, there's some indirection
    back and forth from vg to lv.
    
---

    fix(lvm): handle lvol already exists error
    
    If already exists, then validates the parameters.
    
---

    feat: allow running the lvm code outside of spdk context
    
    Adds a no_spdk variable to the PoolArgs, which is only usable
    for the LVM backend currently, though could extend to ZFS ex.
    
    As a simplistic impl for now we won't attach the mayastor tag
    in this case. Probably not ideal, but for now should be ok.
    
    Also the lvm commands were being trampolined to the spdk reactor.
    This is easily removed by simply checking if we're in and spdk
    thread, and if not then no need to trampoline anyway so this
    was an easy win.


Co-authored-by: Tiago Castro <tiagolobocastro@gmail.com>

Author: mayastor-bors

io-engine-tests/src/compose/mod.rs

@@ -139,6 +139,9 @@ impl Drop for MayastorTest<'_> {
         self.reactor.send_future(async { mayastor_env_stop(0) });
         // wait for mayastor to stop
         let hdl = self.thdl.take().unwrap();
-        hdl.join().unwrap()
+        let result = hdl.join();
+        if !std::thread::panicking() {
+            result.unwrap();
+        }
     }
 }

io-engine-tests/src/lib.rs

@@ -176,7 +176,7 @@ pub fn dd_random_file(path: &str, bs: u32, size: u64) {
         .output()
         .expect("failed exec dd");
 
-    assert!(output.status.success());
+    assert!(output.status.success(), "{:#?}", output);
 }
 
 pub fn truncate_file(path: &str, size: u64) {
@@ -185,23 +185,23 @@ pub fn truncate_file(path: &str, size: u64) {
         .output()
         .expect("failed exec truncate");
 
-    assert!(output.status.success());
+    assert!(output.status.success(), "{:#?}", output);
 }
 
 pub fn truncate_file_bytes(path: &str, size: u64) {
     let output = Command::new("truncate")
         .args(["-s", &format!("{size}"), path])
         .output()
         .expect("failed exec truncate");
-    assert!(output.status.success());
+    assert!(output.status.success(), "{:#?}", output);
 }
 
 pub fn expand_tempfs_file_bytes(path: &str, size: u64) {
     let output = Command::new("truncate")
         .args(["-s", &format!("+{size}"), path])
         .output()
         .expect("failed exec truncate");
-    assert!(output.status.success());
+    assert!(output.status.success(), "{:#?}", output);
 }
 
 /// Automatically assign a loopdev to path
@@ -212,7 +212,7 @@ pub fn setup_loopdev_file(path: &str, sector_size: Option<u64>) -> String {
         .args(["-f", "--show", "-b", &format!("{log_sec}"), path])
         .output()
         .expect("failed exec losetup");
-    assert!(output.status.success());
+    assert!(output.status.success(), "{:#?}", output);
     // return the assigned loop device
     String::from_utf8(output.stdout).unwrap().trim().to_string()
 }
@@ -223,7 +223,7 @@ pub fn detach_loopdev(dev: &str) {
         .args(["-d", dev])
         .output()
         .expect("failed exec losetup");
-    assert!(output.status.success());
+    assert!(output.status.success(), "{:#?}", output);
 }
 
 pub fn fscheck(device: &str) {
@@ -235,7 +235,7 @@ pub fn fscheck(device: &str) {
     io::stdout().write_all(&output.stderr).unwrap();
     io::stdout().write_all(&output.stdout).unwrap();
 
-    assert!(output.status.success());
+    assert!(output.status.success(), "{:#?}", output);
 }
 
 pub fn mkfs(path: &str, fstype: &str) -> bool {
@@ -279,7 +279,7 @@ pub fn compare_files(a: &str, b: &str) {
 
     io::stdout().write_all(&output.stderr).unwrap();
     io::stdout().write_all(&output.stdout).unwrap();
-    assert!(output.status.success());
+    assert!(output.status.success(), "{:#?}", output);
 }
 
 pub fn mount_umount(device: &str) -> Result<String, String> {

io-engine-tests/src/pool.rs

@@ -282,6 +282,7 @@ impl PoolBuilderLocal {
             backend: Default::default(),
             enc_key: None,
             crypto_vbdev_name: None,
+            no_spdk: false,
         })
         .await?;
         Ok(lvs)

io-engine/examples/lvs-eval/main.rs

@@ -135,9 +135,7 @@ async fn create_lvs(args: &CliArgs) -> Lvs {
         md_args: Some(PoolMetadataArgs {
             max_expansion: args.max_expansion.clone(),
         }),
-        backend: Default::default(),
-        enc_key: None,
-        crypto_vbdev_name: None,
+        ..Default::default()
     };
 
     Lvs::create_or_import(lvs_args.clone()).await.unwrap()

io-engine/src/bdev/lvs.rs

@@ -217,6 +217,7 @@ impl Lvs {
             enc_key: self.key.clone(),
             // XXX: Is this path ever exercised apart from test, or casperf perhaps?
             crypto_vbdev_name: self.key.as_ref().map(|_| format!("crypto_{}", self.name)),
+            no_spdk: false,
         };
         match &self.mode {
             LvsMode::Create => match crate::lvs::Lvs::import_from_args(args.clone()).await {

io-engine/src/core/env.rs

@@ -299,40 +299,50 @@ pub struct MayastorCliArgs {
 #[derive(Debug, clap::Parser, Clone)]
 pub struct SpdkTracingArgs {
     /// Number of trace entries per lcore.
-    #[clap(long, default_value_t = 0)]
+    #[clap(long, env = "SPDK_TRACING_ENTRIES", default_value_t = 0)]
     pub entries: u64,
     /// Number of user created threads.
-    #[clap(long, default_value_t = 1)]
+    #[clap(long, env = "SPDK_TRACING_THREADS", default_value_t = 1)]
     pub threads: u32,
 }
+impl Default for SpdkTracingArgs {
+    fn default() -> Self {
+        Self::parse_from(Vec::<String>::new())
+    }
+}
 
 /// DiskPool related arguments.
 #[derive(Debug, clap::Parser, Clone)]
 pub struct PoolCliArgs {
     /// I/O error count threshold.
     /// After this many errors a pool alert is raised as Warning.
-    #[clap(long, default_value_t = 64)]
+    #[clap(long, env, default_value_t = 8)]
     pub io_error_threshold: u64,
 
     /// I/O stall deadline.
     /// If an I/O is stuck longer than this period, then the pool is considered stalled and a
     /// Critical alert is raised.
     /// The pool disk will also be reset and the stall will be cleared once complete and
     /// I/O flows again.
-    #[clap(long, default_value = "30s")]
+    #[clap(long, env, default_value = "30s")]
     pub io_stall_deadline: humantime::Duration,
 
     /// I/O stall transitions threshold.
     /// After this many transitions within the window, a pool alert is raised as Warning.
-    #[clap(long, default_value_t = 3)]
+    #[clap(long, env, default_value_t = 3)]
     pub io_stall_transition_threshold: u64,
 
     /// I/O stall transitions window.
     /// Time window during which stall ↔ resume state transitions are tracked
     /// for flakiness detection.
-    #[clap(long, default_value = "3h")]
+    #[clap(long, env, default_value = "3h")]
     pub io_stall_transition_window: humantime::Duration,
 }
+impl Default for PoolCliArgs {
+    fn default() -> Self {
+        Self::parse_from(Vec::<String>::new())
+    }
+}
 
 fn delay_compat(s: &str) -> Result<bool, String> {
     match s {
@@ -507,8 +517,8 @@ impl Default for MayastorEnvironment {
             developer_delay: false,
             rdma: false,
             bs_cluster_unmap: false,
-            pool_args: PoolCliArgs::parse_from(Vec::<String>::new()),
-            traces: SpdkTracingArgs::parse_from(Vec::<String>::new()),
+            pool_args: PoolCliArgs::default(),
+            traces: SpdkTracingArgs::default(),
         }
     }
 }

io-engine/src/grpc/v1/pool.rs

@@ -278,6 +278,7 @@ impl TryFrom<CreatePoolRequest> for PoolArgs {
             backend: backend.into(),
             enc_key: None,
             crypto_vbdev_name: None,
+            no_spdk: false,
         })
     }
 }
@@ -366,6 +367,7 @@ impl TryFrom<ImportPoolRequest> for PoolArgs {
                 .encryption
                 .as_ref()
                 .map(|_| format!("crypto_{}", args.name)),
+            no_spdk: false,
         })
     }
 }
@@ -578,6 +580,11 @@ impl PoolErrorsNt {
     }
 }
 
+/// Convert something which implements [`PoolOps`] to the proto `Pool` type.
+pub async fn pool_to_proto(pool: &dyn PoolOps) -> Pool {
+    pool.async_into().await
+}
+
 impl AsyncFrom<Box<dyn PoolOps>> for Pool {
     async fn async_from(value: Box<dyn PoolOps>) -> Self {
         let value = value.deref();

io-engine/src/lvm/cli.rs

@@ -1,10 +1,11 @@
 use crate::lvm::{error, error::Error, property::Property};
 
+use nix::errno::Errno;
 use serde::de::Deserialize;
 use snafu::ResultExt;
 use std::ffi::OsStr;
 use strum_macros::{AsRefStr, Display, EnumString};
-use tokio::process::Command;
+use tokio::{io::AsyncWriteExt, process::Command};
 
 /// Common set of query options for a volume group or logical volume.
 /// If the name is present then the name will be used to query.
@@ -30,6 +31,15 @@ impl CmnQueryArgs {
             ..Default::default()
         }
     }
+    /// Find only our entries (ie, with our tag).
+    pub(crate) fn ours_if(spdk: bool) -> Self {
+        if spdk {
+            Self::ours()
+        } else {
+            Self::any()
+        }
+    }
+
     /// Find entries with the given name.
     pub(crate) fn named_opt(self, name: &Option<String>) -> Self {
         let Some(name) = name else {
@@ -111,12 +121,19 @@ enum LvmSubCmd {
     /// Display information about logical volumes.
     #[strum(serialize = "lvs")]
     LVList,
+    /// DeviceMapper commands.
+    #[strum(serialize = "dmsetup")]
+    DMSetup,
+    /// BlockDevice commands.
+    #[strum(serialize = "blockdev")]
+    BlkDev,
 }
 
 /// LVM wrapper over `Command` with added qol such as error mapping and
 /// decoding of json output reports.
 pub(super) struct LvmCmd {
     cmd: &'static str,
+    input: Option<String>,
     cmder: Command,
 }
 
@@ -142,6 +159,7 @@ impl LvmCmd {
         Self {
             cmd,
             cmder: Command::new(cmd),
+            input: None,
         }
     }
     /// Prepare a `Command` for `LvmSubCmd::PVCreate`.
@@ -188,6 +206,14 @@ impl LvmCmd {
     pub(super) fn lv_list() -> Self {
         Self::new(LvmSubCmd::LVList.as_ref())
     }
+    /// Prepare a `Command` for `LvmSubCmd::DMSuspend`.
+    pub(super) fn dm_setup() -> Self {
+        Self::new(LvmSubCmd::DMSetup.as_ref())
+    }
+    /// Prepare a `Command` for `LvmSubCmd::BlkDev`.
+    pub(super) fn blk_dev() -> Self {
+        Self::new(LvmSubCmd::BlkDev.as_ref())
+    }
     /// Runs the LVM command with the provided `Command` arguments et all and
     /// returns an LVM specific report containing an output type `T`.
     /// >> Note: This requires the json output to be specified in args.
@@ -241,7 +267,7 @@ impl LvmCmd {
     /// Tag the given `Property`.
     pub(super) fn tag_if(self, tag: bool, property: Property) -> Self {
         if tag {
-            self.arg(property.add())
+            self.tag(property)
         } else {
             self
         }
@@ -264,6 +290,11 @@ impl LvmCmd {
         self.cmder.args(args);
         self
     }
+    /// Run the command with the given input.
+    pub fn input(mut self, input: String) -> Self {
+        self.input = Some(input);
+        self
+    }
     /// Runs the LVM command with the provided `Command` arguments et al.
     ///
     /// # Errors
@@ -285,14 +316,11 @@ impl LvmCmd {
     pub(super) async fn output(mut self) -> Result<std::process::Output, Error> {
         tracing::trace!("{:?}", self.cmder);
 
-        crate::tokio_run!(async move {
-            let output = self
-                .cmder
-                .output()
-                .await
-                .context(error::LvmBinSpawnErrSnafu {
-                    command: self.cmd.to_string(),
-                })?;
+        let in_spdk = spdk_rs::Thread::is_spdk_thread();
+        let fut = async move {
+            let output = self.cmder().await.context(error::LvmBinSpawnErrSnafu {
+                command: self.cmd.to_string(),
+            })?;
             if !output.status.success() {
                 let error = String::from_utf8_lossy(&output.stderr).to_string();
                 return Err(Error::LvmBinErr {
@@ -301,7 +329,42 @@ impl LvmCmd {
                 });
             }
             Ok(output)
-        })
+        };
+        if in_spdk {
+            crate::tokio_run!(fut)
+        } else {
+            fut.await
+        }
+    }
+
+    async fn cmder(&mut self) -> std::io::Result<std::process::Output> {
+        unsafe {
+            self.cmder.pre_exec(|| {
+                Self::close_range().ok();
+                Ok(())
+            });
+        }
+
+        let Some(input) = self.input.take() else {
+            return self.cmder.output().await;
+        };
+
+        let mut child = self.cmder.stdin(std::process::Stdio::piped()).spawn()?;
+
+        if let Some(mut stdin) = child.stdin.take() {
+            stdin.write_all(input.as_bytes()).await?;
+            stdin.shutdown().await?;
+        }
+
+        child.wait_with_output().await
+    }
+
+    /// The close_range system call closes all open file descriptors from first to last (included).
+    /// Here close from 3 to 1024.
+    /// todo: find a better way such as querying /proc/self/fd ?.
+    fn close_range() -> nix::Result<()> {
+        let res = unsafe { libc::close_range(3, 1024, libc::CLOSE_RANGE_CLOEXEC as i32) };
+        Errno::result(res).map(drop)
     }
 }