From 9fc4bb919ea7f3b40453d44a27aa10049b61a279 Mon Sep 17 00:00:00 2001 From: weisd Date: Fri, 25 Apr 2025 09:54:41 +0800 Subject: [PATCH] fix:#355 multi pools select error --- ecstore/src/disk/error.rs | 4 +++ ecstore/src/disk/local.rs | 39 ++++++++++++++++++++- ecstore/src/disk/mod.rs | 49 +++++++++++++++++++++++++- ecstore/src/disk/remote.rs | 39 +++++++++++++++++++++ ecstore/src/erasure.rs | 7 ++-- ecstore/src/set_disk.rs | 70 +++++++++++++++++++++++++++++++++----- ecstore/src/sets.rs | 46 +++++++++++++++++++++++++ ecstore/src/store.rs | 37 ++++++++++++++++++++ 8 files changed, 275 insertions(+), 16 deletions(-) diff --git a/ecstore/src/disk/error.rs b/ecstore/src/disk/error.rs index d7ea2854a..0d422b106 100644 --- a/ecstore/src/disk/error.rs +++ b/ecstore/src/disk/error.rs @@ -364,6 +364,10 @@ pub fn is_unformatted_disk(err: &Error) -> bool { } pub fn is_err_file_not_found(err: &Error) -> bool { + if let Some(ioerr) = err.downcast_ref::() { + return ioerr.kind() == ErrorKind::NotFound; + } + matches!(err.downcast_ref::(), Some(DiskError::FileNotFound)) } diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 4ab9bb183..74cdaa690 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -1079,30 +1079,39 @@ fn skip_access_checks(p: impl AsRef) -> bool { #[async_trait::async_trait] impl DiskAPI for LocalDisk { + #[tracing::instrument(skip(self))] fn to_string(&self) -> String { self.root.to_string_lossy().to_string() } + #[tracing::instrument(skip(self))] fn is_local(&self) -> bool { true } + #[tracing::instrument(skip(self))] fn host_name(&self) -> String { self.endpoint.host_port() } + #[tracing::instrument(skip(self))] async fn is_online(&self) -> bool { self.check_format_json().await.is_ok() } + #[tracing::instrument(skip(self))] fn endpoint(&self) -> Endpoint { self.endpoint.clone() } + #[tracing::instrument(skip(self))] async fn close(&self) -> Result<()> { Ok(()) } + + #[tracing::instrument(skip(self))] fn path(&self) -> PathBuf { self.root.clone() } + #[tracing::instrument(skip(self))] fn get_disk_location(&self) -> DiskLocation { DiskLocation { pool_idx: { @@ -1179,6 +1188,7 @@ impl DiskAPI for LocalDisk { Ok(Some(disk_id)) } + #[tracing::instrument(skip(self))] async fn set_disk_id(&self, id: Option) -> Result<()> { // 本地不需要设置 // TODO: add check_id_store @@ -1188,6 +1198,7 @@ impl DiskAPI for LocalDisk { } #[must_use] + #[tracing::instrument(skip(self))] async fn read_all(&self, volume: &str, path: &str) -> Result> { if volume == super::RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE { let format_info = self.format_info.read().await; @@ -1202,10 +1213,12 @@ impl DiskAPI for LocalDisk { Ok(data) } + #[tracing::instrument(skip(self))] async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { self.write_all_public(volume, path, data).await } + #[tracing::instrument(skip(self))] async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { @@ -1223,6 +1236,7 @@ impl DiskAPI for LocalDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { @@ -1268,6 +1282,7 @@ impl DiskAPI for LocalDisk { Ok(resp) } + #[tracing::instrument(skip(self))] async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { let volume_dir = self.get_bucket_path(volume)?; check_path_length(volume_dir.join(path).to_string_lossy().as_ref())?; @@ -1405,6 +1420,8 @@ impl DiskAPI for LocalDisk { Ok(()) } + + #[tracing::instrument(skip(self))] async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> { let src_volume_dir = self.get_bucket_path(src_volume)?; let dst_volume_dir = self.get_bucket_path(dst_volume)?; @@ -1704,7 +1721,7 @@ impl DiskAPI for LocalDisk { Ok(()) } - // #[tracing::instrument(skip(self))] + #[tracing::instrument(skip(self))] async fn rename_data( &self, src_volume: &str, @@ -1910,6 +1927,7 @@ impl DiskAPI for LocalDisk { }) } + #[tracing::instrument(skip(self))] async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> { for vol in volumes { if let Err(e) = self.make_volume(vol).await { @@ -1921,6 +1939,8 @@ impl DiskAPI for LocalDisk { } Ok(()) } + + #[tracing::instrument(skip(self))] async fn make_volume(&self, volume: &str) -> Result<()> { if !Self::is_valid_volname(volume) { return Err(Error::msg("Invalid arguments specified")); @@ -1944,6 +1964,8 @@ impl DiskAPI for LocalDisk { Err(Error::from(DiskError::VolumeExists)) } + + #[tracing::instrument(skip(self))] async fn list_volumes(&self) -> Result> { let mut volumes = Vec::new(); @@ -1968,6 +1990,8 @@ impl DiskAPI for LocalDisk { Ok(volumes) } + + #[tracing::instrument(skip(self))] async fn stat_volume(&self, volume: &str) -> Result { let volume_dir = self.get_bucket_path(volume)?; let meta = match utils::fs::lstat(&volume_dir).await { @@ -1995,6 +2019,8 @@ impl DiskAPI for LocalDisk { created: modtime, }) } + + #[tracing::instrument(skip(self))] async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()> { let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { @@ -2013,6 +2039,8 @@ impl DiskAPI for LocalDisk { Ok(()) } + + #[tracing::instrument(skip(self))] async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> { if fi.metadata.is_some() { let volume_dir = self.get_bucket_path(volume)?; @@ -2053,6 +2081,8 @@ impl DiskAPI for LocalDisk { Err(Error::msg("Invalid Argument")) } + + #[tracing::instrument(skip(self))] 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())?; @@ -2106,6 +2136,8 @@ impl DiskAPI for LocalDisk { Ok(RawFileInfo { buf }) } + + #[tracing::instrument(skip(self))] async fn delete_version( &self, volume: &str, @@ -2214,6 +2246,8 @@ impl DiskAPI for LocalDisk { Ok(errs) } + + #[tracing::instrument(skip(self))] async fn read_multiple(&self, req: ReadMultipleReq) -> Result> { let mut results = Vec::new(); let mut found = 0; @@ -2273,6 +2307,7 @@ impl DiskAPI for LocalDisk { Ok(results) } + #[tracing::instrument(skip(self))] async fn delete_volume(&self, volume: &str) -> Result<()> { let p = self.get_bucket_path(volume)?; @@ -2295,6 +2330,7 @@ impl DiskAPI for LocalDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn disk_info(&self, _: &DiskInfoOptions) -> Result { let mut info = Cache::get(self.disk_info_cache.clone()).await?; // TODO: nr_requests, rotational @@ -2454,6 +2490,7 @@ impl DiskAPI for LocalDisk { Ok(data_usage_info) } + #[tracing::instrument(skip(self))] async fn healing(&self) -> Option { let healing_file = path_join(&[ self.path(), diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index e26eebb8b..ddefd1bed 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -49,6 +49,7 @@ pub enum Disk { #[async_trait::async_trait] impl DiskAPI for Disk { + #[tracing::instrument(skip(self))] fn to_string(&self) -> String { match self { Disk::Local(local_disk) => local_disk.to_string(), @@ -56,6 +57,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] fn is_local(&self) -> bool { match self { Disk::Local(local_disk) => local_disk.is_local(), @@ -63,30 +65,39 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] fn host_name(&self) -> String { match self { Disk::Local(local_disk) => local_disk.host_name(), Disk::Remote(remote_disk) => remote_disk.host_name(), } } + + #[tracing::instrument(skip(self))] async fn is_online(&self) -> bool { match self { Disk::Local(local_disk) => local_disk.is_online().await, Disk::Remote(remote_disk) => remote_disk.is_online().await, } } + + #[tracing::instrument(skip(self))] fn endpoint(&self) -> Endpoint { match self { Disk::Local(local_disk) => local_disk.endpoint(), Disk::Remote(remote_disk) => remote_disk.endpoint(), } } + + #[tracing::instrument(skip(self))] async fn close(&self) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.close().await, Disk::Remote(remote_disk) => remote_disk.close().await, } } + + #[tracing::instrument(skip(self))] fn path(&self) -> PathBuf { match self { Disk::Local(local_disk) => local_disk.path(), @@ -94,6 +105,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] fn get_disk_location(&self) -> DiskLocation { match self { Disk::Local(local_disk) => local_disk.get_disk_location(), @@ -101,12 +113,15 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn get_disk_id(&self) -> Result> { match self { Disk::Local(local_disk) => local_disk.get_disk_id().await, Disk::Remote(remote_disk) => remote_disk.get_disk_id().await, } } + + #[tracing::instrument(skip(self))] async fn set_disk_id(&self, id: Option) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.set_disk_id(id).await, @@ -114,6 +129,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn read_all(&self, volume: &str, path: &str) -> Result> { match self { Disk::Local(local_disk) => local_disk.read_all(volume, path).await, @@ -121,6 +137,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.write_all(volume, path, data).await, @@ -128,6 +145,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.delete(volume, path, opt).await, @@ -135,6 +153,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { match self { Disk::Local(local_disk) => local_disk.verify_file(volume, path, fi).await, @@ -142,6 +161,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { match self { Disk::Local(local_disk) => local_disk.check_parts(volume, path, fi).await, @@ -149,6 +169,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.rename_part(src_volume, src_path, dst_volume, dst_path, meta).await, @@ -159,6 +180,8 @@ impl DiskAPI for Disk { } } } + + #[tracing::instrument(skip(self))] async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.rename_file(src_volume, src_path, dst_volume, dst_path).await, @@ -166,6 +189,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result { match self { Disk::Local(local_disk) => local_disk.create_file(_origvolume, volume, path, _file_size).await, @@ -173,6 +197,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn append_file(&self, volume: &str, path: &str) -> Result { match self { Disk::Local(local_disk) => local_disk.append_file(volume, path).await, @@ -180,6 +205,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn read_file(&self, volume: &str, path: &str) -> Result { match self { Disk::Local(local_disk) => local_disk.read_file(volume, path).await, @@ -187,6 +213,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { match self { Disk::Local(local_disk) => local_disk.read_file_stream(volume, path, offset, length).await, @@ -194,6 +221,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result> { match self { Disk::Local(local_disk) => local_disk.list_dir(_origvolume, volume, _dir_path, _count).await, @@ -201,6 +229,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self, wr))] async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.walk_dir(opts, wr).await, @@ -208,6 +237,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn rename_data( &self, src_volume: &str, @@ -222,6 +252,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.make_volumes(volumes).await, @@ -229,6 +260,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn make_volume(&self, volume: &str) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.make_volume(volume).await, @@ -236,6 +268,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn list_volumes(&self) -> Result> { match self { Disk::Local(local_disk) => local_disk.list_volumes().await, @@ -243,6 +276,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn stat_volume(&self, volume: &str) -> Result { match self { Disk::Local(local_disk) => local_disk.stat_volume(volume).await, @@ -250,12 +284,15 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.delete_paths(volume, paths).await, Disk::Remote(remote_disk) => remote_disk.delete_paths(volume, paths).await, } } + + #[tracing::instrument(skip(self))] async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.update_metadata(volume, path, fi, opts).await, @@ -263,6 +300,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.write_metadata(_org_volume, volume, path, fi).await, @@ -285,13 +323,15 @@ impl DiskAPI for Disk { } } - #[tracing::instrument] + #[tracing::instrument(skip(self))] async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result { match self { Disk::Local(local_disk) => local_disk.read_xl(volume, path, read_data).await, Disk::Remote(remote_disk) => remote_disk.read_xl(volume, path, read_data).await, } } + + #[tracing::instrument(skip(self))] async fn delete_version( &self, volume: &str, @@ -305,6 +345,8 @@ impl DiskAPI for Disk { Disk::Remote(remote_disk) => remote_disk.delete_version(volume, path, fi, force_del_marker, opts).await, } } + + #[tracing::instrument(skip(self))] async fn delete_versions( &self, volume: &str, @@ -317,6 +359,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn read_multiple(&self, req: ReadMultipleReq) -> Result> { match self { Disk::Local(local_disk) => local_disk.read_multiple(req).await, @@ -324,6 +367,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn delete_volume(&self, volume: &str) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.delete_volume(volume).await, @@ -331,6 +375,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn disk_info(&self, opts: &DiskInfoOptions) -> Result { match self { Disk::Local(local_disk) => local_disk.disk_info(opts).await, @@ -338,6 +383,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self, cache, we_sleep, scan_mode))] async fn ns_scanner( &self, cache: &DataUsageCache, @@ -351,6 +397,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn healing(&self) -> Option { match self { Disk::Local(local_disk) => local_disk.healing().await, diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index c635d744a..2b4e64dc2 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -70,17 +70,21 @@ impl RemoteDisk { // TODO: all api need to handle errors #[async_trait::async_trait] impl DiskAPI for RemoteDisk { + #[tracing::instrument(skip(self))] fn to_string(&self) -> String { self.endpoint.to_string() } + #[tracing::instrument(skip(self))] fn is_local(&self) -> bool { false } + #[tracing::instrument(skip(self))] fn host_name(&self) -> String { self.endpoint.host_port() } + #[tracing::instrument(skip(self))] async fn is_online(&self) -> bool { // TODO: 连接状态 if (node_service_time_out_client(&self.addr).await).is_ok() { @@ -88,16 +92,20 @@ impl DiskAPI for RemoteDisk { } false } + #[tracing::instrument(skip(self))] fn endpoint(&self) -> Endpoint { self.endpoint.clone() } + #[tracing::instrument(skip(self))] async fn close(&self) -> Result<()> { Ok(()) } + #[tracing::instrument(skip(self))] fn path(&self) -> PathBuf { self.root.clone() } + #[tracing::instrument(skip(self))] fn get_disk_location(&self) -> DiskLocation { DiskLocation { pool_idx: { @@ -124,9 +132,12 @@ impl DiskAPI for RemoteDisk { } } + #[tracing::instrument(skip(self))] async fn get_disk_id(&self) -> Result> { Ok(*self.id.lock().await) } + + #[tracing::instrument(skip(self))] async fn set_disk_id(&self, id: Option) -> Result<()> { let mut lock = self.id.lock().await; *lock = id; @@ -134,6 +145,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn read_all(&self, volume: &str, path: &str) -> Result> { info!("read_all {}/{}", volume, path); let mut client = node_service_time_out_client(&self.addr) @@ -154,6 +166,7 @@ impl DiskAPI for RemoteDisk { Ok(response.data) } + #[tracing::instrument(skip(self))] async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { info!("write_all"); let mut client = node_service_time_out_client(&self.addr) @@ -179,6 +192,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { info!("delete {}/{}/{}", self.endpoint.to_string(), volume, path); let options = serde_json::to_string(&opt)?; @@ -205,6 +219,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { info!("verify_file"); let file_info = serde_json::to_string(&fi)?; @@ -233,6 +248,7 @@ impl DiskAPI for RemoteDisk { Ok(check_parts_resp) } + #[tracing::instrument(skip(self))] async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { info!("check_parts"); let file_info = serde_json::to_string(&fi)?; @@ -261,6 +277,7 @@ impl DiskAPI for RemoteDisk { Ok(check_parts_resp) } + #[tracing::instrument(skip(self))] async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec) -> Result<()> { info!("rename_part {}/{}", src_volume, src_path); let mut client = node_service_time_out_client(&self.addr) @@ -287,6 +304,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(level = "debug", skip(self))] async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> { info!("rename_file"); @@ -365,6 +383,7 @@ impl DiskAPI for RemoteDisk { )) } + #[tracing::instrument(skip(self))] async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result> { info!("list_dir {}/{}", volume, _dir_path); let mut client = node_service_time_out_client(&self.addr) @@ -389,6 +408,7 @@ impl DiskAPI for RemoteDisk { } // FIXME: TODO: use writer + #[tracing::instrument(skip(self, wr))] async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> { let now = std::time::SystemTime::now(); info!("walk_dir {}/{}/{:?}", self.endpoint.to_string(), opts.bucket, opts.filter_prefix); @@ -429,6 +449,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn rename_data( &self, src_volume: &str, @@ -466,6 +487,7 @@ impl DiskAPI for RemoteDisk { Ok(rename_data_resp) } + #[tracing::instrument(skip(self))] async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> { info!("make_volumes"); let mut client = node_service_time_out_client(&self.addr) @@ -489,6 +511,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn make_volume(&self, volume: &str) -> Result<()> { info!("make_volume"); let mut client = node_service_time_out_client(&self.addr) @@ -512,6 +535,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn list_volumes(&self) -> Result> { info!("list_volumes"); let mut client = node_service_time_out_client(&self.addr) @@ -540,6 +564,7 @@ impl DiskAPI for RemoteDisk { Ok(infos) } + #[tracing::instrument(skip(self))] async fn stat_volume(&self, volume: &str) -> Result { info!("stat_volume"); let mut client = node_service_time_out_client(&self.addr) @@ -565,6 +590,7 @@ impl DiskAPI for RemoteDisk { Ok(volume_info) } + #[tracing::instrument(skip(self))] async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()> { info!("delete_paths"); let paths = paths.to_owned(); @@ -589,6 +615,8 @@ impl DiskAPI for RemoteDisk { Ok(()) } + + #[tracing::instrument(skip(self))] 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)?; @@ -618,6 +646,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { info!("write_metadata {}/{}", volume, path); let file_info = serde_json::to_string(&fi)?; @@ -644,6 +673,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn read_version( &self, _org_volume: &str, @@ -707,6 +737,8 @@ impl DiskAPI for RemoteDisk { Ok(raw_file_info) } + + #[tracing::instrument(skip(self))] async fn delete_version( &self, volume: &str, @@ -745,6 +777,8 @@ impl DiskAPI for RemoteDisk { Ok(()) } + + #[tracing::instrument(skip(self))] async fn delete_versions( &self, volume: &str, @@ -790,6 +824,7 @@ impl DiskAPI for RemoteDisk { Ok(errors) } + #[tracing::instrument(skip(self))] async fn read_multiple(&self, req: ReadMultipleReq) -> Result> { info!("read_multiple {}/{}/{}", self.endpoint.to_string(), req.bucket, req.prefix); let read_multiple_req = serde_json::to_string(&req)?; @@ -820,6 +855,7 @@ impl DiskAPI for RemoteDisk { Ok(read_multiple_resps) } + #[tracing::instrument(skip(self))] async fn delete_volume(&self, volume: &str) -> Result<()> { info!("delete_volume {}/{}", self.endpoint.to_string(), volume); let mut client = node_service_time_out_client(&self.addr) @@ -843,6 +879,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn disk_info(&self, opts: &DiskInfoOptions) -> Result { let opts = serde_json::to_string(&opts)?; let mut client = node_service_time_out_client(&self.addr) @@ -868,6 +905,7 @@ impl DiskAPI for RemoteDisk { Ok(disk_info) } + #[tracing::instrument(skip(self, cache, scan_mode, _we_sleep))] async fn ns_scanner( &self, cache: &DataUsageCache, @@ -909,6 +947,7 @@ impl DiskAPI for RemoteDisk { } } + #[tracing::instrument(skip(self))] async fn healing(&self) -> Option { None } diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index b4c45ff11..021ae7bed 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -102,14 +102,11 @@ impl Erasure { self.encode_data(&self.buf, &mut blocks)?; let write_futures = writers.iter_mut().enumerate().map(|(i, w_op)| { - let i_inner = i.clone(); + let i_inner = i; let blocks_inner = blocks.clone(); async move { if let Some(w) = w_op { - match w.write(blocks_inner[i_inner].clone()).await { - Ok(_) => None, - Err(e) => Some(e), - } + (w.write(blocks_inner[i_inner].clone()).await).err() } else { Some(Error::new(DiskError::DiskNotFound)) } diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index 35f41b6a1..19c244190 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -33,7 +33,7 @@ use crate::{ ListMultipartsInfo, ListObjectsV2Info, MakeBucketOptions, MultipartInfo, MultipartUploadResult, ObjectIO, ObjectInfo, ObjectOptions, ObjectPartInfo, ObjectToDelete, PartInfo, PutObjReader, RawFileInfo, StorageAPI, DEFAULT_BITROT_ALGO, }, - store_err::{to_object_err, StorageError}, + store_err::{is_err_object_not_found, to_object_err, StorageError}, store_init::{load_format_erasure, ErasureError}, utils::{ self, @@ -345,7 +345,7 @@ impl SetDisks { false, DeleteOptions { undo_write: true, - old_data_dir: old_data_dir, + old_data_dir, ..Default::default() }, ) @@ -459,6 +459,7 @@ impl SetDisks { Ok(()) } + #[tracing::instrument(skip(disks))] async fn cleanup_multipart_path(disks: &[Option], paths: &[String]) { let mut futures = Vec::with_capacity(disks.len()); @@ -490,6 +491,8 @@ impl SetDisks { warn!("cleanup_multipart_path errs {:?}", &errs); } } + + #[tracing::instrument(skip(disks, meta))] async fn rename_part( disks: &[Option], src_bucket: &str, @@ -580,6 +583,7 @@ impl SetDisks { // errors // } + #[tracing::instrument(skip(disks, files))] async fn write_unique_file_info( disks: &[Option], org_bucket: &str, @@ -931,7 +935,15 @@ impl SetDisks { let (parts_metadata, errs) = Self::read_all_fileinfo(&disks, bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_id_path, "", false, false).await; - let (read_quorum, write_quorum) = Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)?; + let map_err_notfound = |err: Error| { + if is_err_object_not_found(&err) { + return Error::new(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned())); + } + err + }; + + let (read_quorum, write_quorum) = + Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count).map_err(map_err_notfound)?; if read_quorum < 0 { return Err(Error::new(QuorumError::Read)); @@ -946,10 +958,10 @@ impl SetDisks { quorum = write_quorum as usize; if let Some(err) = reduce_write_quorum_errs(&errs, object_op_ignored_errs().as_ref(), quorum) { - return Err(err); + return Err(map_err_notfound(err)); } } else if let Some(err) = reduce_read_quorum_errs(&errs, object_op_ignored_errs().as_ref(), quorum) { - return Err(err); + return Err(map_err_notfound(err)); } let (_, mod_time, etag) = Self::list_online_disks(&disks, &parts_metadata, &errs, quorum); @@ -3875,14 +3887,17 @@ impl ObjectIO for SetDisks { #[async_trait::async_trait] impl StorageAPI for SetDisks { + #[tracing::instrument(skip(self))] async fn backend_info(&self) -> madmin::BackendInfo { unimplemented!() } + #[tracing::instrument(skip(self))] async fn storage_info(&self) -> madmin::StorageInfo { let disks = self.get_disks_internal().await; get_storage_info(&disks, &self.set_endpoints).await } + #[tracing::instrument(skip(self))] async fn local_storage_info(&self) -> madmin::StorageInfo { let disks = self.get_disks_internal().await; @@ -3898,17 +3913,21 @@ impl StorageAPI for SetDisks { get_storage_info(&local_disks, &local_endpoints).await } + #[tracing::instrument(skip(self))] async fn list_bucket(&self, _opts: &BucketOptions) -> Result> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn get_bucket_info(&self, _bucket: &str, _opts: &BucketOptions) -> Result { unimplemented!() } + #[tracing::instrument(skip(self))] async fn copy_object( &self, src_bucket: &str, @@ -4013,6 +4032,7 @@ impl StorageAPI for SetDisks { Ok(fi.to_object_info(src_bucket, src_object, src_opts.versioned || src_opts.version_suspended)) } + #[tracing::instrument(skip(self))] async fn delete_objects( &self, bucket: &str, @@ -4123,6 +4143,8 @@ impl StorageAPI for SetDisks { Ok((del_objects, del_errs)) } + + #[tracing::instrument(skip(self))] async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result { if opts.delete_prefix { self.delete_prefix(bucket, object) @@ -4134,6 +4156,7 @@ impl StorageAPI for SetDisks { unimplemented!() } + #[tracing::instrument(skip(self))] async fn list_objects_v2( self: Arc, _bucket: &str, @@ -4146,6 +4169,8 @@ impl StorageAPI for SetDisks { ) -> Result { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn list_object_versions( self: Arc, _bucket: &str, @@ -4157,6 +4182,8 @@ impl StorageAPI for SetDisks { ) -> Result { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { // let mut _ns = None; // if !opts.no_lock { @@ -4198,6 +4225,7 @@ impl StorageAPI for SetDisks { Ok(oi) } + #[tracing::instrument(skip(self))] async fn put_object_metadata(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { // TODO: nslock @@ -4280,6 +4308,7 @@ impl StorageAPI for SetDisks { Ok(fi.to_object_info(bucket, object, opts.versioned || opts.version_suspended)) } + #[tracing::instrument(skip(self))] async fn get_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { let oi = self.get_object_info(bucket, object, opts).await?; Ok(oi.user_tags) @@ -4306,10 +4335,13 @@ impl StorageAPI for SetDisks { // TODO: versioned Ok(fi.to_object_info(bucket, object, opts.versioned || opts.version_suspended)) } + + #[tracing::instrument(skip(self))] async fn delete_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.put_object_tags(bucket, object, "", opts).await } + #[tracing::instrument(skip(self))] async fn copy_object_part( &self, _src_bucket: &str, @@ -4432,6 +4464,8 @@ impl StorageAPI for SetDisks { Ok(ret) } + + #[tracing::instrument(skip(self))] async fn list_multipart_uploads( &self, bucket: &str, @@ -4471,7 +4505,7 @@ impl StorageAPI for SetDisks { Err(err) => { if DiskError::DiskNotFound.is(&err) { None - } else if DiskError::FileNotFound.is(&err) { + } else if is_err_object_not_found(&err) { return Ok(ListMultipartsInfo { key_marker: key_marker.to_owned(), max_uploads, @@ -4575,9 +4609,9 @@ impl StorageAPI for SetDisks { ..Default::default() }) } - async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { - warn!("new_multipart_upload opt {:?}", opts); + #[tracing::instrument(skip(self))] + async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { let disks = self.disks.read().await; let disks = disks.clone(); @@ -4680,6 +4714,8 @@ impl StorageAPI for SetDisks { Ok(MultipartUploadResult { upload_id }) } + + #[tracing::instrument(skip(self))] async fn get_multipart_info( &self, bucket: &str, @@ -4701,6 +4737,8 @@ impl StorageAPI for SetDisks { ..Default::default() }) } + + #[tracing::instrument(skip(self))] async fn abort_multipart_upload(&self, bucket: &str, object: &str, upload_id: &str, _opts: &ObjectOptions) -> Result<()> { self.check_upload_id_exists(bucket, object, upload_id, false).await?; let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id); @@ -4708,7 +4746,7 @@ impl StorageAPI for SetDisks { self.delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path).await } // complete_multipart_upload 完成 - // #[tracing::instrument(skip(self))] + #[tracing::instrument(skip(self))] async fn complete_multipart_upload( &self, bucket: &str, @@ -4957,24 +4995,32 @@ impl StorageAPI for SetDisks { Ok(fi.to_object_info(bucket, object, opts.versioned || opts.version_suspended)) } + #[tracing::instrument(skip(self))] async fn get_disks(&self, _pool_idx: usize, _set_idx: usize) -> Result>> { Ok(self.get_disks_internal().await) } + #[tracing::instrument(skip(self))] fn set_drive_counts(&self) -> Vec { unimplemented!() } + #[tracing::instrument(skip(self))] async fn delete_bucket(&self, _bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option)> { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn heal_bucket(&self, _bucket: &str, _opts: &HealOpts) -> Result { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn heal_object( &self, bucket: &str, @@ -5025,6 +5071,8 @@ impl StorageAPI for SetDisks { } return Ok((result, err)); } + + #[tracing::instrument(skip(self))] async fn heal_objects( &self, _bucket: &str, @@ -5035,9 +5083,13 @@ impl StorageAPI for SetDisks { ) -> Result<()> { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn get_pool_and_set(&self, _id: &str) -> Result<(Option, Option, Option)> { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn check_abandoned_parts(&self, _bucket: &str, _object: &str, _opts: &HealOpts) -> Result<()> { unimplemented!() } diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index 0a2e716fa..7a4cf93a3 100644 --- a/ecstore/src/sets.rs +++ b/ecstore/src/sets.rs @@ -308,9 +308,11 @@ impl ObjectIO for Sets { #[async_trait::async_trait] impl StorageAPI for Sets { + #[tracing::instrument(skip(self))] async fn backend_info(&self) -> madmin::BackendInfo { unimplemented!() } + #[tracing::instrument(skip(self))] async fn storage_info(&self) -> madmin::StorageInfo { let mut futures = Vec::with_capacity(self.disk_set.len()); @@ -331,6 +333,7 @@ impl StorageAPI for Sets { ..Default::default() } } + #[tracing::instrument(skip(self))] async fn local_storage_info(&self) -> madmin::StorageInfo { let mut futures = Vec::with_capacity(self.disk_set.len()); @@ -350,16 +353,21 @@ impl StorageAPI for Sets { ..Default::default() } } + #[tracing::instrument(skip(self))] async fn list_bucket(&self, _opts: &BucketOptions) -> Result> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn get_bucket_info(&self, _bucket: &str, _opts: &BucketOptions) -> Result { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn copy_object( &self, src_bucket: &str, @@ -416,6 +424,8 @@ impl StorageAPI for Sets { "put_object_reader2 is none".to_owned(), ))) } + + #[tracing::instrument(skip(self))] async fn delete_objects( &self, bucket: &str, @@ -498,6 +508,8 @@ impl StorageAPI for Sets { Ok((del_objects, del_errs)) } + + #[tracing::instrument(skip(self))] async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result { if opts.delete_prefix && !opts.delete_prefix_object { self.delete_prefix(bucket, object).await?; @@ -506,6 +518,8 @@ impl StorageAPI for Sets { self.get_disks_by_key(object).delete_object(bucket, object, opts).await } + + #[tracing::instrument(skip(self))] async fn list_objects_v2( self: Arc, _bucket: &str, @@ -518,6 +532,8 @@ impl StorageAPI for Sets { ) -> Result { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn list_object_versions( self: Arc, _bucket: &str, @@ -529,14 +545,18 @@ impl StorageAPI for Sets { ) -> Result { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.get_disks_by_key(object).get_object_info(bucket, object, opts).await } + #[tracing::instrument(skip(self))] async fn put_object_metadata(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.get_disks_by_key(object).put_object_metadata(bucket, object, opts).await } + #[tracing::instrument(skip(self))] async fn get_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.get_disks_by_key(object).get_object_tags(bucket, object, opts).await } @@ -546,10 +566,13 @@ impl StorageAPI for Sets { .put_object_tags(bucket, object, tags, opts) .await } + + #[tracing::instrument(skip(self))] async fn delete_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.get_disks_by_key(object).delete_object_tags(bucket, object, opts).await } + #[tracing::instrument(skip(self))] async fn copy_object_part( &self, _src_bucket: &str, @@ -566,6 +589,8 @@ impl StorageAPI for Sets { ) -> Result<()> { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn put_object_part( &self, bucket: &str, @@ -579,6 +604,8 @@ impl StorageAPI for Sets { .put_object_part(bucket, object, upload_id, part_id, data, opts) .await } + + #[tracing::instrument(skip(self))] async fn list_multipart_uploads( &self, bucket: &str, @@ -592,9 +619,13 @@ impl StorageAPI for Sets { .list_multipart_uploads(bucket, prefix, key_marker, upload_id_marker, delimiter, max_uploads) .await } + + #[tracing::instrument(skip(self))] async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.get_disks_by_key(object).new_multipart_upload(bucket, object, opts).await } + + #[tracing::instrument(skip(self))] async fn get_multipart_info( &self, bucket: &str, @@ -606,11 +637,15 @@ impl StorageAPI for Sets { .get_multipart_info(bucket, object, upload_id, opts) .await } + + #[tracing::instrument(skip(self))] async fn abort_multipart_upload(&self, bucket: &str, object: &str, upload_id: &str, opts: &ObjectOptions) -> Result<()> { self.get_disks_by_key(object) .abort_multipart_upload(bucket, object, upload_id, opts) .await } + + #[tracing::instrument(skip(self))] async fn complete_multipart_upload( &self, bucket: &str, @@ -623,17 +658,23 @@ impl StorageAPI for Sets { .complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts) .await } + + #[tracing::instrument(skip(self))] async fn get_disks(&self, _pool_idx: usize, _set_idx: usize) -> Result>> { unimplemented!() } + #[tracing::instrument(skip(self))] fn set_drive_counts(&self) -> Vec { unimplemented!() } + #[tracing::instrument(skip(self))] async fn delete_bucket(&self, _bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option)> { let (disks, _) = init_storage_disks_with_errors( &self.endpoints.endpoints, @@ -718,9 +759,11 @@ impl StorageAPI for Sets { } Ok((res, None)) } + #[tracing::instrument(skip(self))] async fn heal_bucket(&self, _bucket: &str, _opts: &HealOpts) -> Result { unimplemented!() } + #[tracing::instrument(skip(self))] async fn heal_object( &self, bucket: &str, @@ -732,6 +775,7 @@ impl StorageAPI for Sets { .heal_object(bucket, object, version_id, opts) .await } + #[tracing::instrument(skip(self))] async fn heal_objects( &self, _bucket: &str, @@ -742,9 +786,11 @@ impl StorageAPI for Sets { ) -> Result<()> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn get_pool_and_set(&self, _id: &str) -> Result<(Option, Option, Option)> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn check_abandoned_parts(&self, _bucket: &str, _object: &str, _opts: &HealOpts) -> Result<()> { unimplemented!() } diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 23a17596b..5a2075775 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -1246,6 +1246,7 @@ lazy_static! { #[async_trait::async_trait] impl StorageAPI for ECStore { + #[tracing::instrument(skip(self))] async fn backend_info(&self) -> madmin::BackendInfo { let (standard_sc_parity, rr_sc_parity) = { if let Some(sc) = GLOBAL_StorageClass.get() { @@ -1290,6 +1291,7 @@ impl StorageAPI for ECStore { ..Default::default() } } + #[tracing::instrument(skip(self))] async fn storage_info(&self) -> madmin::StorageInfo { let Some(notification_sy) = get_global_notification_sys() else { return madmin::StorageInfo::default(); @@ -1297,6 +1299,7 @@ impl StorageAPI for ECStore { notification_sy.storage_info(self).await } + #[tracing::instrument(skip(self))] async fn local_storage_info(&self) -> madmin::StorageInfo { let mut futures = Vec::with_capacity(self.pools.len()); @@ -1316,6 +1319,7 @@ impl StorageAPI for ECStore { madmin::StorageInfo { backend, disks } } + #[tracing::instrument(skip(self))] async fn list_bucket(&self, opts: &BucketOptions) -> Result> { // TODO: opts.cached @@ -1331,6 +1335,7 @@ impl StorageAPI for ECStore { Ok(buckets) } + #[tracing::instrument(skip(self))] async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()> { if is_meta_bucketname(bucket) { return Err(StorageError::BucketNameInvalid(bucket.to_string()).into()); @@ -1360,6 +1365,7 @@ impl StorageAPI for ECStore { .await?; Ok(()) } + #[tracing::instrument(skip(self))] async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> { if !is_meta_bucketname(bucket) { if let Err(err) = check_valid_bucket_name_strict(bucket) { @@ -1403,6 +1409,7 @@ impl StorageAPI for ECStore { Ok(()) } + #[tracing::instrument(skip(self))] async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result { let mut info = self .peer_sys @@ -1420,6 +1427,7 @@ impl StorageAPI for ECStore { } // TODO: review + #[tracing::instrument(skip(self))] async fn copy_object( &self, src_bucket: &str, @@ -1490,6 +1498,7 @@ impl StorageAPI for ECStore { } // TODO: review + #[tracing::instrument(skip(self))] async fn delete_objects( &self, bucket: &str, @@ -1643,6 +1652,7 @@ impl StorageAPI for ECStore { Ok((del_objects, del_errs)) } + #[tracing::instrument(skip(self))] async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result { check_del_obj_args(bucket, object)?; @@ -1717,6 +1727,7 @@ impl StorageAPI for ECStore { // @start_after as marker when continuation_token empty // @delimiter default="/", empty when recursive // @max_keys limit + #[tracing::instrument(skip(self))] async fn list_objects_v2( self: Arc, bucket: &str, @@ -1730,6 +1741,7 @@ impl StorageAPI for ECStore { self.inner_list_objects_v2(bucket, prefix, continuation_token, delimiter, max_keys, fetch_owner, start_after) .await } + #[tracing::instrument(skip(self))] async fn list_object_versions( self: Arc, bucket: &str, @@ -1742,6 +1754,7 @@ impl StorageAPI for ECStore { self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys) .await } + #[tracing::instrument(skip(self))] async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { check_object_args(bucket, object)?; @@ -1758,6 +1771,7 @@ impl StorageAPI for ECStore { Ok(info) } + #[tracing::instrument(skip(self))] async fn get_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { let object = encode_dir_object(object); @@ -1770,6 +1784,7 @@ impl StorageAPI for ECStore { Ok(oi.user_tags) } + #[tracing::instrument(skip(self))] async fn put_object_metadata(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { let object = encode_dir_object(object); if self.single_pool() { @@ -1796,6 +1811,7 @@ impl StorageAPI for ECStore { self.pools[idx].put_object_tags(bucket, object.as_str(), tags, opts).await } + #[tracing::instrument(skip(self))] async fn delete_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { let object = encode_dir_object(object); @@ -1807,6 +1823,8 @@ impl StorageAPI for ECStore { self.pools[idx].delete_object_tags(bucket, object.as_str(), opts).await } + + #[tracing::instrument(skip(self))] async fn copy_object_part( &self, src_bucket: &str, @@ -1828,6 +1846,7 @@ impl StorageAPI for ECStore { unimplemented!() } + #[tracing::instrument(skip(self, data))] async fn put_object_part( &self, bucket: &str, @@ -1859,6 +1878,7 @@ impl StorageAPI for ECStore { }; if let Some(err) = err { + error!("put_object_part err: {:?}", err); return Err(err); } } @@ -1870,6 +1890,7 @@ impl StorageAPI for ECStore { ))) } + #[tracing::instrument(skip(self))] async fn list_multipart_uploads( &self, bucket: &str, @@ -1917,6 +1938,8 @@ impl StorageAPI for ECStore { ..Default::default() }) } + + #[tracing::instrument(skip(self))] async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { check_new_multipart_args(bucket, object)?; @@ -1946,6 +1969,8 @@ impl StorageAPI for ECStore { self.pools[idx].new_multipart_upload(bucket, object, opts).await } + + #[tracing::instrument(skip(self))] async fn get_multipart_info( &self, bucket: &str, @@ -1981,6 +2006,7 @@ impl StorageAPI for ECStore { upload_id.to_owned(), ))) } + #[tracing::instrument(skip(self))] async fn abort_multipart_upload(&self, bucket: &str, object: &str, upload_id: &str, opts: &ObjectOptions) -> Result<()> { check_abort_multipart_args(bucket, object, upload_id)?; @@ -2016,6 +2042,7 @@ impl StorageAPI for ECStore { upload_id.to_owned(), ))) } + #[tracing::instrument(skip(self))] async fn complete_multipart_upload( &self, bucket: &str, @@ -2062,6 +2089,7 @@ impl StorageAPI for ECStore { ))) } + #[tracing::instrument(skip(self))] async fn get_disks(&self, pool_idx: usize, set_idx: usize) -> Result>> { if pool_idx < self.pools.len() && set_idx < self.pools[pool_idx].disk_set.len() { self.pools[pool_idx].disk_set[set_idx].get_disks(0, 0).await @@ -2070,6 +2098,7 @@ impl StorageAPI for ECStore { } } + #[tracing::instrument(skip(self))] fn set_drive_counts(&self) -> Vec { let mut counts = vec![0; self.pools.len()]; @@ -2078,6 +2107,8 @@ impl StorageAPI for ECStore { } counts } + + #[tracing::instrument(skip(self))] async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option)> { info!("heal_format"); let mut r = HealResultItem { @@ -2112,9 +2143,11 @@ impl StorageAPI for ECStore { Ok((r, None)) } + #[tracing::instrument(skip(self))] async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result { self.peer_sys.heal_bucket(bucket, opts).await } + #[tracing::instrument(skip(self))] async fn heal_object( &self, bucket: &str, @@ -2188,6 +2221,8 @@ impl StorageAPI for ECStore { Ok((HealResultItem::default(), Some(Error::new(DiskError::FileNotFound)))) } + + #[tracing::instrument(skip(self))] async fn heal_objects( &self, bucket: &str, @@ -2299,6 +2334,7 @@ impl StorageAPI for ECStore { Ok(()) } + #[tracing::instrument(skip(self))] async fn get_pool_and_set(&self, id: &str) -> Result<(Option, Option, Option)> { for (pool_idx, pool) in self.pools.iter().enumerate() { for (set_idx, set) in pool.format.erasure.sets.iter().enumerate() { @@ -2313,6 +2349,7 @@ impl StorageAPI for ECStore { Err(Error::new(DiskError::DiskNotFound)) } + #[tracing::instrument(skip(self))] async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()> { let object = utils::path::encode_dir_object(object); if self.single_pool() {