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:#355 multi pools select error

Browse Source
This commit is contained in:
weisd
2025-04-25 09:54:41 +08:00
parent 7dae5f8ab7
commit 9fc4bb919e
8 changed files with 275 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/disk/error.rs
+4
View File
@@ -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::<io::Error>() {
return ioerr.kind() == ErrorKind::NotFound;
}
matches!(err.downcast_ref::<DiskError>(), Some(DiskError::FileNotFound))
}
ecstore/src/disk/local.rs
+38 -1
View File
@@ -1079,30 +1079,39 @@ fn skip_access_checks(p: impl AsRef<str>) -> 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<Uuid>) -> 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<Vec<u8>> {
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<u8>) -> 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<CheckPartsResp> {
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<CheckPartsResp> {
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<Vec<VolumeInfo>> {
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<VolumeInfo> {
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<Vec<ReadMultipleResp>> {
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<DiskInfo> {
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<HealingTracker> {
let healing_file = path_join(&[
self.path(),
ecstore/src/disk/mod.rs
+48 -1
View File
@@ -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<Option<Uuid>> {
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<Uuid>) -> 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<Vec<u8>> {
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<u8>) -> 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<CheckPartsResp> {
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<CheckPartsResp> {
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<u8>) -> 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<FileWriter> {
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<FileWriter> {
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<FileReader> {
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<FileReader> {
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<Vec<String>> {
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<W: AsyncWrite + Unpin + Send>(&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<Vec<VolumeInfo>> {
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<VolumeInfo> {
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<RawFileInfo> {
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<Vec<ReadMultipleResp>> {
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<DiskInfo> {
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<HealingTracker> {
match self {
Disk::Local(local_disk) => local_disk.healing().await,
ecstore/src/disk/remote.rs
+39
View File
@@ -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<Option<Uuid>> {
Ok(*self.id.lock().await)
}
#[tracing::instrument(skip(self))]
async fn set_disk_id(&self, id: Option<Uuid>) -> 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<Vec<u8>> {
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<u8>) -> 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<CheckPartsResp> {
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<CheckPartsResp> {
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<u8>) -> 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<Vec<String>> {
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<W: AsyncWrite + Unpin + Send>(&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<Vec<VolumeInfo>> {
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<VolumeInfo> {
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<Vec<ReadMultipleResp>> {
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<DiskInfo> {
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<HealingTracker> {
None
}
ecstore/src/erasure.rs
+2 -5
View File
@@ -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))
}
ecstore/src/set_disk.rs
+61 -9
View File
@@ -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<DiskStore>], 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<DiskStore>],
src_bucket: &str,
@@ -580,6 +583,7 @@ impl SetDisks {
// errors
// }
#[tracing::instrument(skip(disks, files))]
async fn write_unique_file_info(
disks: &[Option<DiskStore>],
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<Vec<BucketInfo>> {
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<BucketInfo> {
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<ObjectInfo> {
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<Self>,
_bucket: &str,
@@ -4146,6 +4169,8 @@ impl StorageAPI for SetDisks {
) -> Result<ListObjectsV2Info> {
unimplemented!()
}
#[tracing::instrument(skip(self))]
async fn list_object_versions(
self: Arc<Self>,
_bucket: &str,
@@ -4157,6 +4182,8 @@ impl StorageAPI for SetDisks {
) -> Result<ListObjectVersionsInfo> {
unimplemented!()
}
#[tracing::instrument(skip(self))]
async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo> {
// 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<ObjectInfo> {
// 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<String> {
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<ObjectInfo> {
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<MultipartUploadResult> {
warn!("new_multipart_upload opt {:?}", opts);
#[tracing::instrument(skip(self))]
async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<MultipartUploadResult> {
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<Vec<Option<DiskStore>>> {
Ok(self.get_disks_internal().await)
}
#[tracing::instrument(skip(self))]
fn set_drive_counts(&self) -> Vec<usize> {
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<Error>)> {
unimplemented!()
}
#[tracing::instrument(skip(self))]
async fn heal_bucket(&self, _bucket: &str, _opts: &HealOpts) -> Result<HealResultItem> {
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<usize>, Option<usize>, Option<usize>)> {
unimplemented!()
}
#[tracing::instrument(skip(self))]
async fn check_abandoned_parts(&self, _bucket: &str, _object: &str, _opts: &HealOpts) -> Result<()> {
unimplemented!()
}
ecstore/src/sets.rs
+46
View File
@@ -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<Vec<BucketInfo>> {
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<BucketInfo> {
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<ObjectInfo> {
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<Self>,
_bucket: &str,
@@ -518,6 +532,8 @@ impl StorageAPI for Sets {
) -> Result<ListObjectsV2Info> {
unimplemented!()
}
#[tracing::instrument(skip(self))]
async fn list_object_versions(
self: Arc<Self>,
_bucket: &str,
@@ -529,14 +545,18 @@ impl StorageAPI for Sets {
) -> Result<ListObjectVersionsInfo> {
unimplemented!()
}
#[tracing::instrument(skip(self))]
async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo> {
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<ObjectInfo> {
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<String> {
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<ObjectInfo> {
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<MultipartUploadResult> {
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<Vec<Option<DiskStore>>> {
unimplemented!()
}
#[tracing::instrument(skip(self))]
fn set_drive_counts(&self) -> Vec<usize> {
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<Error>)> {
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<HealResultItem> {
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<usize>, Option<usize>, Option<usize>)> {
unimplemented!()
}
#[tracing::instrument(skip(self))]
async fn check_abandoned_parts(&self, _bucket: &str, _object: &str, _opts: &HealOpts) -> Result<()> {
unimplemented!()
}
ecstore/src/store.rs
+37
View File
@@ -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<Vec<BucketInfo>> {
// 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<BucketInfo> {
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<ObjectInfo> {
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<Self>,
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<Self>,
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<ObjectInfo> {
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<String> {
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<ObjectInfo> {
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<ObjectInfo> {
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<MultipartUploadResult> {
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<Vec<Option<DiskStore>>> {
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<usize> {
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<Error>)> {
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<HealResultItem> {
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<usize>, Option<usize>, Option<usize>)> {
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() {