Try #888:

control-plane/csi-driver/src/bin/node/block_vol.rs

@@ -88,7 +88,7 @@ pub(crate) async fn publish_block_volume(msg: &NodePublishVolumeRequest) -> Resu
         }
 
         if let Err(error) =
-            mount::blockdevice_mount(&device_path, target_path.as_str(), msg.readonly)
+            mount::blockdevice_mount(&device_path, target_path.as_str(), msg.readonly).await
         {
             return Err(failure!(
                 Code::Internal,
@@ -108,14 +108,14 @@ pub(crate) async fn publish_block_volume(msg: &NodePublishVolumeRequest) -> Resu
     }
 }
 
-pub(crate) fn unpublish_block_volume(msg: &NodeUnpublishVolumeRequest) -> Result<(), Status> {
+pub(crate) async fn unpublish_block_volume(msg: &NodeUnpublishVolumeRequest) -> Result<(), Status> {
     let target_path = &msg.target_path;
     let volume_id = &msg.volume_id;
 
     // block volumes are mounted on block special file, which is not
     // a regular file.
     if mount::find_mount(None, Some(target_path)).is_some() {
-        match mount::blockdevice_unmount(target_path) {
+        match mount::blockdevice_unmount(target_path).await {
             Ok(_) => {}
             Err(err) => {
                 return Err(Status::new(

control-plane/csi-driver/src/bin/node/error.rs

@@ -3,6 +3,7 @@ use nix::errno::Errno;
 use nvmeadm::{error::NvmeError, nvmf_discovery};
 use snafu::Snafu;
 use std::{process::ExitCode, string::FromUtf8Error};
+use tokio::task::JoinError;
 
 /// A Device Attach/Detach error.
 /// todo: should this be an enum?
@@ -171,8 +172,6 @@ pub(crate) enum FsfreezeError {
     },
     #[snafu(display("Not a filesystem mount: volume ID: {volume_id}"))]
     BlockDeviceMount { volume_id: String },
-    #[snafu(display("Not a valid fsfreeze command"))]
-    InvalidFreezeCommand,
 }
 
 impl From<FsfreezeError> for ExitCode {
@@ -184,7 +183,62 @@ impl From<FsfreezeError> for ExitCode {
             FsfreezeError::FsfreezeFailed { errno, .. } => ExitCode::from(errno as u8),
             FsfreezeError::InternalFailure { .. } => ExitCode::from(Errno::ELIBACC as u8),
             FsfreezeError::BlockDeviceMount { .. } => ExitCode::from(Errno::EMEDIUMTYPE as u8),
-            FsfreezeError::InvalidFreezeCommand => ExitCode::from(Errno::EINVAL as u8),
+        }
+    }
+}
+
+/// Error for the mount operations.
+#[derive(Debug, Snafu)]
+#[snafu(visibility(pub(crate)), context(suffix(false)))]
+pub(crate) enum MountError {
+    #[snafu(display("Failed mount, error: {error}"))]
+    MountFailed { error: std::io::Error },
+    #[snafu(display("Failed unmount, error: {error}"))]
+    UnmountFailed { error: std::io::Error },
+    #[snafu(display("Failed to wait for thread termination, error: {join_error}"))]
+    FailedWaitForThread { join_error: JoinError },
+}
+
+impl From<MountError> for ExitCode {
+    fn from(value: MountError) -> Self {
+        match value {
+            MountError::MountFailed { .. } => ExitCode::from(Errno::ELIBACC as u8),
+            MountError::UnmountFailed { .. } => ExitCode::from(Errno::ELIBACC as u8),
+            MountError::FailedWaitForThread { .. } => ExitCode::from(Errno::ESRCH as u8),
+        }
+    }
+}
+
+/// Error for the csi driver operations.
+#[derive(Debug, Snafu)]
+#[snafu(visibility(pub(crate)), context(suffix(false)))]
+pub(crate) enum CsiDriverError {
+    #[snafu(display("{source}"))]
+    Fsfreeze { source: FsfreezeError },
+    #[snafu(display("{source}"))]
+    Mount { source: MountError },
+    #[snafu(display("Not a valid mayastor csi driver command"))]
+    InvalidCsiDriverCommand,
+}
+
+impl From<FsfreezeError> for CsiDriverError {
+    fn from(value: FsfreezeError) -> Self {
+        Self::Fsfreeze { source: value }
+    }
+}
+
+impl From<MountError> for CsiDriverError {
+    fn from(value: MountError) -> Self {
+        Self::Mount { source: value }
+    }
+}
+
+impl From<CsiDriverError> for ExitCode {
+    fn from(value: CsiDriverError) -> Self {
+        match value {
+            CsiDriverError::Fsfreeze { source } => source.into(),
+            CsiDriverError::Mount { source } => source.into(),
+            CsiDriverError::InvalidCsiDriverCommand => ExitCode::from(Errno::EINVAL as u8),
         }
     }
 }

control-plane/csi-driver/src/bin/node/filesystem_ops.rs

@@ -5,9 +5,9 @@
 
 use crate::{findmnt::get_devicepath, mount};
 use csi_driver::filesystem::FileSystem as Fs;
+use devinfo::{blkid::probe::Probe, mountinfo::MountInfo, DevInfoError};
 
 use anyhow::anyhow;
-use devinfo::{blkid::probe::Probe, mountinfo::MountInfo, DevInfoError};
 use std::{process::Output, str, str::FromStr};
 use tokio::process::Command;
 use tonic::async_trait;
@@ -95,7 +95,7 @@ pub(crate) trait FileSystemOps: Send + Sync {
     /// Get the default mount options along with the user passed options for specific filesystems.
     fn mount_flags(&self, mount_flags: Vec<String>) -> Vec<String>;
     /// Unmount the filesystem if the filesystem uuid and the provided uuid differ.
-    fn unmount_on_fs_id_diff(
+    async fn unmount_on_fs_id_diff(
         &self,
         device_path: &str,
         fs_staging_path: &str,
@@ -146,7 +146,7 @@ impl FileSystemOps for Ext4Fs {
         mount_flags
     }
 
-    fn unmount_on_fs_id_diff(
+    async fn unmount_on_fs_id_diff(
         &self,
         _device_path: &str,
         _fs_staging_path: &str,
@@ -270,13 +270,15 @@ impl FileSystemOps for XFs {
         mount_flags
     }
 
-    fn unmount_on_fs_id_diff(
+    async fn unmount_on_fs_id_diff(
         &self,
         device_path: &str,
         fs_staging_path: &str,
         volume_uuid: &Uuid,
     ) -> Result<(), Error> {
         mount::unmount_on_fs_id_diff(device_path, fs_staging_path, volume_uuid)
+            .await
+            .map_err(|error| error.to_string())
     }
 
     /// Xfs filesystem needs an unmount to clear the log, so that the parameters can be changed.
@@ -288,12 +290,12 @@ impl FileSystemOps for XFs {
         options: &[String],
         volume_uuid: &Uuid,
     ) -> Result<(), Error> {
-        mount::filesystem_mount(device, staging_path, &FileSystem(Fs::Xfs), options).map_err(|error| {
+        mount::filesystem_mount(device, staging_path, &FileSystem(Fs::Xfs), options).await.map_err(|error| {
             format!(
                 "(xfs repairing) Failed to mount device {device} onto {staging_path} for {volume_uuid} : {error}",
             )
         })?;
-        mount::filesystem_unmount(staging_path).map_err(|error| {
+        mount::filesystem_unmount(staging_path).await.map_err(|error| {
             format!(
                 "(xfs repairing) Failed to unmount device {device} from {staging_path} for {volume_uuid} : {error}",
             )
@@ -358,13 +360,13 @@ impl FileSystemOps for BtrFs {
         mount_flags
     }
 
-    fn unmount_on_fs_id_diff(
+    async fn unmount_on_fs_id_diff(
         &self,
         device_path: &str,
         fs_staging_path: &str,
         volume_uuid: &Uuid,
     ) -> Result<(), Error> {
-        mount::unmount_on_fs_id_diff(device_path, fs_staging_path, volume_uuid)
+        mount::unmount_on_fs_id_diff(device_path, fs_staging_path, volume_uuid).await
     }
 
     /// `btrfs check --readonly` is a not a `DANGEROUS OPTION` as it only exists to calm potential

control-plane/csi-driver/src/bin/node/filesystem_vol.rs

@@ -91,7 +91,7 @@ pub(crate) async fn stage_fs_volume(
         // If clone's fs id change was requested and we were not able to change it in first attempt
         // unmount and continue the stage again.
         let continue_stage = if fs_id.is_some() {
-            continue_after_unmount_on_fs_id_diff(fstype ,device_path, fs_staging_path, &volume_uuid)
+            continue_after_unmount_on_fs_id_diff(fstype ,device_path, fs_staging_path, &volume_uuid).await
                 .map_err(|error| {
                     failure!(
                     Code::FailedPrecondition,
@@ -107,7 +107,7 @@ pub(crate) async fn stage_fs_volume(
         if !continue_stage {
             // todo: validate other flags?
             if mnt.mount_flags.readonly() != existing.options.readonly() {
-                mount::remount(fs_staging_path, mnt.mount_flags.readonly())?;
+                mount::remount(fs_staging_path, mnt.mount_flags.readonly()).await?;
             }
 
             return Ok(());
@@ -161,7 +161,8 @@ pub(crate) async fn stage_fs_volume(
 
     debug!("Mounting device {} onto {}", device_path, fs_staging_path);
 
-    if let Err(error) = mount::filesystem_mount(device_path, fs_staging_path, fstype, &mount_flags)
+    if let Err(error) =
+        mount::filesystem_mount(device_path, fs_staging_path, fstype, &mount_flags).await
     {
         return Err(failure!(
             Code::Internal,
@@ -209,7 +210,7 @@ pub(crate) async fn unstage_fs_volume(msg: &NodeUnstageVolumeRequest) -> Result<
             ));
         }
 
-        if let Err(error) = mount::filesystem_unmount(fs_staging_path) {
+        if let Err(error) = mount::filesystem_unmount(fs_staging_path).await {
             return Err(failure!(
                 Code::Internal,
                 "Failed to unstage volume {}: failed to unmount device {:?} from {}: {}",
@@ -227,7 +228,7 @@ pub(crate) async fn unstage_fs_volume(msg: &NodeUnstageVolumeRequest) -> Result<
 }
 
 /// Publish a filesystem volume
-pub(crate) fn publish_fs_volume(
+pub(crate) async fn publish_fs_volume(
     msg: &NodePublishVolumeRequest,
     mnt: &MountVolume,
     filesystems: &[FileSystem],
@@ -328,7 +329,7 @@ pub(crate) fn publish_fs_volume(
 
     debug!("Mounting {} to {}", fs_staging_path, target_path);
 
-    if let Err(error) = mount::bind_mount(fs_staging_path, target_path, false) {
+    if let Err(error) = mount::bind_mount(fs_staging_path, target_path, false).await {
         return Err(failure!(
             Code::Internal,
             "Failed to publish volume {}: failed to mount {} to {}: {}",
@@ -345,7 +346,7 @@ pub(crate) fn publish_fs_volume(
 
         debug!("Remounting {} as readonly", target_path);
 
-        if let Err(error) = mount::bind_remount(target_path, &options) {
+        if let Err(error) = mount::bind_remount(target_path, &options).await {
             let message = format!(
                 "Failed to publish volume {volume_id}: failed to mount {fs_staging_path} to {target_path} as readonly: {error}"
             );
@@ -354,7 +355,7 @@ pub(crate) fn publish_fs_volume(
 
             debug!("Unmounting {}", target_path);
 
-            if let Err(error) = mount::bind_unmount(target_path) {
+            if let Err(error) = mount::bind_unmount(target_path).await {
                 error!("Failed to unmount {}: {}", target_path, error);
             }
 
@@ -367,7 +368,7 @@ pub(crate) fn publish_fs_volume(
     Ok(())
 }
 
-pub(crate) fn unpublish_fs_volume(msg: &NodeUnpublishVolumeRequest) -> Result<(), Status> {
+pub(crate) async fn unpublish_fs_volume(msg: &NodeUnpublishVolumeRequest) -> Result<(), Status> {
     // filesystem mount
     let target_path = &msg.target_path;
     let volume_id = &msg.volume_id;
@@ -398,7 +399,7 @@ pub(crate) fn unpublish_fs_volume(msg: &NodeUnpublishVolumeRequest) -> Result<()
 
     debug!("Unmounting {}", target_path);
 
-    if let Err(error) = mount::bind_unmount(target_path) {
+    if let Err(error) = mount::bind_unmount(target_path).await {
         return Err(failure!(
             Code::Internal,
             "Failed to unpublish volume {}: failed to unmount {}: {}",
@@ -427,14 +428,15 @@ pub(crate) fn unpublish_fs_volume(msg: &NodeUnpublishVolumeRequest) -> Result<()
 
 /// Check if we can continue the staging incase the change fs id failed mid way and we want to retry
 /// the flow.
-fn continue_after_unmount_on_fs_id_diff(
+async fn continue_after_unmount_on_fs_id_diff(
     fstype: &FileSystem,
     device_path: &str,
     fs_staging_path: &str,
     volume_uuid: &Uuid,
 ) -> Result<bool, String> {
     fstype
         .fs_ops()?
-        .unmount_on_fs_id_diff(device_path, fs_staging_path, volume_uuid)?;
+        .unmount_on_fs_id_diff(device_path, fs_staging_path, volume_uuid)
+        .await?;
     Ok(fstype == &Fs::Xfs.into())
 }

control-plane/csi-driver/src/bin/node/main.rs

@@ -1,4 +1,4 @@
-use crate::error::FsfreezeError;
+use crate::error::CsiDriverError;
 use std::process::ExitCode;
 
 /// todo: cleanup this module cfg repetition..
@@ -31,6 +31,8 @@ mod match_dev;
 #[cfg(target_os = "linux")]
 mod mount;
 #[cfg(target_os = "linux")]
+mod mount_utils;
+#[cfg(target_os = "linux")]
 mod node;
 #[cfg(target_os = "linux")]
 mod nodeplugin_grpc;
@@ -52,7 +54,7 @@ async fn main() -> anyhow::Result<ExitCode> {
         error
     }) {
         Ok(_) => Ok(ExitCode::SUCCESS),
-        Err(error) => match error.downcast::<FsfreezeError>() {
+        Err(error) => match error.downcast::<CsiDriverError>() {
             Ok(error) => Ok(error.into()),
             Err(error) => Err(error),
         },