Merge #1550

1550: feat(replica/resize): add support for resizing a replica r=dsharma-dc a=dsharma-dc

Added the support for resizing a replica. The replicas need to be expanded as part of the volume expand workflow.

Co-authored-by: Diwakar Sharma <[email protected]>

io-engine/src/bin/io-engine-client/v1/replica_cli.rs

@@ -32,7 +32,6 @@ pub fn subcommands() -> Command {
             Arg::new("size")
                 .short('s')
                 .long("size")
-
                 .required(true)
                 .value_name("NUMBER")
                 .help("Size of the replica"))
@@ -107,6 +106,20 @@ pub fn subcommands() -> Command {
             .index(1)
             .help("Replica uuid"),
     );
+    let resize = Command::new("resize")
+        .about("Resize replica")
+        .arg(
+            Arg::new("uuid")
+                .required(true)
+                .index(1)
+                .help("Replica uuid"),
+        )
+        .arg(
+            Arg::new("size")
+                .required(true)
+                .index(2)
+                .help("Requested new size of the replica"),
+        );
     Command::new("replica")
         .subcommand_required(true)
         .arg_required_else_help(true)
@@ -115,6 +128,7 @@ pub fn subcommands() -> Command {
         .subcommand(destroy)
         .subcommand(share)
         .subcommand(unshare)
+        .subcommand(resize)
         .subcommand(Command::new("list").about("List replicas"))
         .subcommand(Command::new("stats").about("IO stats of replicas"))
 }
@@ -126,6 +140,7 @@ pub async fn handler(ctx: Context, matches: &ArgMatches) -> crate::Result<()> {
         ("list", args) => replica_list(ctx, args).await,
         ("share", args) => replica_share(ctx, args).await,
         ("unshare", args) => replica_unshare(ctx, args).await,
+        ("resize", args) => replica_resize(ctx, args).await,
         ("stats", args) => replica_stat(ctx, args).await,
         (cmd, _) => {
             Err(Status::not_found(format!("command {cmd} does not exist")))
@@ -412,6 +427,46 @@ async fn replica_unshare(
     Ok(())
 }
 
+async fn replica_resize(
+    mut ctx: Context,
+    matches: &ArgMatches,
+) -> crate::Result<()> {
+    let uuid = matches
+        .get_one::<String>("uuid")
+        .ok_or_else(|| ClientError::MissingValue {
+            field: "uuid".to_string(),
+        })?
+        .to_owned();
+
+    let requested_size =
+        parse_size(matches.get_one::<String>("size").ok_or_else(|| {
+            ClientError::MissingValue {
+                field: "size".to_string(),
+            }
+        })?)
+        .map_err(|s| Status::invalid_argument(format!("Bad size '{s}'")))
+        .context(GrpcStatus)?;
+
+    let _ = ctx
+        .v1
+        .replica
+        .resize_replica(v1_rpc::replica::ResizeReplicaRequest {
+            uuid: uuid.clone(),
+            requested_size: requested_size.get_bytes() as u64,
+        })
+        .await
+        .context(GrpcStatus)?;
+
+    match ctx.output {
+        OutputFormat::Json => {}
+        OutputFormat::Default => {
+            println!("replica {} is resized", &uuid);
+        }
+    }
+
+    Ok(())
+}
+
 // TODO : There's no v1 rpc for stat.
 async fn replica_stat(
     mut ctx: Context,

io-engine/src/grpc/v0/mayastor_grpc.rs

@@ -249,6 +249,18 @@ impl From<LvsError> for tonic::Status {
                 Errno::EMEDIUMTYPE => Status::aborted(e.to_string()),
                 _ => Status::internal(e.to_string()),
             },
+            LvsError::RepResize {
+                source, ..
+            } => match source {
+                Errno::ENOSPC | Errno::ENOMEM => {
+                    Status::resource_exhausted(e.to_string())
+                }
+                Errno::EPERM => Status::permission_denied(e.to_string()),
+                Errno::EINVAL | Errno::ENOENT => {
+                    Status::invalid_argument(e.to_string())
+                }
+                _ => Status::internal(e.to_string()),
+            },
             LvsError::RepExists {
                 ..
             } => Status::already_exists(e.to_string()),

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

@@ -489,4 +489,36 @@ impl ReplicaRpc for ReplicaService {
         )
         .await
     }
+
+    #[named]
+    async fn resize_replica(
+        &self,
+        request: Request<ResizeReplicaRequest>,
+    ) -> GrpcResult<Replica> {
+        self.locked(
+            GrpcClientContext::new(&request, function_name!()),
+            async move {
+                let args = request.into_inner();
+                info!("{args:?}");
+                let rx = rpc_submit::<_, _, LvsError>(async move {
+                    let mut lvol = Bdev::lookup_by_uuid_str(&args.uuid)
+                        .and_then(|b| Lvol::try_from(b).ok())
+                        .ok_or(LvsError::RepResize {
+                            source: Errno::ENOENT,
+                            name: args.uuid.to_owned(),
+                        })?;
+                    let requested_size = args.requested_size;
+                    lvol.resize_replica(requested_size).await?;
+                    debug!("resized {:?}", lvol);
+                    Ok(Replica::from(lvol))
+                })?;
+
+                rx.await
+                    .map_err(|_| Status::cancelled("cancelled"))?
+                    .map_err(Status::from)
+                    .map(Response::new)
+            },
+        )
+        .await
+    }
 }

io-engine/src/lvs/lvs_error.rs

@@ -86,6 +86,11 @@ pub enum Error {
         name: String,
         msg: String,
     },
+    #[snafu(display("failed to resize lvol {}", name))]
+    RepResize {
+        source: Errno,
+        name: String,
+    },
     #[snafu(display("bdev {} is not a lvol", name))]
     NotALvol {
         source: Errno,
@@ -218,6 +223,9 @@ impl ToErrno for Error {
             Self::RepDestroy {
                 source, ..
             } => source,
+            Self::RepResize {
+                source, ..
+            } => source,
             Self::NotALvol {
                 source, ..
             } => source,

io-engine/src/lvs/lvs_lvol.rs

@@ -30,6 +30,7 @@ use spdk_rs::libspdk::{
     spdk_lvol,
     vbdev_lvol_destroy,
     vbdev_lvol_get_from_bdev,
+    vbdev_lvol_resize,
     LVS_CLEAR_WITH_UNMAP,
 };
 
@@ -114,6 +115,16 @@ impl Display for PropName {
     }
 }
 
+/// Resize context to be passed as callback args pointer to spdk.
+struct ResizeCbCtx {
+    /// The lvol to be resized.
+    lvol: *mut spdk_lvol,
+    /// Oneshot sender for sending resize operation result.
+    sender: *mut c_void,
+    /// The new requested size of lvol.
+    req_size: u64,
+}
+
 /// Lvol space usage.
 #[derive(Default, Copy, Clone, Debug)]
 pub struct LvolSpaceUsage {
@@ -594,6 +605,11 @@ pub trait LvsLvol: LogicalVolume + Share {
     /// Wrapper function to destroy replica and its associated snapshot if
     /// replica is identified as last clone.
     async fn destroy_replica(mut self) -> Result<String, Error>;
+
+    /// Resize a replica. The resize can be expand or shrink, depending
+    /// upon if required size is more or less than current size of
+    /// the replpica.
+    async fn resize_replica(&mut self, resize_to: u64) -> Result<(), Error>;
 }
 
 ///  LogicalVolume implement Generic interface for Lvol.
@@ -1026,4 +1042,64 @@ impl LvsLvol for Lvol {
         }
         Ok(name)
     }
+
+    /// Resize a replica. The resize can be expand or shrink, depending
+    /// upon if required size is more or less than current size of
+    /// the replica.
+    async fn resize_replica(&mut self, resize_to: u64) -> Result<(), Error> {
+        let (s, r) = pair::<ErrnoResult<*mut spdk_lvol>>();
+        let mut ctx = ResizeCbCtx {
+            lvol: self.as_inner_ptr(),
+            sender: cb_arg(s),
+            req_size: resize_to,
+        };
+
+        unsafe {
+            vbdev_lvol_resize(
+                self.as_inner_ptr(),
+                resize_to,
+                Some(lvol_resize_cb),
+                &mut ctx as *mut _ as *mut c_void,
+            );
+        }
+
+        let cb_ret = r.await.expect("lvol resize callback dropped");
+
+        match cb_ret {
+            Ok(_) => {
+                info!("Resized {:?} successfully", self);
+                Ok(())
+            }
+            Err(errno) => {
+                error!("Resize {:?} failed, errno {errno}", self);
+                Err(Error::RepResize {
+                    source: errno,
+                    name: self.name(),
+                })
+            }
+        }
+    }
+}
+
+extern "C" fn lvol_resize_cb(cb_arg: *mut c_void, errno: i32) {
+    let mut retcode = errno;
+    let ctx = cb_arg as *mut ResizeCbCtx;
+    let (lvol, req_size) =
+        unsafe { (Lvol::from_inner_ptr((*ctx).lvol), (*ctx).req_size) };
+    let sender = unsafe {
+        Box::from_raw(
+            (*ctx).sender as *mut oneshot::Sender<ErrnoResult<*mut spdk_lvol>>,
+        )
+    };
+
+    if retcode == 0 && (lvol.size() < req_size) {
+        // Make sure resize worked, and account for metadata while comparing
+        // i.e. the actual size will be a little more than requested.
+        debug_assert!(false, "errno 0 - replica resize must have succeeded !");
+        retcode = -libc::EAGAIN;
+    }
+
+    sender
+        .send(errno_result_from_i32(lvol.as_inner_ptr(), retcode))
+        .expect("Receiver is gone");
 }