feat(lvm/vg): create and destroy vgs

Rather than importing existing VGs we now create the PV's and the VG itself.
On destruction, if the VG has no non-mayastor LVs then we destroy it, otherwise
we leave it behind, untagged.

Signed-off-by: Tiago Castro <tiagolobocastro@gmail.com>

Author: tiagolobocastro

io-engine/src/core/env.rs

@@ -46,6 +46,7 @@ use crate::{
     core::{
         nic,
         reactor::{Reactor, ReactorState, Reactors},
+        runtime,
         Cores,
         MayastorFeatures,
         Mthread,
@@ -486,6 +487,12 @@ async fn do_shutdown(arg: *mut c_void) {
     }
     nexus::shutdown_nexuses().await;
     crate::lvs::Lvs::export_all().await;
+
+    runtime::spawn_await(async {
+        crate::lvm::VolumeGroup::export_all().await;
+    })
+    .await;
+
     unsafe {
         spdk_rpc_finish();
         spdk_subsystem_fini(Some(reactors_stop), arg);

io-engine/src/core/runtime.rs

@@ -3,7 +3,8 @@
 //! runtime to do whatever it needs to do. The tokio threads are
 //! unaffinitized such that they do not run on any of our reactors.
 
-use futures::Future;
+use crate::core::Reactor;
+use futures::{channel::oneshot, Future};
 use once_cell::sync::Lazy;
 use tokio::task::JoinHandle;
 
@@ -14,6 +15,22 @@ pub fn spawn(f: impl Future<Output = ()> + Send + 'static) {
     RUNTIME.spawn(f);
 }
 
+/// Spawn a future on the tokio runtime and await its completion.
+pub async fn spawn_await(f: impl Future<Output = ()> + Send + 'static) {
+    let (s, r) = oneshot::channel();
+
+    RUNTIME.spawn(async move {
+        f.await;
+
+        if let Ok(r) = Reactor::spawn_at_primary(async move {
+            s.send(()).ok();
+        }) {
+            r.await.ok();
+        }
+    });
+    r.await.ok();
+}
+
 /// block on the given future until it completes
 pub fn block_on(f: impl Future<Output = ()> + Send + 'static) {
     RUNTIME.block_on(f);

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

@@ -59,7 +59,7 @@ impl PoolRpc for PoolService {
                 "an lvs pool already uses the disk",
             ));
         };
-        VolumeGroup::import_or_create(args)
+        VolumeGroup::create(args)
             .await
             .map_err(Status::from)
             .map(Pool::from)

io-engine/src/grpc/v1/lvm/replica.rs

@@ -114,7 +114,7 @@ impl ReplicaService {
                     })?);
 
                 if let Err(error) = lvol.share_nvmf(Some(props)).await {
-                    error!("Failed to import lvol: {error}...");
+                    error!("Failed to share lvol: {error}...");
                     if created {
                         // if we have created it here, then let's undo it
                         lvol.destroy().await.ok();

io-engine/src/lvm/cli.rs

@@ -78,12 +78,24 @@ impl CmnQueryArgs {
 /// The following commands implement the core LVM functionality.
 #[derive(AsRefStr, EnumString, Display)]
 enum LvmSubCmd {
+    /// Initialize physical volume(s) for use by LVM.
+    #[strum(serialize = "pvcreate")]
+    PVCreate,
+    /// Remove LVM label(s) from physical volume(s).
+    #[strum(serialize = "pvremove")]
+    PVRemove,
     /// Display information about volume groups.
     #[strum(serialize = "vgs")]
     VGList,
+    /// Create a volume group.
+    #[strum(serialize = "vgcreate")]
+    VGCreate,
     /// Change volume group attributes.
     #[strum(serialize = "vgchange")]
     VGChange,
+    /// Remove volume group(s).
+    #[strum(serialize = "vgremove")]
+    VGRemove,
     /// Create a logical volume.
     #[strum(serialize = "lvcreate")]
     LVCreate,
@@ -132,6 +144,18 @@ impl LvmCmd {
             cmder: Command::new(cmd),
         }
     }
+    /// Prepare a `Command` for `LvmSubCmd::PVCreate`.
+    pub(super) fn pv_create() -> Self {
+        Self::new(LvmSubCmd::PVCreate.as_ref())
+    }
+    /// Prepare a `Command` for `LvmSubCmd::PVRemove`.
+    pub(super) fn pv_remove() -> Self {
+        Self::new(LvmSubCmd::PVRemove.as_ref())
+    }
+    /// Prepare a `Command` for `LvmSubCmd::VGCreate`.
+    pub(super) fn vg_create() -> Self {
+        Self::new(LvmSubCmd::VGCreate.as_ref())
+    }
     /// Prepare a `Command` for `LvmSubCmd::VGList`.
     pub(super) fn vg_list() -> Self {
         Self::new(LvmSubCmd::VGList.as_ref())
@@ -140,6 +164,10 @@ impl LvmCmd {
     pub(super) fn vg_change(vg_name: &str) -> Self {
         Self::new(LvmSubCmd::VGChange.as_ref()).arg(vg_name)
     }
+    /// Prepare a `Command` for `LvmSubCmd::VGRemove`.
+    pub(super) fn vg_remove() -> Self {
+        Self::new(LvmSubCmd::VGRemove.as_ref())
+    }
     /// Prepare a `Command` for `LvmSubCmd::LVCreate`.
     pub(super) fn lv_create() -> Self {
         Self::new(LvmSubCmd::LVCreate.as_ref())
@@ -322,7 +350,10 @@ pub(super) mod de {
             where
                 E: de::Error,
             {
-                let iter = s.split(',').map(FromStr::from_str);
+                let iter = s
+                    .split(',')
+                    .skip_while(|&x| x.is_empty())
+                    .map(FromStr::from_str);
                 Result::from_iter(iter).map_err(de::Error::custom)
             }
         }

io-engine/src/lvm/lv_replica.rs

@@ -108,6 +108,10 @@ pub struct LogicalVolume {
     #[serde(rename = "lv_tags")]
     #[serde(deserialize_with = "de::comma_separated")]
     tags: Vec<Property>,
+    /// Tags of the parent VG converted to our properties.
+    #[serde(rename = "vg_tags")]
+    #[serde(deserialize_with = "de::comma_separated")]
+    vg_tags: Vec<Property>,
 
     #[serde(skip)]
     runtime: RunLogicalVolume,
@@ -267,6 +271,9 @@ impl LogicalVolume {
     /// The LV is then imported as an spdk BDEV, which allows it to be shared
     /// via nvmf or open locally (ex: by the nexus).
     pub(crate) async fn import(&mut self) -> Result<(), Error> {
+        if !self.ours() || !self.vg_ours() {
+            return Ok(());
+        }
         self.import_bdev().await
     }
 
@@ -555,6 +562,14 @@ impl LogicalVolume {
     pub(crate) fn uuid(&self) -> &str {
         &self.lv_name
     }
+    /// Check if this LV is owned by us.
+    pub(crate) fn ours(&self) -> bool {
+        self.tags.contains(&Property::Lvm)
+    }
+    /// Check if the parent VG is owned by us.
+    pub(crate) fn vg_ours(&self) -> bool {
+        self.vg_tags.contains(&Property::Lvm)
+    }
     /// Get the name of the logical volume.
     pub(crate) fn name(&self) -> &Option<String> {
         &self.name

io-engine/src/lvm/vg_pool.rs

@@ -124,17 +124,46 @@ impl VolumeGroup {
             .into_iter()
             // todo: not needed as we did the select?
             .filter(|vg| vg.matches(opts))
-            .collect::<Vec<_>>();
+            .fold(Vec::<VolumeGroup>::new(), |mut acc, vg| {
+                match acc.iter_mut().find(|e_vg| e_vg.name == vg.name) {
+                    None => acc.push(vg),
+                    Some(e_vg) => {
+                        e_vg.disks.extend(vg.disks);
+                    }
+                }
+                acc
+            });
 
         Ok(vgs)
     }
 
     /// Import a volume group with the name provided or create one with the name
     /// and disks provided currently only import is supported.
-    pub(crate) async fn import_or_create(
-        args: PoolArgs,
-    ) -> Result<VolumeGroup, Error> {
-        let vg = Self::import_inner(args).await?;
+    pub(crate) async fn create(args: PoolArgs) -> Result<VolumeGroup, Error> {
+        let vg =
+            match VolumeGroup::lookup(CmnQueryArgs::any().named(&args.name))
+                .await
+            {
+                Ok(_) => Self::import_inner(args).await,
+                Err(Error::NotFound {
+                    ..
+                }) => {
+                    LvmCmd::pv_create().args(&args.disks).run().await?;
+
+                    LvmCmd::vg_create()
+                        .arg(&args.name)
+                        .tag(Property::Lvm)
+                        .args(args.disks)
+                        .run()
+                        .await?;
+                    let lookup = CmnQueryArgs::ours()
+                        .named(&args.name)
+                        .uuid_opt(&args.uuid);
+                    VolumeGroup::lookup(lookup).await
+                }
+                Err(error) => Err(error),
+            }?;
+
         info!("The lvm vg pool '{}' has been created", vg.name());
         Ok(vg)
     }
@@ -149,6 +178,14 @@ impl VolumeGroup {
             .with_vg(CmnQueryArgs::ours().uuid(self.uuid()).named(self.name()));
         LogicalVolume::list(&query).await
     }
+    async fn list_foreign_lvs(&self) -> Result<Vec<LogicalVolume>, Error> {
+        let query = super::QueryArgs::new()
+            .with_lv(CmnQueryArgs::any())
+            .with_vg(CmnQueryArgs::any().uuid(self.uuid()).named(self.name()));
+        LogicalVolume::list(&query)
+            .await
+            .map(|lvs| lvs.into_iter().filter(|lv| !lv.ours()).collect())
+    }
 
     /// Import a volume group by its name, match the disks on the volume group
     /// and if true add our tag to the volume group to make it available
@@ -191,22 +228,33 @@ impl VolumeGroup {
     }
 
     /// Delete the volume group.
-    /// > Note: Currently the vg is only exported.
-    /// todo: currently only exporting the VG.
+    /// > Note: The Vg is first exported and then destroyed.
     pub(crate) async fn destroy(mut self) -> Result<(), Error> {
-        // As currently only import of volume group is supported
-        // exporting the volume group on destroy.
-        let name = self.name().to_string();
         self.export().await?;
-        info!("LVM pool '{name}' has been destroyed successfully");
+
+        let foreign_lvs = self.list_foreign_lvs().await?;
+        let name = self.name().to_string();
+
+        if foreign_lvs.is_empty() {
+            LvmCmd::vg_remove()
+                .arg(format!("--select=vg_name={name}"))
+                .arg("-y")
+                .run()
+                .await?;
+
+            LvmCmd::pv_remove().args(&self.disks).run().await?;
+
+            info!("LVM pool '{}' has been destroyed successfully", self.name());
+        } else {
+            warn!("LVM pool '{}' is not destroyed as it contains foreign lvs: {foreign_lvs:?}", self.name());
+        }
         self.ptpl().destroy().ok();
         Ok(())
     }
 
     /// Exports the volume group by unloading all logical volumes and finally
     /// removing our tag from it
     pub(crate) async fn export(&mut self) -> Result<(), Error> {
-        // todo: unload all logical volumes from this VG
         let lvs = self.list_lvs().await?;
         for mut lv in lvs {
             lv.export_bdev().await?;
@@ -215,7 +263,21 @@ impl VolumeGroup {
         LvmCmd::vg_change(self.name())
             .untag(Property::Lvm)
             .run()
-            .await
+            .await?;
+
+        info!("LVM pool '{}' has been exported successfully", self.name);
+        Ok(())
+    }
+
+    /// Export all VG instances.
+    pub(crate) async fn export_all() {
+        let Ok(pools) = VolumeGroup::list(&CmnQueryArgs::ours()).await else {
+            return;
+        };
+
+        for mut pool in pools {
+            pool.export().await.ok();
+        }
     }
 
     /// Create a logical volume in this volume group.