add remote support: delete_version,rename_part,delete_paths,update_metadata

Signed-off-by: junxiang Mu <1948535941@qq.com>

common/protos/src/generated/flatbuffers_generated/models.rs

@@ -1,10 +1,9 @@
 // automatically generated by the FlatBuffers compiler, do not modify
 
-
 // @generated
 
-use core::mem;
 use core::cmp::Ordering;
+use core::mem;
 
 extern crate flatbuffers;
 use self::flatbuffers::{EndianScalar, Follow};
@@ -12,112 +11,114 @@ use self::flatbuffers::{EndianScalar, Follow};
 #[allow(unused_imports, dead_code)]
 pub mod models {
 
-  use core::mem;
-  use core::cmp::Ordering;
+    use core::cmp::Ordering;
+    use core::mem;
 
-  extern crate flatbuffers;
-  use self::flatbuffers::{EndianScalar, Follow};
+    extern crate flatbuffers;
+    use self::flatbuffers::{EndianScalar, Follow};
 
-pub enum PingBodyOffset {}
-#[derive(Copy, Clone, PartialEq)]
+    pub enum PingBodyOffset {}
+    #[derive(Copy, Clone, PartialEq)]
 
-pub struct PingBody<'a> {
-  pub _tab: flatbuffers::Table<'a>,
-}
-
-impl<'a> flatbuffers::Follow<'a> for PingBody<'a> {
-  type Inner = PingBody<'a>;
-  #[inline]
-  unsafe fn follow(buf: &'a [u8], loc: usize) -> Self::Inner {
-    Self { _tab: flatbuffers::Table::new(buf, loc) }
-  }
-}
-
-impl<'a> PingBody<'a> {
-  pub const VT_PAYLOAD: flatbuffers::VOffsetT = 4;
-
-  pub const fn get_fully_qualified_name() -> &'static str {
-    "models.PingBody"
-  }
-
-  #[inline]
-  pub unsafe fn init_from_table(table: flatbuffers::Table<'a>) -> Self {
-    PingBody { _tab: table }
-  }
-  #[allow(unused_mut)]
-  pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr, A: flatbuffers::Allocator + 'bldr>(
-    _fbb: &'mut_bldr mut flatbuffers::FlatBufferBuilder<'bldr, A>,
-    args: &'args PingBodyArgs<'args>
-  ) -> flatbuffers::WIPOffset<PingBody<'bldr>> {
-    let mut builder = PingBodyBuilder::new(_fbb);
-    if let Some(x) = args.payload { builder.add_payload(x); }
-    builder.finish()
-  }
-
-
-  #[inline]
-  pub fn payload(&self) -> Option<flatbuffers::Vector<'a, u8>> {
-    // Safety:
-    // Created from valid Table for this object
-    // which contains a valid value in this slot
-    unsafe { self._tab.get::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'a, u8>>>(PingBody::VT_PAYLOAD, None)}
-  }
-}
-
-impl flatbuffers::Verifiable for PingBody<'_> {
-  #[inline]
-  fn run_verifier(
-    v: &mut flatbuffers::Verifier, pos: usize
-  ) -> Result<(), flatbuffers::InvalidFlatbuffer> {
-    use self::flatbuffers::Verifiable;
-    v.visit_table(pos)?
-     .visit_field::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'_, u8>>>("payload", Self::VT_PAYLOAD, false)?
-     .finish();
-    Ok(())
-  }
-}
-pub struct PingBodyArgs<'a> {
-    pub payload: Option<flatbuffers::WIPOffset<flatbuffers::Vector<'a, u8>>>,
-}
-impl<'a> Default for PingBodyArgs<'a> {
-  #[inline]
-  fn default() -> Self {
-    PingBodyArgs {
-      payload: None,
+    pub struct PingBody<'a> {
+        pub _tab: flatbuffers::Table<'a>,
     }
-  }
-}
 
-pub struct PingBodyBuilder<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> {
-  fbb_: &'b mut flatbuffers::FlatBufferBuilder<'a, A>,
-  start_: flatbuffers::WIPOffset<flatbuffers::TableUnfinishedWIPOffset>,
-}
-impl<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> PingBodyBuilder<'a, 'b, A> {
-  #[inline]
-  pub fn add_payload(&mut self, payload: flatbuffers::WIPOffset<flatbuffers::Vector<'b , u8>>) {
-    self.fbb_.push_slot_always::<flatbuffers::WIPOffset<_>>(PingBody::VT_PAYLOAD, payload);
-  }
-  #[inline]
-  pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a, A>) -> PingBodyBuilder<'a, 'b, A> {
-    let start = _fbb.start_table();
-    PingBodyBuilder {
-      fbb_: _fbb,
-      start_: start,
+    impl<'a> flatbuffers::Follow<'a> for PingBody<'a> {
+        type Inner = PingBody<'a>;
+        #[inline]
+        unsafe fn follow(buf: &'a [u8], loc: usize) -> Self::Inner {
+            Self {
+                _tab: flatbuffers::Table::new(buf, loc),
+            }
+        }
     }
-  }
-  #[inline]
-  pub fn finish(self) -> flatbuffers::WIPOffset<PingBody<'a>> {
-    let o = self.fbb_.end_table(self.start_);
-    flatbuffers::WIPOffset::new(o.value())
-  }
-}
 
-impl core::fmt::Debug for PingBody<'_> {
-  fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
-    let mut ds = f.debug_struct("PingBody");
-      ds.field("payload", &self.payload());
-      ds.finish()
-  }
-}
-}  // pub mod models
+    impl<'a> PingBody<'a> {
+        pub const VT_PAYLOAD: flatbuffers::VOffsetT = 4;
 
+        pub const fn get_fully_qualified_name() -> &'static str {
+            "models.PingBody"
+        }
+
+        #[inline]
+        pub unsafe fn init_from_table(table: flatbuffers::Table<'a>) -> Self {
+            PingBody { _tab: table }
+        }
+        #[allow(unused_mut)]
+        pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr, A: flatbuffers::Allocator + 'bldr>(
+            _fbb: &'mut_bldr mut flatbuffers::FlatBufferBuilder<'bldr, A>,
+            args: &'args PingBodyArgs<'args>,
+        ) -> flatbuffers::WIPOffset<PingBody<'bldr>> {
+            let mut builder = PingBodyBuilder::new(_fbb);
+            if let Some(x) = args.payload {
+                builder.add_payload(x);
+            }
+            builder.finish()
+        }
+
+        #[inline]
+        pub fn payload(&self) -> Option<flatbuffers::Vector<'a, u8>> {
+            // Safety:
+            // Created from valid Table for this object
+            // which contains a valid value in this slot
+            unsafe {
+                self._tab
+                    .get::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'a, u8>>>(PingBody::VT_PAYLOAD, None)
+            }
+        }
+    }
+
+    impl flatbuffers::Verifiable for PingBody<'_> {
+        #[inline]
+        fn run_verifier(v: &mut flatbuffers::Verifier, pos: usize) -> Result<(), flatbuffers::InvalidFlatbuffer> {
+            use self::flatbuffers::Verifiable;
+            v.visit_table(pos)?
+                .visit_field::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'_, u8>>>("payload", Self::VT_PAYLOAD, false)?
+                .finish();
+            Ok(())
+        }
+    }
+    pub struct PingBodyArgs<'a> {
+        pub payload: Option<flatbuffers::WIPOffset<flatbuffers::Vector<'a, u8>>>,
+    }
+    impl<'a> Default for PingBodyArgs<'a> {
+        #[inline]
+        fn default() -> Self {
+            PingBodyArgs { payload: None }
+        }
+    }
+
+    pub struct PingBodyBuilder<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> {
+        fbb_: &'b mut flatbuffers::FlatBufferBuilder<'a, A>,
+        start_: flatbuffers::WIPOffset<flatbuffers::TableUnfinishedWIPOffset>,
+    }
+    impl<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> PingBodyBuilder<'a, 'b, A> {
+        #[inline]
+        pub fn add_payload(&mut self, payload: flatbuffers::WIPOffset<flatbuffers::Vector<'b, u8>>) {
+            self.fbb_
+                .push_slot_always::<flatbuffers::WIPOffset<_>>(PingBody::VT_PAYLOAD, payload);
+        }
+        #[inline]
+        pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a, A>) -> PingBodyBuilder<'a, 'b, A> {
+            let start = _fbb.start_table();
+            PingBodyBuilder {
+                fbb_: _fbb,
+                start_: start,
+            }
+        }
+        #[inline]
+        pub fn finish(self) -> flatbuffers::WIPOffset<PingBody<'a>> {
+            let o = self.fbb_.end_table(self.start_);
+            flatbuffers::WIPOffset::new(o.value())
+        }
+    }
+
+    impl core::fmt::Debug for PingBody<'_> {
+        fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+            let mut ds = f.debug_struct("PingBody");
+            ds.field("payload", &self.payload());
+            ds.finish()
+        }
+    }
+} // pub mod models

common/protos/src/node.proto

@@ -88,6 +88,20 @@ message DeleteResponse {
   optional string error_info = 2;
 }
 
+message RenamePartRequst {
+  string disk = 1;
+  string src_volume = 2;
+  string src_path = 3;
+  string dst_volume = 4;
+  string dst_path = 5;
+  bytes meta = 6;
+}
+
+message RenamePartResponse {
+  bool success = 1;
+  optional string error_info = 2;
+}
+
 message RenameFileRequst {
   string disk = 1;
   string src_volume = 2;
@@ -219,6 +233,30 @@ message StatVolumeResponse {
   optional string error_info = 3;
 }
 
+message DeletePathsRequest {
+  string disk = 1;
+  string volume = 2;
+  repeated string paths = 3;
+}
+
+message DeletePathsResponse {
+  bool success = 1;
+  optional string error_info = 2;
+}
+
+message UpdateMetadataRequest {
+  string disk = 1;
+  string volume = 2;
+  string path = 3;
+  string file_info = 4;
+  string opts = 5;
+}
+
+message UpdateMetadataResponse {
+  bool success = 1;
+  optional string error_info = 2;
+}
+
 message WriteMetadataRequest {
   string disk = 1;  // indicate which one in the disks
   string volume = 2;
@@ -258,6 +296,21 @@ message ReadXLResponse {
   optional string error_info = 3;
 }
 
+message DeleteVersionRequest {
+  string disk = 1;
+  string volume = 2;
+  string path = 3;
+  string file_info = 4;
+  bool force_del_marker = 5;
+  string opts = 6;
+}
+
+message DeleteVersionResponse {
+  bool success = 1;
+  string raw_file_info = 2;
+  optional string error_info = 3;
+}
+
 message DeleteVersionsRequest {
   string disk = 1;
   string volume = 2;
@@ -307,6 +360,7 @@ service NodeService {
   rpc ReadAll(ReadAllRequest) returns (ReadAllResponse) {};
   rpc WriteAll(WriteAllRequest) returns (WriteAllResponse) {};
   rpc Delete(DeleteRequest) returns (DeleteResponse) {};
+  rpc RenamePart(RenamePartRequst) returns (RenamePartResponse) {};
   rpc RenameFile(RenameFileRequst) returns (RenameFileResponse) {};
   rpc Write(WriteRequest) returns (WriteResponse) {};
 //  rpc Append(AppendRequest) returns (AppendResponse) {};
@@ -318,9 +372,12 @@ service NodeService {
   rpc MakeVolume(MakeVolumeRequest) returns (MakeVolumeResponse) {};
   rpc ListVolumes(ListVolumesRequest) returns (ListVolumesResponse) {};
   rpc StatVolume(StatVolumeRequest) returns (StatVolumeResponse) {};
+  rpc DeletePaths(DeletePathsRequest) returns (DeletePathsResponse) {};
+  rpc UpdateMetadata(UpdateMetadataRequest) returns (UpdateMetadataResponse) {};
   rpc WriteMetadata(WriteMetadataRequest) returns (WriteMetadataResponse) {};
   rpc ReadVersion(ReadVersionRequest) returns (ReadVersionResponse) {};
   rpc ReadXL(ReadXLRequest) returns (ReadXLResponse) {};
+  rpc DeleteVersion(DeleteVersionRequest) returns (DeleteVersionResponse) {};
   rpc DeleteVersions(DeleteVersionsRequest) returns (DeleteVersionsResponse) {};
   rpc ReadMultiple(ReadMultipleRequest) returns (ReadMultipleResponse) {};
   rpc DeleteVolume(DeleteVolumeRequest) returns (DeleteVolumeResponse) {};

ecstore/src/disk/mod.rs

@@ -129,6 +129,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     async fn read_all(&self, volume: &str, path: &str) -> Result<Vec<u8>>;
 }
 
+#[derive(Debug, Serialize, Deserialize)]
 pub struct UpdateMetadataOpts {
     pub no_persistence: bool,
 }

ecstore/src/disk/remote.rs

@@ -5,10 +5,10 @@ use futures::lock::Mutex;
 use protos::{
     node_service_time_out_client,
     proto_gen::node_service::{
-        node_service_client::NodeServiceClient, DeleteRequest, DeleteVersionsRequest, DeleteVolumeRequest, ListDirRequest,
-        ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, ReadAllRequest, ReadMultipleRequest, ReadVersionRequest,
-        ReadXlRequest, RenameDataRequest, RenameFileRequst, StatVolumeRequest, WalkDirRequest, WriteAllRequest,
-        WriteMetadataRequest,
+        node_service_client::NodeServiceClient, DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest,
+        DeleteVolumeRequest, ListDirRequest, ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, ReadAllRequest,
+        ReadMultipleRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest, RenameFileRequst, RenamePartRequst,
+        StatVolumeRequest, UpdateMetadataRequest, WalkDirRequest, WriteAllRequest, WriteMetadataRequest,
     },
     DEFAULT_GRPC_SERVER_MESSAGE_LEN,
 };
@@ -201,15 +201,25 @@ impl DiskAPI for RemoteDisk {
 
         Ok(())
     }
-    async fn rename_part(
-        &self,
-        _src_volume: &str,
-        _src_path: &str,
-        _dst_volume: &str,
-        _dst_path: &str,
-        _meta: Vec<u8>,
-    ) -> Result<()> {
-        unimplemented!()
+    async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec<u8>) -> Result<()> {
+        info!("rename_part");
+        let mut client = self.get_client_v2().await?;
+        let request = Request::new(RenamePartRequst {
+            disk: self.root.to_string_lossy().to_string(),
+            src_volume: src_volume.to_string(),
+            src_path: src_path.to_string(),
+            dst_volume: dst_volume.to_string(),
+            dst_path: dst_path.to_string(),
+            meta,
+        });
+
+        let response = client.rename_part(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
+
+        Ok(())
     }
     async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
         info!("rename_file");
@@ -410,13 +420,45 @@ impl DiskAPI for RemoteDisk {
         Ok(volume_info)
     }
 
-    async fn delete_paths(&self, _volume: &str, _paths: &[&str]) -> Result<()> {
-        // TODO:
-        unimplemented!()
+    async fn delete_paths(&self, volume: &str, paths: &[&str]) -> Result<()> {
+        info!("delete_paths");
+        let paths = paths.iter().map(|s| s.to_string()).collect::<Vec<String>>();
+        let mut client = self.get_client_v2().await?;
+        let request = Request::new(DeletePathsRequest {
+            disk: self.root.to_string_lossy().to_string(),
+            volume: volume.to_string(),
+            paths,
+        });
+
+        let response = client.delete_paths(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
+
+        Ok(())
     }
-    async fn update_metadata(&self, _volume: &str, _path: &str, _fi: FileInfo, _opts: UpdateMetadataOpts) -> Result<()> {
-        // TODO:
-        unimplemented!()
+    async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: UpdateMetadataOpts) -> Result<()> {
+        info!("update_metadata");
+        let file_info = serde_json::to_string(&fi)?;
+        let opts = serde_json::to_string(&opts)?;
+
+        let mut client = self.get_client_v2().await?;
+        let request = Request::new(UpdateMetadataRequest {
+            disk: self.root.to_string_lossy().to_string(),
+            volume: volume.to_string(),
+            path: path.to_string(),
+            file_info,
+            opts,
+        });
+
+        let response = client.update_metadata(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
+
+        Ok(())
     }
 
     async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
@@ -491,13 +533,35 @@ impl DiskAPI for RemoteDisk {
     }
     async fn delete_version(
         &self,
-        _volume: &str,
-        _path: &str,
-        _fi: FileInfo,
-        _force_del_marker: bool,
-        _opts: DeleteOptions,
+        volume: &str,
+        path: &str,
+        fi: FileInfo,
+        force_del_marker: bool,
+        opts: DeleteOptions,
     ) -> Result<RawFileInfo> {
-        unimplemented!()
+        info!("delete_version");
+        let file_info = serde_json::to_string(&fi)?;
+        let opts = serde_json::to_string(&opts)?;
+
+        let mut client = self.get_client_v2().await?;
+        let request = Request::new(DeleteVersionRequest {
+            disk: self.root.to_string_lossy().to_string(),
+            volume: volume.to_string(),
+            path: path.to_string(),
+            file_info,
+            force_del_marker,
+            opts,
+        });
+
+        let response = client.delete_version(request).await?.into_inner();
+
+        if !response.success {
+            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
+        }
+
+        let raw_file_info = serde_json::from_str::<RawFileInfo>(&response.raw_file_info)?;
+
+        Ok(raw_file_info)
     }
     async fn delete_versions(
         &self,

rustfs/src/grpc.rs

@@ -1,5 +1,5 @@
 use ecstore::{
-    disk::{DeleteOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, WalkDirOptions},
+    disk::{DeleteOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, UpdateMetadataOpts, WalkDirOptions},
     erasure::{ReadAt, Write},
     peer::{LocalPeerS3Client, PeerS3Client},
     store::{all_local_disk_path, find_local_disk},
@@ -12,14 +12,16 @@ use protos::{
     models::{PingBody, PingBodyBuilder},
     proto_gen::node_service::{
         node_service_server::{NodeService as Node, NodeServiceServer as NodeServer},
-        DeleteBucketRequest, DeleteBucketResponse, DeleteRequest, DeleteResponse, DeleteVersionsRequest, DeleteVersionsResponse,
-        DeleteVolumeRequest, DeleteVolumeResponse, GetBucketInfoRequest, GetBucketInfoResponse, ListBucketRequest,
-        ListBucketResponse, ListDirRequest, ListDirResponse, ListVolumesRequest, ListVolumesResponse, MakeBucketRequest,
-        MakeBucketResponse, MakeVolumeRequest, MakeVolumeResponse, MakeVolumesRequest, MakeVolumesResponse, PingRequest,
-        PingResponse, ReadAllRequest, ReadAllResponse, ReadAtRequest, ReadAtResponse, ReadMultipleRequest, ReadMultipleResponse,
-        ReadVersionRequest, ReadVersionResponse, ReadXlRequest, ReadXlResponse, RenameDataRequest, RenameDataResponse,
-        RenameFileRequst, RenameFileResponse, StatVolumeRequest, StatVolumeResponse, WalkDirRequest, WalkDirResponse,
-        WriteAllRequest, WriteAllResponse, WriteMetadataRequest, WriteMetadataResponse, WriteRequest, WriteResponse,
+        DeleteBucketRequest, DeleteBucketResponse, DeletePathsRequest, DeletePathsResponse, DeleteRequest, DeleteResponse,
+        DeleteVersionRequest, DeleteVersionResponse, DeleteVersionsRequest, DeleteVersionsResponse, DeleteVolumeRequest,
+        DeleteVolumeResponse, GetBucketInfoRequest, GetBucketInfoResponse, ListBucketRequest, ListBucketResponse, ListDirRequest,
+        ListDirResponse, ListVolumesRequest, ListVolumesResponse, MakeBucketRequest, MakeBucketResponse, MakeVolumeRequest,
+        MakeVolumeResponse, MakeVolumesRequest, MakeVolumesResponse, PingRequest, PingResponse, ReadAllRequest, ReadAllResponse,
+        ReadAtRequest, ReadAtResponse, ReadMultipleRequest, ReadMultipleResponse, ReadVersionRequest, ReadVersionResponse,
+        ReadXlRequest, ReadXlResponse, RenameDataRequest, RenameDataResponse, RenameFileRequst, RenameFileResponse,
+        RenamePartRequst, RenamePartResponse, StatVolumeRequest, StatVolumeResponse, UpdateMetadataRequest,
+        UpdateMetadataResponse, WalkDirRequest, WalkDirResponse, WriteAllRequest, WriteAllResponse, WriteMetadataRequest,
+        WriteMetadataResponse, WriteRequest, WriteResponse,
     },
 };
 
@@ -281,6 +283,36 @@ impl Node for NodeService {
         }
     }
 
+    async fn rename_part(&self, request: Request<RenamePartRequst>) -> Result<Response<RenamePartResponse>, Status> {
+        let request = request.into_inner();
+        if let Some(disk) = self.find_disk(&request.disk).await {
+            match disk
+                .rename_part(
+                    &request.src_volume,
+                    &request.src_path,
+                    &request.dst_volume,
+                    &request.dst_path,
+                    request.meta,
+                )
+                .await
+            {
+                Ok(_) => Ok(tonic::Response::new(RenamePartResponse {
+                    success: true,
+                    error_info: None,
+                })),
+                Err(err) => Ok(tonic::Response::new(RenamePartResponse {
+                    success: false,
+                    error_info: Some(err.to_string()),
+                })),
+            }
+        } else {
+            Ok(tonic::Response::new(RenamePartResponse {
+                success: false,
+                error_info: Some("can not find disk".to_string()),
+            }))
+        }
+    }
+
     async fn rename_file(&self, request: Request<RenameFileRequst>) -> Result<Response<RenameFileResponse>, Status> {
         let request = request.into_inner();
         if let Some(disk) = self.find_disk(&request.disk).await {
@@ -594,6 +626,68 @@ impl Node for NodeService {
         }
     }
 
+    async fn delete_paths(&self, request: Request<DeletePathsRequest>) -> Result<Response<DeletePathsResponse>, Status> {
+        let request = request.into_inner();
+        if let Some(disk) = self.find_disk(&request.disk).await {
+            let paths = request.paths.iter().map(|s| s.as_str()).collect::<Vec<&str>>();
+            match disk.delete_paths(&request.volume, &paths).await {
+                Ok(_) => Ok(tonic::Response::new(DeletePathsResponse {
+                    success: true,
+                    error_info: None,
+                })),
+                Err(err) => Ok(tonic::Response::new(DeletePathsResponse {
+                    success: false,
+                    error_info: Some(err.to_string()),
+                })),
+            }
+        } else {
+            Ok(tonic::Response::new(DeletePathsResponse {
+                success: false,
+                error_info: Some("can not find disk".to_string()),
+            }))
+        }
+    }
+
+    async fn update_metadata(&self, request: Request<UpdateMetadataRequest>) -> Result<Response<UpdateMetadataResponse>, Status> {
+        let request = request.into_inner();
+        if let Some(disk) = self.find_disk(&request.disk).await {
+            let file_info = match serde_json::from_str::<FileInfo>(&request.file_info) {
+                Ok(file_info) => file_info,
+                Err(_) => {
+                    return Ok(tonic::Response::new(UpdateMetadataResponse {
+                        success: false,
+                        error_info: Some("can not decode FileInfoVersions".to_string()),
+                    }));
+                }
+            };
+            let opts = match serde_json::from_str::<UpdateMetadataOpts>(&request.opts) {
+                Ok(opts) => opts,
+                Err(_) => {
+                    return Ok(tonic::Response::new(UpdateMetadataResponse {
+                        success: false,
+                        error_info: Some("can not decode UpdateMetadataOpts".to_string()),
+                    }));
+                }
+            };
+
+            match disk.update_metadata(&request.volume, &request.path, file_info, opts).await {
+                Ok(_) => Ok(tonic::Response::new(UpdateMetadataResponse {
+                    success: true,
+                    error_info: None,
+                })),
+                Err(err) => Ok(tonic::Response::new(UpdateMetadataResponse {
+                    success: false,
+                    error_info: Some(err.to_string()),
+                })),
+            }
+        } else {
+            Ok(tonic::Response::new(UpdateMetadataResponse {
+                success: false,
+                error_info: Some("can not find disk".to_string()),
+            }))
+        }
+    }
+
     async fn write_metadata(&self, request: Request<WriteMetadataRequest>) -> Result<Response<WriteMetadataResponse>, Status> {
         let request = request.into_inner();
         if let Some(disk) = self.find_disk(&request.disk).await {
@@ -699,6 +793,60 @@ impl Node for NodeService {
         }
     }
 
+    async fn delete_version(&self, request: Request<DeleteVersionRequest>) -> Result<Response<DeleteVersionResponse>, Status> {
+        let request = request.into_inner();
+        if let Some(disk) = self.find_disk(&request.disk).await {
+            let file_info = match serde_json::from_str::<FileInfo>(&request.file_info) {
+                Ok(file_info) => file_info,
+                Err(_) => {
+                    return Ok(tonic::Response::new(DeleteVersionResponse {
+                        success: false,
+                        raw_file_info: "".to_string(),
+                        error_info: Some("can not decode FileInfoVersions".to_string()),
+                    }));
+                }
+            };
+            let opts = match serde_json::from_str::<DeleteOptions>(&request.opts) {
+                Ok(opts) => opts,
+                Err(_) => {
+                    return Ok(tonic::Response::new(DeleteVersionResponse {
+                        success: false,
+                        raw_file_info: "".to_string(),
+                        error_info: Some("can not decode DeleteOptions".to_string()),
+                    }));
+                }
+            };
+            match disk
+                .delete_version(&request.volume, &request.path, file_info, request.force_del_marker, opts)
+                .await
+            {
+                Ok(raw_file_info) => match serde_json::to_string(&raw_file_info) {
+                    Ok(raw_file_info) => Ok(tonic::Response::new(DeleteVersionResponse {
+                        success: true,
+                        raw_file_info,
+                        error_info: None,
+                    })),
+                    Err(err) => Ok(tonic::Response::new(DeleteVersionResponse {
+                        success: false,
+                        raw_file_info: "".to_string(),
+                        error_info: Some(err.to_string()),
+                    })),
+                },
+                Err(err) => Ok(tonic::Response::new(DeleteVersionResponse {
+                    success: false,
+                    raw_file_info: "".to_string(),
+                    error_info: Some(err.to_string()),
+                })),
+            }
+        } else {
+            Ok(tonic::Response::new(DeleteVersionResponse {
+                success: false,
+                raw_file_info: "".to_string(),
+                error_info: Some("can not find disk".to_string()),
+            }))
+        }
+    }
+
     async fn delete_versions(&self, request: Request<DeleteVersionsRequest>) -> Result<Response<DeleteVersionsResponse>, Status> {
         let request = request.into_inner();
         if let Some(disk) = self.find_disk(&request.disk).await {