gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix typo

Browse Source
This commit is contained in:
houseme
2025-05-27 16:56:44 +08:00
parent ca8f399832
commit ade4d33eb1
8 changed files with 196 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/disk/local.rs
+67 -78
View File
@@ -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<Path>,
file_path: impl AsRef<Path>,
) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
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<FileInfo>) -> 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<u8>) -> 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<Path>,
) -> 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<str>) -> 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<Vec<u8>> {
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<CheckPartsResp> {
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<VolumeInfo> {
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();
ecstore/src/file_meta.rs
+12 -12
View File
@@ -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)
ecstore/src/heal/data_scanner_metric.rs
+2 -2
View File
@@ -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<CurrentScannerCycle>) {
info!("ScannerMetrics set_cycle {c:?}");
debug!("ScannerMetrics set_cycle {c:?}");
*self.cycle_info.write().await = c;
}
ecstore/src/heal/heal_ops.rs
+24 -24
View File
@@ -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<HealSequence>) {
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<HealSequence>) {
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<Self> {
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;
});
}
ecstore/src/peer.rs
+2 -2
View File
@@ -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<HealResu
bs_clone.write().await[index] = DRIVE_STATE_OK.to_string();
as_clone.write().await[index] = DRIVE_STATE_OK.to_string();
if bucket == RUESTFS_RESERVED_BUCKET {
if bucket == RUSTFS_RESERVED_BUCKET {
return None;
}
ecstore/src/set_disk.rs
+4 -8
View File
@@ -5890,7 +5890,7 @@ mod tests {
erasure: ErasureInfo {
data_blocks: 4,
parity_blocks: 2,
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], // 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("<nil>"));
assert!(joined.contains("error1"));
ecstore/src/store_api.rs
+12 -10
View File
@@ -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,
rustfs/src/main.rs
+73 -50
View File
@@ -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::<hyper::Error>() {
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::<std::io::Error>() {
error!("Unknown connection IO error:{}", io_err);
} else {
error!("Unknown connection error type:{:?}", err);
}
}