diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index a814baca3..799b987bc 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -43,7 +43,7 @@ use crate::utils::fs::{ }; use crate::utils::os::get_info; use crate::utils::path::{ - self, clean, decode_dir_object, encode_dir_object, has_suffix, path_join, path_join_buf, GLOBAL_DIR_SUFFIX, + clean, decode_dir_object, encode_dir_object, has_suffix, path_join, path_join_buf, GLOBAL_DIR_SUFFIX, GLOBAL_DIR_SUFFIX_WITH_SLASH, SLASH_SEPARATOR, }; use crate::{ @@ -69,7 +69,7 @@ use tokio::fs::{self, File}; use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWrite, AsyncWriteExt, ErrorKind}; use tokio::sync::mpsc::Sender; use tokio::sync::RwLock; -use tracing::{error, info, warn}; +use tracing::{debug, error, info, warn}; use uuid::Uuid; #[derive(Debug)] @@ -127,7 +127,7 @@ impl LocalDisk { // TODO: 删除 tmp 数据 } - let format_path = Path::new(super::RUSTFS_META_BUCKET) + let format_path = Path::new(RUSTFS_META_BUCKET) .join(Path::new(super::FORMAT_CONFIG_FILE)) .absolutize_virtually(&root)? .into_owned(); @@ -192,7 +192,7 @@ impl LocalDisk { let cache = Cache::new(update_fn, Duration::from_secs(1), Opts::default()); - // TODO: DIRECT suport + // TODO: DIRECT support // TODD: DiskInfo let mut disk = Self { root: root.clone(), @@ -272,10 +272,10 @@ impl LocalDisk { Ok(md) } async fn make_meta_volumes(&self) -> Result<()> { - let buckets = format!("{}/{}", super::RUSTFS_META_BUCKET, super::BUCKET_META_PREFIX); - let multipart = format!("{}/{}", super::RUSTFS_META_BUCKET, "multipart"); - let config = format!("{}/{}", super::RUSTFS_META_BUCKET, "config"); - let tmp = format!("{}/{}", super::RUSTFS_META_BUCKET, "tmp"); + let buckets = format!("{}/{}", RUSTFS_META_BUCKET, BUCKET_META_PREFIX); + let multipart = format!("{}/{}", RUSTFS_META_BUCKET, "multipart"); + let config = format!("{}/{}", RUSTFS_META_BUCKET, "config"); + let tmp = format!("{}/{}", RUSTFS_META_BUCKET, "tmp"); let defaults = vec![buckets.as_str(), multipart.as_str(), config.as_str(), tmp.as_str()]; self.make_volumes(defaults).await @@ -341,7 +341,7 @@ impl LocalDisk { rename(&delete_path, &trash_path).await.map_err(Error::new).err() }; - if immediate_purge || delete_path.to_string_lossy().ends_with(path::SLASH_SEPARATOR) { + if immediate_purge || delete_path.to_string_lossy().ends_with(SLASH_SEPARATOR) { warn!("move_to_trash immediate_purge {:?}", &delete_path.to_string_lossy()); let trash_path2 = self.get_object_path(super::RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?; let _ = rename_all( @@ -443,7 +443,7 @@ impl LocalDisk { return Err(Error::new(DiskError::FileNotFound)); } - let meta_path = file_path.as_ref().join(Path::new(super::STORAGE_FORMAT_FILE)); + let meta_path = file_path.as_ref().join(Path::new(STORAGE_FORMAT_FILE)); let res = { if read_data { @@ -530,12 +530,12 @@ impl LocalDisk { volume_dir: impl AsRef, file_path: impl AsRef, ) -> Result<(Vec, Option)> { - let mut f = match utils::fs::open_file(file_path.as_ref(), utils::fs::O_RDONLY).await { + let mut f = match utils::fs::open_file(file_path.as_ref(), O_RDONLY).await { Ok(f) => f, Err(e) => { if os_is_not_exist(&e) { if !skip_access_checks(volume) { - if let Err(er) = utils::fs::access(volume_dir.as_ref()).await { + if let Err(er) = access(volume_dir.as_ref()).await { if os_is_not_exist(&er) { warn!("read_all_data_with_dmtime os err {:?}", &er); return Err(Error::new(DiskError::VolumeNotFound)); @@ -553,7 +553,7 @@ impl LocalDisk { } else if is_sys_err_too_many_files(&e) { return Err(Error::new(DiskError::TooManyOpenFiles)); } else if is_sys_err_invalid_arg(&e) { - if let Ok(meta) = utils::fs::lstat(file_path.as_ref()).await { + if let Ok(meta) = lstat(file_path.as_ref()).await { if meta.is_dir() { return Err(Error::new(DiskError::FileNotFound)); } @@ -587,7 +587,7 @@ impl LocalDisk { async fn delete_versions_internal(&self, volume: &str, path: &str, fis: &Vec) -> Result<()> { let volume_dir = self.get_bucket_path(volume)?; - let xlpath = self.get_object_path(volume, format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str())?; + let xlpath = self.get_object_path(volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str())?; let data = match self.read_all_data_with_dmtime(volume, volume_dir.as_path(), &xlpath).await { Ok((data, _)) => data, @@ -650,14 +650,8 @@ impl LocalDisk { let volume_dir = self.get_bucket_path(volume)?; - self.write_all_private( - volume, - format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str(), - &buf, - true, - volume_dir, - ) - .await?; + self.write_all_private(volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str(), &buf, true, volume_dir) + .await?; Ok(()) } @@ -672,12 +666,12 @@ impl LocalDisk { self.write_all_internal(&tmp_file_path, buf, sync, tmp_volume_dir).await?; - os::rename_all(tmp_file_path, file_path, volume_dir).await + rename_all(tmp_file_path, file_path, volume_dir).await } // write_all_public for trail async fn write_all_public(&self, volume: &str, path: &str, data: Vec) -> Result<()> { - if volume == super::RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE { + if volume == RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE { let mut format_info = self.format_info.write().await; format_info.data.clone_from(&data); } @@ -713,7 +707,7 @@ impl LocalDisk { sync: bool, skip_parent: impl AsRef, ) -> Result<()> { - let flags = utils::fs::O_CREATE | utils::fs::O_WRONLY | utils::fs::O_TRUNC; + let flags = O_CREATE | O_WRONLY | utils::fs::O_TRUNC; let mut f = { if sync { @@ -924,7 +918,7 @@ impl LocalDisk { continue; } - let name = path::path_join_buf(&[current, entry]); + let name = path_join_buf(&[current, entry]); if !dir_stack.is_empty() { if let Some(pop) = dir_stack.pop() { @@ -1068,7 +1062,7 @@ fn skip_access_checks(p: impl AsRef) -> bool { super::RUSTFS_META_TMP_DELETED_BUCKET, super::RUSTFS_META_TMP_BUCKET, super::RUSTFS_META_MULTIPART_BUCKET, - super::RUSTFS_META_BUCKET, + RUSTFS_META_BUCKET, ]; for v in vols.iter() { @@ -1203,7 +1197,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 { + if volume == RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE { let format_info = self.format_info.read().await; if !format_info.data.is_empty() { return Ok(format_info.data.clone()); @@ -1225,7 +1219,7 @@ impl DiskAPI for LocalDisk { async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { - if let Err(e) = utils::fs::access(&volume_dir).await { + if let Err(e) = access(&volume_dir).await { return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); } } @@ -1243,7 +1237,7 @@ impl DiskAPI for LocalDisk { 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) { - if let Err(e) = utils::fs::access(&volume_dir).await { + if let Err(e) = access(&volume_dir).await { return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); } } @@ -1259,7 +1253,7 @@ impl DiskAPI for LocalDisk { .join(path) .join(fi.data_dir.map_or("".to_string(), |dir| dir.to_string())) .join(format!("part.{}", part.number)); - let err = (self + let err = self .bitrot_verify( &part_path, erasure.shard_file_size(part.size), @@ -1267,7 +1261,7 @@ impl DiskAPI for LocalDisk { &checksum_info.hash, erasure.shard_size(erasure.block_size), ) - .await) + .await .err(); resp.results[i] = conv_part_err_to_int(&err); if resp.results[i] == CHECK_PART_UNKNOWN { @@ -1384,7 +1378,7 @@ impl DiskAPI for LocalDisk { } } - if let Err(e) = utils::fs::remove_std(&dst_file_path) { + if let Err(e) = remove_std(&dst_file_path) { if is_sys_err_not_empty(&e) || is_sys_err_not_dir(&e) { warn!("rename_part remove dst failed {:?} err {:?}", &dst_file_path, e); return Err(Error::new(DiskError::FileAccessDenied)); @@ -1396,7 +1390,7 @@ impl DiskAPI for LocalDisk { } } - if let Err(err) = os::rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await { + if let Err(err) = rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await { if let Some(e) = err.to_io_err() { if is_sys_err_not_empty(&e) || is_sys_err_not_dir(&e) { warn!("rename_part rename all failed {:?} err {:?}", &dst_file_path, e); @@ -1429,7 +1423,7 @@ impl DiskAPI for LocalDisk { let src_volume_dir = self.get_bucket_path(src_volume)?; let dst_volume_dir = self.get_bucket_path(dst_volume)?; if !skip_access_checks(src_volume) { - if let Err(e) = utils::fs::access(&src_volume_dir).await { + if let Err(e) = access(&src_volume_dir).await { if os_is_not_exist(&e) { return Err(Error::from(DiskError::VolumeNotFound)); } else if is_sys_err_io(&e) { @@ -1440,7 +1434,7 @@ impl DiskAPI for LocalDisk { } } if !skip_access_checks(dst_volume) { - if let Err(e) = utils::fs::access(&dst_volume_dir).await { + if let Err(e) = access(&dst_volume_dir).await { if os_is_not_exist(&e) { return Err(Error::from(DiskError::VolumeNotFound)); } else if is_sys_err_io(&e) { @@ -1484,7 +1478,7 @@ impl DiskAPI for LocalDisk { } } - if let Err(e) = utils::fs::remove(&dst_file_path).await { + if let Err(e) = remove(&dst_file_path).await { if is_sys_err_not_empty(&e) || is_sys_err_not_dir(&e) { return Err(Error::new(DiskError::FileAccessDenied)); } else if is_sys_err_io(&e) { @@ -1495,7 +1489,7 @@ impl DiskAPI for LocalDisk { } } - if let Err(err) = os::rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await { + if let Err(err) = rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await { if let Some(e) = err.to_io_err() { if is_sys_err_not_empty(&e) || is_sys_err_not_dir(&e) { return Err(Error::new(DiskError::FileAccessDenied)); @@ -1521,7 +1515,7 @@ impl DiskAPI for LocalDisk { if !origvolume.is_empty() { let origvolume_dir = self.get_bucket_path(origvolume)?; if !skip_access_checks(origvolume) { - if let Err(e) = utils::fs::access(origvolume_dir).await { + if let Err(e) = access(origvolume_dir).await { return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); } } @@ -1552,7 +1546,7 @@ impl DiskAPI for LocalDisk { let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { - if let Err(e) = utils::fs::access(&volume_dir).await { + if let Err(e) = access(&volume_dir).await { return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); } } @@ -1571,7 +1565,7 @@ impl DiskAPI for LocalDisk { // warn!("disk read_file: volume: {}, path: {}", volume, path); let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { - if let Err(e) = utils::fs::access(&volume_dir).await { + if let Err(e) = access(&volume_dir).await { return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); } } @@ -1609,7 +1603,7 @@ impl DiskAPI for LocalDisk { let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { - if let Err(e) = utils::fs::access(&volume_dir).await { + if let Err(e) = access(&volume_dir).await { return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); } } @@ -1655,7 +1649,7 @@ impl DiskAPI for LocalDisk { if !origvolume.is_empty() { let origvolume_dir = self.get_bucket_path(origvolume)?; if !skip_access_checks(origvolume) { - if let Err(e) = utils::fs::access(origvolume_dir).await { + if let Err(e) = access(origvolume_dir).await { return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); } } @@ -1668,7 +1662,7 @@ impl DiskAPI for LocalDisk { Ok(res) => res, Err(e) => { if is_err_file_not_found(&e) && !skip_access_checks(volume) { - if let Err(e) = utils::fs::access(&volume_dir).await { + if let Err(e) = access(&volume_dir).await { return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); } } @@ -1750,8 +1744,8 @@ impl DiskAPI for LocalDisk { } // xl.meta 路径 - let src_file_path = src_volume_dir.join(Path::new(format!("{}/{}", &src_path, super::STORAGE_FORMAT_FILE).as_str())); - let dst_file_path = dst_volume_dir.join(Path::new(format!("{}/{}", &dst_path, super::STORAGE_FORMAT_FILE).as_str())); + let src_file_path = src_volume_dir.join(Path::new(format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str())); + let dst_file_path = dst_volume_dir.join(Path::new(format!("{}/{}", &dst_path, STORAGE_FORMAT_FILE).as_str())); // data_dir 路径 let has_data_dir_path = { @@ -1846,7 +1840,7 @@ impl DiskAPI for LocalDisk { let new_dst_buf = xlmeta.marshal_msg()?; - self.write_all(src_volume, format!("{}/{}", &src_path, super::STORAGE_FORMAT_FILE).as_str(), new_dst_buf) + self.write_all(src_volume, format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str(), new_dst_buf) .await .map_err(|err| { if let Some(e) = err.to_io_err() { @@ -1858,7 +1852,7 @@ impl DiskAPI for LocalDisk { if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() { let no_inline = fi.data.is_none() && fi.size > 0; if no_inline { - if let Err(err) = os::rename_all(&src_data_path, &dst_data_path, &skip_parent).await { + if let Err(err) = rename_all(&src_data_path, &dst_data_path, &skip_parent).await { let _ = self.delete_file(&dst_volume_dir, dst_data_path, false, false).await; info!( "rename all failed src_data_path: {:?}, dst_data_path: {:?}, err: {:?}", @@ -1881,7 +1875,7 @@ impl DiskAPI for LocalDisk { if let Err(err) = self .write_all_private( dst_volume, - format!("{}/{}/{}", &dst_path, &old_data_dir.to_string(), super::STORAGE_FORMAT_FILE).as_str(), + format!("{}/{}/{}", &dst_path, &old_data_dir.to_string(), STORAGE_FORMAT_FILE).as_str(), &dst_buf, true, &skip_parent, @@ -1900,7 +1894,7 @@ impl DiskAPI for LocalDisk { } } - if let Err(err) = os::rename_all(&src_file_path, &dst_file_path, &skip_parent).await { + if let Err(err) = rename_all(&src_file_path, &dst_file_path, &skip_parent).await { if let Some((_, dst_data_path)) = has_data_dir_path.as_ref() { let _ = self.delete_file(&dst_volume_dir, dst_data_path, false, false).await; } @@ -1916,7 +1910,7 @@ impl DiskAPI for LocalDisk { if let Some(src_file_path_parent) = src_file_path.parent() { if src_volume != super::RUSTFS_META_MULTIPART_BUCKET { - let _ = utils::fs::remove_std(src_file_path_parent); + let _ = remove_std(src_file_path_parent); } else { let _ = self .delete_file(&dst_volume_dir, &src_file_path_parent.to_path_buf(), true, false) @@ -1951,7 +1945,7 @@ impl DiskAPI for LocalDisk { let volume_dir = self.get_bucket_path(volume)?; - if let Err(e) = utils::fs::access(&volume_dir).await { + if let Err(e) = access(&volume_dir).await { if os_is_not_exist(&e) { os::make_dir_all(&volume_dir, self.root.as_path()).await?; return Ok(()); @@ -1981,7 +1975,7 @@ impl DiskAPI for LocalDisk { })?; for entry in entries { - if !utils::path::has_suffix(&entry, SLASH_SEPARATOR) || !Self::is_valid_volname(utils::path::clean(&entry).as_str()) { + if !has_suffix(&entry, SLASH_SEPARATOR) || !Self::is_valid_volname(clean(&entry).as_str()) { continue; } @@ -1997,17 +1991,17 @@ impl DiskAPI for LocalDisk { #[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 { + let meta = match lstat(&volume_dir).await { Ok(res) => res, Err(e) => { - if os_is_not_exist(&e) { - return Err(Error::new(DiskError::VolumeNotFound)); + return if os_is_not_exist(&e) { + Err(Error::new(DiskError::VolumeNotFound)) } else if os_is_permission(&e) { - return Err(Error::new(DiskError::DiskAccessDenied)); + Err(Error::new(DiskError::DiskAccessDenied)) } else if is_sys_err_io(&e) { - return Err(Error::new(DiskError::FaultyDisk)); + Err(Error::new(DiskError::FaultyDisk)) } else { - return Err(Error::new(e)); + Err(Error::new(e)) } } }; @@ -2027,7 +2021,7 @@ impl DiskAPI for LocalDisk { async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()> { let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { - utils::fs::access(&volume_dir) + access(&volume_dir) .await .map_err(|e| convert_access_error(e, DiskError::VolumeAccessDenied))? } @@ -2052,7 +2046,7 @@ impl DiskAPI for LocalDisk { check_path_length(file_path.to_string_lossy().as_ref())?; let buf = self - .read_all(volume, format!("{}/{}", &path, super::STORAGE_FORMAT_FILE).as_str()) + .read_all(volume, format!("{}/{}", &path, STORAGE_FORMAT_FILE).as_str()) .await .map_err(|e| { if is_err_file_not_found(&e) && fi.version_id.is_some() { @@ -2073,12 +2067,7 @@ impl DiskAPI for LocalDisk { let wbuf = xl_meta.marshal_msg()?; return self - .write_all_meta( - volume, - format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str(), - &wbuf, - !opts.no_persistence, - ) + .write_all_meta(volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str(), &wbuf, !opts.no_persistence) .await; } @@ -2087,7 +2076,7 @@ impl DiskAPI for LocalDisk { #[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())?; + let p = self.get_object_path(volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str())?; let mut meta = FileMeta::new(); if !fi.fresh { @@ -2103,10 +2092,10 @@ impl DiskAPI for LocalDisk { let fm_data = meta.marshal_msg()?; - self.write_all(volume, format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str(), fm_data) + self.write_all(volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str(), fm_data) .await?; - return Ok(()); + Ok(()) } #[tracing::instrument(level = "debug", skip(self))] @@ -2182,11 +2171,11 @@ impl DiskAPI for LocalDisk { return self.write_metadata("", volume, path, fi).await; } - if fi.version_id.is_some() { - return Err(Error::new(DiskError::FileVersionNotFound)); + return if fi.version_id.is_some() { + Err(Error::new(DiskError::FileVersionNotFound)) } else { - return Err(Error::new(DiskError::FileNotFound)); - } + Err(Error::new(DiskError::FileNotFound)) + }; } }; @@ -2357,7 +2346,7 @@ impl DiskAPI for LocalDisk { self.scanning.fetch_add(1, Ordering::SeqCst); defer!(|| { self.scanning.fetch_sub(1, Ordering::SeqCst) }); - // must befor metadata_sys + // must before metadata_sys let Some(store) = new_object_layer_fn() else { return Err(Error::msg("errServerNotInitialized")) }; let mut cache = cache.clone(); @@ -2375,7 +2364,7 @@ impl DiskAPI for LocalDisk { } } - let vcfg = (BucketVersioningSys::get(&cache.info.name).await).ok(); + let vcfg = BucketVersioningSys::get(&cache.info.name).await.ok(); let loc = self.get_disk_location(); let disks = store.get_disks(loc.pool_idx.unwrap(), loc.disk_idx.unwrap()).await?; @@ -2489,7 +2478,7 @@ impl DiskAPI for LocalDisk { ) .await?; data_usage_info.info.last_update = Some(SystemTime::now()); - info!("ns_scanner completed: {data_usage_info:?}"); + debug!("ns_scanner completed: {data_usage_info:?}"); Ok(data_usage_info) } @@ -2546,7 +2535,7 @@ mod test { super::super::RUSTFS_META_TMP_DELETED_BUCKET, super::super::RUSTFS_META_TMP_BUCKET, super::super::RUSTFS_META_MULTIPART_BUCKET, - super::super::RUSTFS_META_BUCKET, + RUSTFS_META_BUCKET, ]; let paths: Vec<_> = vols.iter().map(|v| Path::new(v).join("test")).collect(); diff --git a/ecstore/src/file_meta.rs b/ecstore/src/file_meta.rs index 336b11e3c..7d3f8803b 100644 --- a/ecstore/src/file_meta.rs +++ b/ecstore/src/file_meta.rs @@ -349,9 +349,9 @@ impl FileMeta { self.versions.sort_by(|a, b| { match (a.header.mod_time, b.header.mod_time) { (Some(a_time), Some(b_time)) => b_time.cmp(&a_time), // Descending order - (Some(_), None) => std::cmp::Ordering::Less, - (None, Some(_)) => std::cmp::Ordering::Greater, - (None, None) => std::cmp::Ordering::Equal, + (Some(_), None) => Ordering::Less, + (None, Some(_)) => Ordering::Greater, + (None, None) => Ordering::Equal, } }); } @@ -515,18 +515,18 @@ impl FileMeta { continue; } - match ver.header.version_type { - VersionType::Invalid => return Err(Error::msg("invalid file meta version")), - VersionType::Delete => return Ok(None), + return match ver.header.version_type { + VersionType::Invalid => Err(Error::msg("invalid file meta version")), + VersionType::Delete => Ok(None), VersionType::Object => { let v = self.get_idx(i)?; self.versions.remove(i); let a = v.object.map(|v| v.data_dir).unwrap_or_default(); - return Ok(a); + Ok(a) } - } + }; } Err(Error::new(DiskError::FileVersionNotFound)) @@ -1079,20 +1079,20 @@ impl PartialOrd for FileMetaVersionHeader { impl Ord for FileMetaVersionHeader { fn cmp(&self, other: &Self) -> Ordering { match self.mod_time.cmp(&other.mod_time) { - core::cmp::Ordering::Equal => {} + Ordering::Equal => {} ord => return ord, } match self.version_type.cmp(&other.version_type) { - core::cmp::Ordering::Equal => {} + Ordering::Equal => {} ord => return ord, } match self.signature.cmp(&other.signature) { - core::cmp::Ordering::Equal => {} + Ordering::Equal => {} ord => return ord, } match self.version_id.cmp(&other.version_id) { - core::cmp::Ordering::Equal => {} + Ordering::Equal => {} ord => return ord, } self.flags.cmp(&other.flags) diff --git a/ecstore/src/heal/data_scanner_metric.rs b/ecstore/src/heal/data_scanner_metric.rs index d96ad89c6..dedd4417f 100644 --- a/ecstore/src/heal/data_scanner_metric.rs +++ b/ecstore/src/heal/data_scanner_metric.rs @@ -14,7 +14,7 @@ use std::{ time::SystemTime, }; use tokio::sync::RwLock; -use tracing::info; +use tracing::{debug, info}; use super::data_scanner::{CurrentScannerCycle, UpdateCurrentPathFn}; @@ -149,7 +149,7 @@ impl ScannerMetrics { } pub async fn set_cycle(&mut self, c: Option) { - info!("ScannerMetrics set_cycle {c:?}"); + debug!("ScannerMetrics set_cycle {c:?}"); *self.cycle_info.write().await = c; } diff --git a/ecstore/src/heal/heal_ops.rs b/ecstore/src/heal/heal_ops.rs index f2e065fc7..f53719ec0 100644 --- a/ecstore/src/heal/heal_ops.rs +++ b/ecstore/src/heal/heal_ops.rs @@ -57,18 +57,18 @@ pub type HealEntryFn = pub const BG_HEALING_UUID: &str = "0000-0000-0000-0000"; pub const HEALING_TRACKER_FILENAME: &str = ".healing.bin"; -const KEEP_HEAL_SEQ_STATE_DURATION: std::time::Duration = Duration::from_secs(10 * 60); +const KEEP_HEAL_SEQ_STATE_DURATION: Duration = Duration::from_secs(10 * 60); const HEAL_NOT_STARTED_STATUS: &str = "not started"; const HEAL_RUNNING_STATUS: &str = "running"; const HEAL_STOPPED_STATUS: &str = "stopped"; const HEAL_FINISHED_STATUS: &str = "finished"; -pub const RUESTFS_RESERVED_BUCKET: &str = "rustfs"; -pub const RUESTFS_RESERVED_BUCKET_PATH: &str = "/rustfs"; +pub const RUSTFS_RESERVED_BUCKET: &str = "rustfs"; +pub const RUSTFS_RESERVED_BUCKET_PATH: &str = "/rustfs"; pub const LOGIN_PATH_PREFIX: &str = "/login"; const MAX_UNCONSUMED_HEAL_RESULT_ITEMS: usize = 1000; -const HEAL_UNCONSUMED_TIMEOUT: std::time::Duration = Duration::from_secs(24 * 60 * 60); +const HEAL_UNCONSUMED_TIMEOUT: Duration = Duration::from_secs(24 * 60 * 60); pub const NOP_HEAL: &str = ""; lazy_static! {} @@ -125,7 +125,7 @@ pub fn new_bg_heal_sequence() -> HealSequence { HealSequence { start_time: SystemTime::now(), client_token: BG_HEALING_UUID.to_string(), - bucket: RUESTFS_RESERVED_BUCKET.to_string(), + bucket: RUSTFS_RESERVED_BUCKET.to_string(), setting: hs, current_status: Arc::new(RwLock::new(HealSequenceStatus { summary: HEAL_NOT_STARTED_STATUS.to_string(), @@ -195,12 +195,12 @@ impl Default for HealSequence { } impl HealSequence { - pub fn new(bucket: &str, obj_profix: &str, client_addr: &str, hs: HealOpts, force_start: bool) -> Self { + pub fn new(bucket: &str, obj_prefix: &str, client_addr: &str, hs: HealOpts, force_start: bool) -> Self { let client_token = Uuid::new_v4().to_string(); Self { bucket: bucket.to_string(), - object: obj_profix.to_string(), + object: obj_prefix.to_string(), report_progress: true, client_token, client_address: client_addr.to_string(), @@ -354,14 +354,14 @@ impl HealSequence { self.count_failed(heal_type.clone()).await; } if !self.report_progress { - if let Some(err) = res.err { + return if let Some(err) = res.err { if err.to_string() == ERR_SKIP_FILE { return Ok(()); } - return Err(err); + Err(err) } else { - return Ok(()); - } + Ok(()) + }; } res.result.heal_item_type = heal_type.clone(); if let Some(err) = res.err.as_ref() { @@ -406,7 +406,7 @@ impl HealSequence { async fn traverse_and_heal(h: Arc) { let buckets_only = false; - let result = (Self::heal_items(h.clone(), buckets_only).await).err(); + let result = Self::heal_items(h.clone(), buckets_only).await.err(); let _ = h.traverse_and_heal_done_tx.read().await.send(result).await; } @@ -542,12 +542,12 @@ pub async fn heal_sequence_start(h: Arc) { match err { Some(err) => { let mut current_status_w = h.current_status.write().await; - (current_status_w).summary = HEAL_STOPPED_STATUS.to_string(); - (current_status_w).failure_detail = err.to_string(); + current_status_w.summary = HEAL_STOPPED_STATUS.to_string(); + current_status_w.failure_detail = err.to_string(); }, None => { let mut current_status_w = h.current_status.write().await; - (current_status_w).summary = HEAL_FINISHED_STATUS.to_string(); + current_status_w.summary = HEAL_FINISHED_STATUS.to_string(); } } } @@ -567,11 +567,11 @@ pub struct AllHealState { impl AllHealState { pub fn new(cleanup: bool) -> Arc { - let hstate = Arc::new(AllHealState::default()); + let state = Arc::new(AllHealState::default()); let (_, mut rx) = broadcast::channel(1); if cleanup { - let hstate_clone = hstate.clone(); - tokio::spawn(async move { + let state_clone = state.clone(); + spawn(async move { loop { select! { result = rx.recv() =>{ @@ -580,14 +580,14 @@ impl AllHealState { } } _ = sleep(Duration::from_secs(5 * 60)) => { - hstate_clone.periodic_heal_seqs_clean().await; + state_clone.periodic_heal_seqs_clean().await; } } } }); } - hstate + state } pub async fn pop_heal_local_disks(&self, heal_local_disks: &[Endpoint]) { @@ -698,13 +698,13 @@ impl AllHealState { let _ = self.mu.write().await; let now = SystemTime::now(); - let mut keys_to_reomve = Vec::new(); + let mut keys_to_remove = Vec::new(); for (k, v) in self.heal_seq_map.read().await.iter() { if v.has_ended().await && now.duration_since(*(v.end_time.read().await)).unwrap() > KEEP_HEAL_SEQ_STATE_DURATION { - keys_to_reomve.push(k.clone()) + keys_to_remove.push(k.clone()) } } - for key in keys_to_reomve.iter() { + for key in keys_to_remove.iter() { self.heal_seq_map.write().await.remove(key); } } @@ -808,7 +808,7 @@ impl AllHealState { // For background heal do nothing, do not spawn an unnecessary goroutine. } else { let heal_sequence_clone = heal_sequence.clone(); - tokio::spawn(async { + spawn(async { heal_sequence_start(heal_sequence_clone).await; }); } diff --git a/ecstore/src/peer.rs b/ecstore/src/peer.rs index f352121f9..89aa02be6 100644 --- a/ecstore/src/peer.rs +++ b/ecstore/src/peer.rs @@ -5,7 +5,7 @@ use crate::global::GLOBAL_LOCAL_DISK_MAP; use crate::heal::heal_commands::{ HealOpts, DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE, DRIVE_STATE_OK, HEAL_ITEM_BUCKET, }; -use crate::heal::heal_ops::RUESTFS_RESERVED_BUCKET; +use crate::heal::heal_ops::RUSTFS_RESERVED_BUCKET; use crate::quorum::{bucket_op_ignored_errs, reduce_write_quorum_errs}; use crate::store::all_local_disk; use crate::utils::proto_err_to_err; @@ -701,7 +701,7 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result 0 for is_valid() to return true + index: 1, // Must be > 0 for is_valid() to return true distribution: vec![1, 2, 3, 4, 5, 6], // Must match data_blocks + parity_blocks ..Default::default() }, @@ -5902,7 +5902,7 @@ mod tests { erasure: ErasureInfo { data_blocks: 6, parity_blocks: 3, - index: 1, // Must be > 0 for is_valid() to return true + index: 1, // Must be > 0 for is_valid() to return true distribution: vec![1, 2, 3, 4, 5, 6, 7, 8, 9], // Must match data_blocks + parity_blocks ..Default::default() }, @@ -5914,7 +5914,7 @@ mod tests { erasure: ErasureInfo { data_blocks: 2, parity_blocks: 1, - index: 1, // Must be > 0 for is_valid() to return true + index: 1, // Must be > 0 for is_valid() to return true distribution: vec![1, 2, 3], // Must match data_blocks + parity_blocks ..Default::default() }, @@ -6019,11 +6019,7 @@ mod tests { #[test] fn test_join_errs() { // Test joining error messages - let errs = vec![ - None, - Some(Error::from_string("error1")), - Some(Error::from_string("error2")), - ]; + let errs = vec![None, Some(Error::from_string("error1")), Some(Error::from_string("error2"))]; let joined = join_errs(&errs); assert!(joined.contains("")); assert!(joined.contains("error1")); diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index e61ff578d..598dc8818 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -1153,7 +1153,11 @@ mod tests { assert_eq!(file_info.get_etag(), None); // With etag - file_info.metadata.as_mut().unwrap().insert("etag".to_string(), "test-etag".to_string()); + file_info + .metadata + .as_mut() + .unwrap() + .insert("etag".to_string(), "test-etag".to_string()); assert_eq!(file_info.get_etag(), Some("test-etag".to_string())); } @@ -1282,10 +1286,7 @@ mod tests { file_info.set_healing(); assert!(file_info.metadata.is_some()); - assert_eq!( - file_info.metadata.as_ref().unwrap().get(RUSTFS_HEALING), - Some(&"true".to_string()) - ); + assert_eq!(file_info.metadata.as_ref().unwrap().get(RUSTFS_HEALING), Some(&"true".to_string())); } #[test] @@ -1656,10 +1657,11 @@ mod tests { assert!(!object_info.is_compressed()); // With compression metadata - object_info.user_defined.as_mut().unwrap().insert( - format!("{}compression", RESERVED_METADATA_PREFIX), - "gzip".to_string() - ); + object_info + .user_defined + .as_mut() + .unwrap() + .insert(format!("{}compression", RESERVED_METADATA_PREFIX), "gzip".to_string()); assert!(object_info.is_compressed()); } @@ -1866,7 +1868,7 @@ mod tests { let shard_size = erasure.shard_size(1000); assert_eq!(shard_size, 1000); // 1000 / 1 = 1000 - // Test with zero block size - this will cause division by zero in shard_size + // Test with zero block size - this will cause division by zero in shard_size // So we need to test with non-zero block_size but zero data_blocks was already fixed above let erasure = ErasureInfo { data_blocks: 4, diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index ca7070f16..85a934bd4 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -14,7 +14,7 @@ use crate::auth::IAMAuth; use crate::console::{init_console_cfg, CONSOLE_CONFIG}; // Ensure the correct path for parse_license is imported use crate::server::{wait_for_shutdown, ServiceState, ServiceStateManager, ShutdownSignal, SHUTDOWN_TIMEOUT}; -// use bytes::Bytes; +use bytes::Bytes; use chrono::Datelike; use clap::Parser; use common::{ @@ -37,7 +37,7 @@ use ecstore::{ }; use ecstore::{global::set_global_rustfs_port, notification_sys::new_global_notification_sys}; use grpc::make_server; -// use http::{HeaderMap, Request as HttpRequest, Response}; +use http::{HeaderMap, Request as HttpRequest, Response}; use hyper_util::server::graceful::GracefulShutdown; use hyper_util::{ rt::{TokioExecutor, TokioIo}, @@ -61,10 +61,9 @@ use tokio::signal::unix::{signal, SignalKind}; use tokio_rustls::TlsAcceptor; use tonic::{metadata::MetadataValue, Request, Status}; use tower_http::cors::CorsLayer; -// use tracing::{instrument, Span}; -use tracing::instrument; -// use tower_http::trace::TraceLayer; +use tower_http::trace::TraceLayer; use tracing::{debug, error, info, warn}; +use tracing::{instrument, Span}; const MI_B: usize = 1024 * 1024; @@ -324,49 +323,49 @@ async fn run(opt: config::Opt) -> Result<()> { let mut sigint_inner = sigint_inner; let hybrid_service = TowerToHyperService::new( tower::ServiceBuilder::new() - //.layer( - // TraceLayer::new_for_http() - // .make_span_with(|request: &HttpRequest<_>| { - // let span = tracing::info_span!("http-request", - // status_code = tracing::field::Empty, - // method = %request.method(), - // uri = %request.uri(), - // version = ?request.version(), - // ); - // for (header_name, header_value) in request.headers() { - // if header_name == "user-agent" || header_name == "content-type" || header_name == "content-length" - // { - // span.record(header_name.as_str(), header_value.to_str().unwrap_or("invalid")); - // } - // } - // - // span - // }) - // .on_request(|request: &HttpRequest<_>, _span: &Span| { - // info!( - // counter.rustfs_api_requests_total = 1_u64, - // key_request_method = %request.method().to_string(), - // key_request_uri_path = %request.uri().path().to_owned(), - // "handle request api total", - // ); - // debug!("http started method: {}, url path: {}", request.method(), request.uri().path()) - // }) - // .on_response(|response: &Response<_>, latency: Duration, _span: &Span| { - // _span.record("http response status_code", tracing::field::display(response.status())); - // debug!("http response generated in {:?}", latency) - // }) - // .on_body_chunk(|chunk: &Bytes, latency: Duration, _span: &Span| { - // info!(histogram.request.body.len = chunk.len(), "histogram request body length",); - // debug!("http body sending {} bytes in {:?}", chunk.len(), latency) - // }) - // .on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, _span: &Span| { - // debug!("http stream closed after {:?}", stream_duration) - // }) - // .on_failure(|_error, latency: Duration, _span: &Span| { - // info!(counter.rustfs_api_requests_failure_total = 1_u64, "handle request api failure total"); - // debug!("http request failure error: {:?} in {:?}", _error, latency) - // }), - // ) + .layer( + TraceLayer::new_for_http() + .make_span_with(|request: &HttpRequest<_>| { + let span = tracing::info_span!("http-request", + status_code = tracing::field::Empty, + method = %request.method(), + uri = %request.uri(), + version = ?request.version(), + ); + for (header_name, header_value) in request.headers() { + if header_name == "user-agent" || header_name == "content-type" || header_name == "content-length" + { + span.record(header_name.as_str(), header_value.to_str().unwrap_or("invalid")); + } + } + + span + }) + .on_request(|request: &HttpRequest<_>, _span: &Span| { + info!( + counter.rustfs_api_requests_total = 1_u64, + key_request_method = %request.method().to_string(), + key_request_uri_path = %request.uri().path().to_owned(), + "handle request api total", + ); + debug!("http started method: {}, url path: {}", request.method(), request.uri().path()) + }) + .on_response(|response: &Response<_>, latency: Duration, _span: &Span| { + _span.record("http response status_code", tracing::field::display(response.status())); + debug!("http response generated in {:?}", latency) + }) + .on_body_chunk(|chunk: &Bytes, latency: Duration, _span: &Span| { + info!(histogram.request.body.len = chunk.len(), "histogram request body length",); + debug!("http body sending {} bytes in {:?}", chunk.len(), latency) + }) + .on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, _span: &Span| { + debug!("http stream closed after {:?}", stream_duration) + }) + .on_failure(|_error, latency: Duration, _span: &Span| { + info!(counter.rustfs_api_requests_failure_total = 1_u64, "handle request api failure total"); + debug!("http request failure error: {:?} in {:?}", _error, latency) + }), + ) .layer(CorsLayer::permissive()) .service(hybrid(s3_service, rpc_service)), ); @@ -452,7 +451,8 @@ async fn run(opt: config::Opt) -> Result<()> { let conn = http_server_clone.serve_connection(TokioIo::new(tls_socket), value_clone); let conn = graceful_clone.watch(conn); if let Err(err) = conn.await { - error!("Https Connection error: {}", err); + // Handle hyper::Error and low-level IO errors at a more granular level + handle_connection_error(&*err); } }); }); @@ -467,7 +467,8 @@ async fn run(opt: config::Opt) -> Result<()> { let conn = http_server_clone.serve_connection(TokioIo::new(socket), value_clone); let conn = graceful_clone.watch(conn); if let Err(err) = conn.await { - error!("Http Connection error: {}", err); + // Handle hyper::Error and low-level IO errors at a more granular level + handle_connection_error(&*err); } }); debug!("Http handshake success"); @@ -584,3 +585,25 @@ async fn run(opt: config::Opt) -> Result<()> { info!("server is stopped state: {:?}", state_manager.current_state()); Ok(()) } + +fn handle_connection_error(err: &(dyn std::error::Error + 'static)) { + if let Some(hyper_err) = err.downcast_ref::() { + if hyper_err.is_incomplete_message() { + warn!("The HTTP connection is closed prematurely and the message is not completed:{}", hyper_err); + } else if hyper_err.is_closed() { + warn!("The HTTP connection is closed:{}", hyper_err); + } else if hyper_err.is_parse() { + error!("HTTP message parsing failed:{}", hyper_err); + } else if hyper_err.is_user() { + error!("HTTP user-custom error:{}", hyper_err); + } else if hyper_err.is_canceled() { + warn!("The HTTP connection is canceled:{}", hyper_err); + } else { + error!("Unknown hyper error:{:?}", hyper_err); + } + } else if let Some(io_err) = err.downcast_ref::() { + error!("Unknown connection IO error:{}", io_err); + } else { + error!("Unknown connection error type:{:?}", err); + } +}