From 5417965e38b6b8d4b20fbf0dc3380be4ec15f858 Mon Sep 17 00:00:00 2001 From: weisd Date: Mon, 23 Sep 2024 17:56:30 +0800 Subject: [PATCH] review disk api --- ecstore/src/disk/local.rs | 20 +++++++- ecstore/src/disk/mod.rs | 98 ++++++++++++++++++++++++-------------- ecstore/src/disk/remote.rs | 20 +++++++- 3 files changed, 100 insertions(+), 38 deletions(-) diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 5ac8a9350..78ab0175f 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -1,7 +1,7 @@ use super::{endpoint::Endpoint, error::DiskError, format::FormatV3}; use super::{ DeleteOptions, DiskAPI, DiskLocation, FileInfoVersions, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, - ReadMultipleResp, ReadOptions, RenameDataResp, VolumeInfo, WalkDirOptions, + ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, }; use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE}; use crate::{ @@ -451,12 +451,23 @@ fn skip_access_checks(p: impl AsRef) -> bool { #[async_trait::async_trait] impl DiskAPI for LocalDisk { + fn to_string(&self) -> String { + self.root.to_string_lossy().to_string() + } fn is_local(&self) -> bool { true } + fn host_name(&self) -> String { + self.endpoint.host_port() + } async fn is_online(&self) -> bool { true } + + fn endpoint(&self) -> Endpoint { + self.endpoint.clone() + } + async fn close(&self) -> Result<()> { Ok(()) } @@ -959,7 +970,12 @@ impl DiskAPI for LocalDisk { created: modtime, }) } - + async fn delete_paths(&self, volume: &str, paths: &[&str]) -> Result<()> { + unimplemented!() + } + async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: UpdateMetadataOpts) { + unimplemented!() + } async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { let p = self.get_object_path(volume, format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str())?; diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index efee53ee1..fd46396c0 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -19,6 +19,7 @@ use crate::{ store_api::{FileInfo, RawFileInfo}, }; use bytes::Bytes; +use endpoint::Endpoint; use protos::proto_gen::node_service::{node_service_client::NodeServiceClient, ReadAtRequest, WriteRequest}; use serde::{Deserialize, Serialize}; use std::{fmt::Debug, io::SeekFrom, path::PathBuf, sync::Arc}; @@ -44,50 +45,34 @@ pub async fn new_disk(ep: &endpoint::Endpoint, opt: &DiskOption) -> Result bool; + fn to_string(&self) -> String; async fn is_online(&self) -> bool; - fn path(&self) -> PathBuf; + fn is_local(&self) -> bool; + // LastConn + fn host_name(&self) -> String; + fn endpoint(&self) -> Endpoint; async fn close(&self) -> Result<()>; async fn get_disk_id(&self) -> Result>; async fn set_disk_id(&self, id: Option) -> Result<()>; + + fn path(&self) -> PathBuf; fn get_location(&self) -> DiskLocation; - async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>; - async fn read_all(&self, volume: &str, path: &str) -> Result; - async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()>; - async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>; - async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result; - async fn append_file(&self, volume: &str, path: &str) -> Result; - async fn read_file(&self, volume: &str, path: &str) -> Result; - // 读目录下的所有文件、目录 - async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result>; + // Healing + // DiskInfo + // NSScanner + + // Volume operations. + async fn make_volume(&self, volume: &str) -> Result<()>; + async fn make_volumes(&self, volume: Vec<&str>) -> Result<()>; + async fn list_volumes(&self) -> Result>; + async fn stat_volume(&self, volume: &str) -> Result; + async fn delete_volume(&self, volume: &str) -> Result<()>; + // 并发边读边写 TODO: wr io.Writer async fn walk_dir(&self, opts: WalkDirOptions) -> Result>; - async fn rename_data( - &self, - src_volume: &str, - src_path: &str, - file_info: FileInfo, - dst_volume: &str, - dst_path: &str, - ) -> Result; - async fn make_volumes(&self, volume: Vec<&str>) -> Result<()>; - async fn delete_volume(&self, volume: &str) -> Result<()>; - async fn list_volumes(&self) -> Result>; - async fn make_volume(&self, volume: &str) -> Result<()>; - async fn stat_volume(&self, volume: &str) -> Result; - - async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()>; - async fn read_version( - &self, - org_volume: &str, - volume: &str, - path: &str, - version_id: &str, - opts: &ReadOptions, - ) -> Result; - async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result; + // Metadata operations async fn delete_version( &self, volume: &str, @@ -102,7 +87,50 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { versions: Vec, opts: DeleteOptions, ) -> Result>>; + async fn delete_paths(&self, volume: &str, paths: &[&str]) -> Result<()>; + async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()>; + async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: UpdateMetadataOpts); + async fn read_version( + &self, + org_volume: &str, + volume: &str, + path: &str, + version_id: &str, + opts: &ReadOptions, + ) -> Result; + async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result; + async fn rename_data( + &self, + src_volume: &str, + src_path: &str, + file_info: FileInfo, + dst_volume: &str, + dst_path: &str, + ) -> Result; + + // File operations. + // 读目录下的所有文件、目录 + async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result>; + async fn read_file(&self, volume: &str, path: &str) -> Result; + async fn append_file(&self, volume: &str, path: &str) -> Result; + async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result; + // ReadFileStream + async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>; + // RenamePart + // CheckParts + async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>; + // VerifyFile + // StatInfoFile + // ReadParts async fn read_multiple(&self, req: ReadMultipleReq) -> Result>; + // CleanAbandonedData + async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()>; + async fn read_all(&self, volume: &str, path: &str) -> Result; + // GetDiskLoc +} + +pub struct UpdateMetadataOpts { + pub no_persistence: bool, } pub struct DiskLocation { diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index 17a359445..5eed152c2 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -30,7 +30,7 @@ use crate::{ use super::{ endpoint::Endpoint, DeleteOptions, DiskAPI, DiskLocation, DiskOption, FileInfoVersions, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter, RenameDataResp, - VolumeInfo, WalkDirOptions, + UpdateMetadataOpts, VolumeInfo, WalkDirOptions, }; #[derive(Debug)] @@ -97,9 +97,17 @@ impl RemoteDisk { // TODO: all api need to handle errors #[async_trait::async_trait] impl DiskAPI for RemoteDisk { + fn to_string(&self) -> String { + self.endpoint.to_string() + } + fn is_local(&self) -> bool { false } + + fn host_name(&self) -> String { + self.endpoint.host_port() + } async fn is_online(&self) -> bool { // TODO: 连接状态 if let Ok(_) = self.get_client_v2().await { @@ -107,6 +115,9 @@ impl DiskAPI for RemoteDisk { } false } + fn endpoint(&self) -> Endpoint { + self.endpoint.clone() + } async fn close(&self) -> Result<()> { Ok(()) } @@ -390,6 +401,13 @@ impl DiskAPI for RemoteDisk { Ok(volume_info) } + async fn delete_paths(&self, volume: &str, paths: &[&str]) -> Result<()> { + unimplemented!() + } + async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: UpdateMetadataOpts) { + unimplemented!() + } + async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { info!("write_metadata"); let file_info = serde_json::to_string(&fi)?;