From 2b3f03eae90cb013ae2e160b79c899e756d193a2 Mon Sep 17 00:00:00 2001 From: weisd Date: Wed, 18 Sep 2024 17:04:42 +0800 Subject: [PATCH 1/6] add log --- ecstore/src/store_init.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ecstore/src/store_init.rs b/ecstore/src/store_init.rs index 6e6eb2fc1..91d580f31 100644 --- a/ecstore/src/store_init.rs +++ b/ecstore/src/store_init.rs @@ -65,8 +65,9 @@ pub async fn connect_load_init_formats( // new format and save let fms = init_format_erasure(disks, set_count, set_drive_count, deployment_id); - let _errs = save_format_file_all(disks, &fms).await; + let errs = save_format_file_all(disks, &fms).await; + warn!("save_format_file_all errs {:?}", &errs); // TODO: check quorum // reduceWriteQuorumErrs(&errs)?; From 60d02f3afdaefeae206bb633b4f5b099237f32ae Mon Sep 17 00:00:00 2001 From: weisd Date: Wed, 18 Sep 2024 22:57:08 +0800 Subject: [PATCH 2/6] fix get_disk_id --- ecstore/src/disk/error.rs | 3 ++ ecstore/src/disk/local.rs | 85 ++++++++++++++++++++++++++++++++++----- 2 files changed, 77 insertions(+), 11 deletions(-) diff --git a/ecstore/src/disk/error.rs b/ecstore/src/disk/error.rs index c8c0c84b1..1b879d7ab 100644 --- a/ecstore/src/disk/error.rs +++ b/ecstore/src/disk/error.rs @@ -34,6 +34,9 @@ pub enum DiskError { #[error("volume not empty")] VolumeNotEmpty, + + #[error("corrupted backend")] + CorruptedBackend } impl DiskError { diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 5dfcb92d2..3c34f3368 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -26,17 +26,23 @@ use uuid::Uuid; #[derive(Debug)] pub struct FormatInfo { pub id: Option, - pub _data: Vec, - pub _file_info: Option, - pub _last_check: Option, + pub data: Vec, + pub file_info: Option, + pub last_check: Option, } -impl FormatInfo {} +impl FormatInfo { + pub fn last_check_valid(&self)->bool{ + let now = OffsetDateTime::now_utc(); + self.file_info.is_some() && self.id.is_some() && self.last_check.is_some() && (now.unix_timestamp() - self.last_check.unwrap().unix_timestamp() <=1) + } + +} #[derive(Debug)] pub struct LocalDisk { pub root: PathBuf, - pub _format_path: PathBuf, + pub format_path: PathBuf, pub format_info: Mutex, // pub id: Mutex>, // pub format_data: Mutex>, @@ -79,14 +85,14 @@ impl LocalDisk { let format_info = FormatInfo { id, - _data: format_data, - _file_info: format_meta, - _last_check: format_last_check, + data: format_data, + file_info: format_meta, + last_check: format_last_check, }; let disk = Self { root, - _format_path: format_path, + format_path: format_path, format_info: Mutex::new(format_info), // // format_legacy, // format_file_info: Mutex::new(format_meta), @@ -99,6 +105,17 @@ impl LocalDisk { Ok(disk) } + async fn check_format_json(&self) ->Result{ + let p = self.format_path; + let md = fs::metadata(&p).await.map_err(|e|match e.kind(){ + ErrorKind::NotFound => DiskError::DiskNotFound, + ErrorKind::PermissionDenied => DiskError::FileAccessDenied, + _ => { + warn!("check_format_json err {:?}",e); + DiskError::CorruptedBackend}, + })?; + 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"); @@ -441,9 +458,55 @@ impl DiskAPI for LocalDisk { async fn get_disk_id(&self) -> Option { warn!("local get_disk_id"); // TODO: check format file - let format_info = self.format_info.lock().await; + let mut format_info = self.format_info.lock().await; - format_info.id.clone() + let id = format_info.id.clone(); + + if format_info.last_check_valid(){ + return id + } + + let file_meta = self.check_format_json().await?; + + if let Some(file_info) = format_info.file_info{ + if file_meta == file_info{ + + format_info.last_check = Some(OffsetDateTime::now_utc()); + + return id + } + } + + + let b = fs::read(&format_info.file_path).await.map_err(|e|match e.kind(){ + ErrorKind::NotFound => DiskError::DiskNotFound, + ErrorKind::PermissionDenied => DiskError::FileAccessDenied, + _ => { + warn!("check_format_json err {:?}",e); + DiskError::CorruptedBackend}, + })?; + + + let fm = FormatV3::try_from(b.as_slice()).map_err(|e|{ + warn!("decode format.json err {:?}",e); + DiskError::CorruptedBackend + })?; + + let (m,n) = fm.find_disk_index_by_disk_id(fm.erasure.this)?; + + let disk_id = fm.erasure.this; + + if m != self.endpoint.set_idx || n != self.endpoint.disk_idx { + return DiskError::InconsistentDisk; + } + + format_info.id = Some(disk_id); + format_info.file_info= Some(file_meta); + format_info.data = b; + format_info.last_check = Some(OffsetDateTime::now_utc()); +; + +Some(disk_id) // TODO: 判断源文件id,是否有效 } From 1f1e73ce083d8508f41addab3153403d01892b4d Mon Sep 17 00:00:00 2001 From: weisd Date: Thu, 19 Sep 2024 09:41:29 +0800 Subject: [PATCH 3/6] fix get_disk_id --- ecstore/src/disk/local.rs | 106 ++++++++++++++++++++----------------- ecstore/src/disk/mod.rs | 2 +- ecstore/src/disk/remote.rs | 4 +- ecstore/src/sets.rs | 2 +- ecstore/src/utils/fs.rs | 24 +++++++++ ecstore/src/utils/mod.rs | 1 + 6 files changed, 85 insertions(+), 54 deletions(-) create mode 100644 ecstore/src/utils/fs.rs diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 3c34f3368..e0a1679ac 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -32,11 +32,13 @@ pub struct FormatInfo { } impl FormatInfo { - pub fn last_check_valid(&self)->bool{ + pub fn last_check_valid(&self) -> bool { let now = OffsetDateTime::now_utc(); - self.file_info.is_some() && self.id.is_some() && self.last_check.is_some() && (now.unix_timestamp() - self.last_check.unwrap().unix_timestamp() <=1) + self.file_info.is_some() + && self.id.is_some() + && self.last_check.is_some() + && (now.unix_timestamp() - self.last_check.unwrap().unix_timestamp() <= 1) } - } #[derive(Debug)] @@ -44,6 +46,7 @@ pub struct LocalDisk { pub root: PathBuf, pub format_path: PathBuf, pub format_info: Mutex, + pub endpoint: Endpoint, // pub id: Mutex>, // pub format_data: Mutex>, // pub format_file_info: Mutex>, @@ -92,6 +95,7 @@ impl LocalDisk { let disk = Self { root, + endpoint: ep.clone(), format_path: format_path, format_info: Mutex::new(format_info), // // format_legacy, @@ -105,16 +109,16 @@ impl LocalDisk { Ok(disk) } - async fn check_format_json(&self) ->Result{ - let p = self.format_path; - let md = fs::metadata(&p).await.map_err(|e|match e.kind(){ - ErrorKind::NotFound => DiskError::DiskNotFound, - ErrorKind::PermissionDenied => DiskError::FileAccessDenied, - _ => { - warn!("check_format_json err {:?}",e); - DiskError::CorruptedBackend}, - })?; - Ok(md) + async fn check_format_json(&self) -> Result { + let md = fs::metadata(&self.format_path).await.map_err(|e| match e.kind() { + ErrorKind::NotFound => DiskError::DiskNotFound, + ErrorKind::PermissionDenied => DiskError::FileAccessDenied, + _ => { + warn!("check_format_json err {:?}", e); + DiskError::CorruptedBackend + } + })?; + Ok(md) } async fn make_meta_volumes(&self) -> Result<()> { let buckets = format!("{}/{}", super::RUSTFS_META_BUCKET, super::BUCKET_META_PREFIX); @@ -455,58 +459,60 @@ impl DiskAPI for LocalDisk { self.root.clone() } - async fn get_disk_id(&self) -> Option { + async fn get_disk_id(&self) -> Result> { warn!("local get_disk_id"); // TODO: check format file let mut format_info = self.format_info.lock().await; - let id = format_info.id.clone(); + let id = format_info.id.clone(); - if format_info.last_check_valid(){ - return id - } - - let file_meta = self.check_format_json().await?; - - if let Some(file_info) = format_info.file_info{ - if file_meta == file_info{ - - format_info.last_check = Some(OffsetDateTime::now_utc()); - - return id + if format_info.last_check_valid() { + return Ok(id); } - } + let file_meta = self.check_format_json().await?; - let b = fs::read(&format_info.file_path).await.map_err(|e|match e.kind(){ - ErrorKind::NotFound => DiskError::DiskNotFound, - ErrorKind::PermissionDenied => DiskError::FileAccessDenied, - _ => { - warn!("check_format_json err {:?}",e); - DiskError::CorruptedBackend}, - })?; + if let Some(file_info) = &format_info.file_info { + if utils::fs::same_file(&file_meta, file_info) { + format_info.last_check = Some(OffsetDateTime::now_utc()); + return Ok(id); + } + } - let fm = FormatV3::try_from(b.as_slice()).map_err(|e|{ - warn!("decode format.json err {:?}",e); + let b = fs::read(&self.format_path).await.map_err(|e| match e.kind() { + ErrorKind::NotFound => DiskError::DiskNotFound, + ErrorKind::PermissionDenied => DiskError::FileAccessDenied, + _ => { + warn!("check_format_json err {:?}", e); + DiskError::CorruptedBackend + } + })?; + + let fm = FormatV3::try_from(b.as_slice()).map_err(|e| { + warn!("decode format.json err {:?}", e); DiskError::CorruptedBackend - })?; - - let (m,n) = fm.find_disk_index_by_disk_id(fm.erasure.this)?; + })?; - let disk_id = fm.erasure.this; + let (m, n) = fm.find_disk_index_by_disk_id(fm.erasure.this)?; - if m != self.endpoint.set_idx || n != self.endpoint.disk_idx { - return DiskError::InconsistentDisk; - } + let disk_id = fm.erasure.this; - format_info.id = Some(disk_id); - format_info.file_info= Some(file_meta); - format_info.data = b; - format_info.last_check = Some(OffsetDateTime::now_utc()); -; + match (self.endpoint.set_idx, self.endpoint.disk_idx) { + (Some(set_idx), Some(disk_idx)) => { + if m != set_idx || n != disk_idx { + return Err(Error::new(DiskError::InconsistentDisk)); + } + } + _ => return Err(Error::new(DiskError::InconsistentDisk)), + } -Some(disk_id) + format_info.id = Some(disk_id); + format_info.file_info = Some(file_meta); + format_info.data = b; + format_info.last_check = Some(OffsetDateTime::now_utc()); + + Ok(Some(disk_id)) // TODO: 判断源文件id,是否有效 } diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index d2f63c4f9..487edd405 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -47,7 +47,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { fn is_local(&self) -> bool; fn path(&self) -> PathBuf; async fn close(&self) -> Result<()>; - async fn get_disk_id(&self) -> Option; + async fn get_disk_id(&self) -> Result>; async fn set_disk_id(&self, id: Option) -> Result<()>; async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>; diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index b0d5a8a0c..d9e47a908 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -105,8 +105,8 @@ impl DiskAPI for RemoteDisk { self.root.clone() } - async fn get_disk_id(&self) -> Option { - self.id.lock().await.clone() + async fn get_disk_id(&self) -> Result> { + Ok(self.id.lock().await.clone()) } async fn set_disk_id(&self, id: Option) -> Result<()> { let mut lock = self.id.lock().await; diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index 8698b2103..c9e049d56 100644 --- a/ecstore/src/sets.rs +++ b/ecstore/src/sets.rs @@ -78,7 +78,7 @@ impl Sets { disk = local_disk; } - if let Some(_disk_id) = disk.as_ref().unwrap().get_disk_id().await { + if let Some(_disk_id) = disk.as_ref().unwrap().get_disk_id().await? { set_drive.push(disk); } else { warn!("sets new set_drive {}-{} get_disk_id is none", i, j); diff --git a/ecstore/src/utils/fs.rs b/ecstore/src/utils/fs.rs new file mode 100644 index 000000000..730047855 --- /dev/null +++ b/ecstore/src/utils/fs.rs @@ -0,0 +1,24 @@ +use std::{fs::Metadata, os::unix::fs::MetadataExt}; + +pub fn same_file(f1: &Metadata, f2: &Metadata) -> bool { + if f1.dev() != f2.dev() { + return false; + } + + if f1.ino() != f2.ino() { + return false; + } + + if f1.size() != f2.size() { + return false; + } + if f1.permissions() != f2.permissions() { + return false; + } + + if f1.mtime() != f2.mtime() { + return false; + } + + true +} diff --git a/ecstore/src/utils/mod.rs b/ecstore/src/utils/mod.rs index 2b78d9e75..90e120d08 100644 --- a/ecstore/src/utils/mod.rs +++ b/ecstore/src/utils/mod.rs @@ -1,5 +1,6 @@ pub mod crypto; pub mod ellipses; +pub mod fs; pub mod hash; pub mod net; pub mod path; From c71e86b8d1eaaf688743403576801d7f657950ad Mon Sep 17 00:00:00 2001 From: weisd Date: Thu, 19 Sep 2024 13:12:32 +0800 Subject: [PATCH 4/6] fix: use option writer when disk is none --- ecstore/src/erasure.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index 13cb5af18..a2261d97e 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -46,7 +46,7 @@ impl Erasure { pub async fn encode( &self, body: S, - writers: &mut [FileWriter], + writers: &mut [Option], // block_size: usize, total_size: usize, _write_quorum: usize, @@ -84,7 +84,10 @@ impl Erasure { let mut errs = Vec::new(); for (i, w) in writers.iter_mut().enumerate() { - match w.write(blocks[i].as_ref()).await { + if w.is_none() { + continue; + } + match w.as_mut().unwrap().write(blocks[i].as_ref()).await { Ok(_) => errs.push(None), Err(e) => errs.push(Some(e)), } From bc9868f926782746be8cfc47dec2a78d5c564e35 Mon Sep 17 00:00:00 2001 From: weisd Date: Thu, 19 Sep 2024 18:06:44 +0800 Subject: [PATCH 5/6] add: monitor_and_connect_endpoints --- ecstore/src/disk/local.rs | 15 ++++- ecstore/src/disk/mod.rs | 14 +++++ ecstore/src/disk/remote.rs | 31 ++++++++--- ecstore/src/sets.rs | 75 ++++++++++++++++++++----- ecstore/src/store.rs | 110 +++++++++++++++++-------------------- 5 files changed, 163 insertions(+), 82 deletions(-) diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index e0a1679ac..816ed6aa3 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -1,7 +1,7 @@ use super::{endpoint::Endpoint, error::DiskError, format::FormatV3}; use super::{ - DeleteOptions, DiskAPI, FileInfoVersions, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, - ReadOptions, RenameDataResp, VolumeInfo, WalkDirOptions, + DeleteOptions, DiskAPI, DiskLocation, FileInfoVersions, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, + ReadMultipleResp, ReadOptions, RenameDataResp, VolumeInfo, WalkDirOptions, }; use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE}; use crate::{ @@ -452,6 +452,9 @@ impl DiskAPI for LocalDisk { fn is_local(&self) -> bool { true } + async fn is_online(&self) -> bool { + true + } async fn close(&self) -> Result<()> { Ok(()) } @@ -459,6 +462,14 @@ impl DiskAPI for LocalDisk { self.root.clone() } + fn get_location(&self) -> DiskLocation { + DiskLocation { + pool_idx: self.endpoint.pool_idx, + set_idx: self.endpoint.set_idx, + disk_idx: self.endpoint.pool_idx, + } + } + async fn get_disk_id(&self) -> Result> { warn!("local get_disk_id"); // TODO: check format file diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 487edd405..8c089c8ed 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -45,10 +45,12 @@ pub async fn new_disk(ep: &endpoint::Endpoint, opt: &DiskOption) -> Result bool; + async fn is_online(&self) -> bool; fn path(&self) -> PathBuf; async fn close(&self) -> Result<()>; async fn get_disk_id(&self) -> Result>; async fn set_disk_id(&self, id: Option) -> Result<()>; + fn get_location(&self) -> DiskLocation; async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>; async fn read_all(&self, volume: &str, path: &str) -> Result; @@ -95,6 +97,18 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { async fn read_multiple(&self, req: ReadMultipleReq) -> Result>; } +pub struct DiskLocation { + pub pool_idx: Option, + pub set_idx: Option, + pub disk_idx: Option, +} + +impl DiskLocation { + pub fn valid(&self) -> bool { + self.pool_idx.is_some() && self.set_idx.is_some() && self.disk_idx.is_some() + } +} + #[derive(Debug, Default, Clone, Serialize, Deserialize)] pub struct FileInfoVersions { // Name of the volume. diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index d9e47a908..f8f4eff7e 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -28,9 +28,9 @@ use crate::{ }; use super::{ - endpoint::Endpoint, DeleteOptions, DiskAPI, DiskOption, FileInfoVersions, FileReader, FileWriter, MetaCacheEntry, - ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter, RenameDataResp, VolumeInfo, - WalkDirOptions, + endpoint::Endpoint, DeleteOptions, DiskAPI, DiskLocation, DiskOption, FileInfoVersions, FileReader, FileWriter, + MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter, RenameDataResp, + VolumeInfo, WalkDirOptions, }; #[derive(Debug)] @@ -39,6 +39,7 @@ pub struct RemoteDisk { channel: Arc>>, url: url::Url, pub root: PathBuf, + endpoint: Endpoint, } impl RemoteDisk { @@ -50,6 +51,7 @@ impl RemoteDisk { url: ep.url.clone(), root, id: Mutex::new(None), + endpoint: ep.clone(), }) } @@ -98,6 +100,13 @@ impl DiskAPI for RemoteDisk { fn is_local(&self) -> bool { false } + async fn is_online(&self) -> bool { + // TODO: 连接状态 + if let Ok(_) = self.get_client_v2().await { + return true; + } + false + } async fn close(&self) -> Result<()> { Ok(()) } @@ -105,6 +114,14 @@ impl DiskAPI for RemoteDisk { self.root.clone() } + fn get_location(&self) -> DiskLocation { + DiskLocation { + pool_idx: self.endpoint.pool_idx, + set_idx: self.endpoint.set_idx, + disk_idx: self.endpoint.pool_idx, + } + } + async fn get_disk_id(&self) -> Result> { Ok(self.id.lock().await.clone()) } @@ -253,11 +270,11 @@ impl DiskAPI for RemoteDisk { }); let response = client.walk_dir(request).await?.into_inner(); - + if !response.success { return Err(Error::from_string(response.error_info.unwrap_or("".to_string()))); } - + let entries = response .meta_cache_entry .into_iter() @@ -288,7 +305,7 @@ impl DiskAPI for RemoteDisk { }); let response = client.rename_data(request).await?.into_inner(); - + if !response.success { return Err(Error::from_string(response.error_info.unwrap_or("".to_string()))); } @@ -363,7 +380,7 @@ impl DiskAPI for RemoteDisk { }); let response = client.stat_volume(request).await?.into_inner(); - + if !response.success { return Err(Error::from_string(response.error_info.unwrap_or("".to_string()))); } diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index c9e049d56..dc6dd3dda 100644 --- a/ecstore/src/sets.rs +++ b/ecstore/src/sets.rs @@ -1,9 +1,4 @@ -use std::collections::HashMap; - -use futures::future::join_all; -use http::HeaderMap; -use tracing::warn; -use uuid::Uuid; +use std::{collections::HashMap, sync::Arc, time::Duration}; use crate::{ disk::{ @@ -21,13 +16,19 @@ use crate::{ }, utils::hash, }; +use futures::future::join_all; +use http::HeaderMap; +use tokio::sync::RwLock; +use tokio_util::sync::CancellationToken; +use tracing::{debug, warn}; +use uuid::Uuid; #[derive(Debug)] pub struct Sets { pub id: Uuid, // pub sets: Vec, // pub disk_set: Vec>>, // [set_count_idx][set_drive_count_idx] = disk_idx - pub disk_set: Vec, // [set_count_idx][set_drive_count_idx] = disk_idx + pub disk_set: Vec>, // [set_count_idx][set_drive_count_idx] = disk_idx pub pool_idx: usize, pub endpoints: PoolEndpoints, pub format: FormatV3, @@ -35,6 +36,7 @@ pub struct Sets { pub set_count: usize, pub set_drive_count: usize, pub distribution_algo: DistributionAlgoVersion, + ctx: CancellationToken, } impl Sets { @@ -44,7 +46,7 @@ impl Sets { fm: &FormatV3, pool_idx: usize, partiy_count: usize, - ) -> Result { + ) -> Result> { let set_count = fm.erasure.sets.len(); let set_drive_count = fm.erasure.sets[0].len(); @@ -52,9 +54,14 @@ impl Sets { for i in 0..set_count { let mut set_drive = Vec::with_capacity(set_drive_count); + let mut set_endpoints = Vec::with_capacity(set_drive_count); for j in 0..set_drive_count { let idx = i * set_drive_count + j; let mut disk = disks[idx].clone(); + + let endpoint = endpoints.endpoints.as_ref().get(idx).cloned(); + set_endpoints.push(endpoint); + if disk.is_none() { warn!("sets new set_drive {}-{} is none", i, j); set_drive.push(None); @@ -89,17 +96,18 @@ impl Sets { warn!("sets new set_drive {:?}", &set_drive); let set_disks = SetDisks { - disks: set_drive, + disks: RwLock::new(set_drive), set_drive_count, parity_count: partiy_count, set_index: i, pool_index: pool_idx, + set_endpoints, }; - disk_set.push(set_disks); + disk_set.push(Arc::new(set_disks)); } - let sets = Self { + let sets = Arc::new(Self { id: fm.id, // sets: todo!(), disk_set, @@ -110,15 +118,54 @@ impl Sets { set_count, set_drive_count, distribution_algo: fm.erasure.distribution_algo.clone(), - }; + ctx: CancellationToken::new(), + }); + + let asets = sets.clone(); + + tokio::spawn(async move { asets.monitor_and_connect_endpoints().await }); Ok(sets) } - pub fn get_disks(&self, set_idx: usize) -> SetDisks { + + pub async fn monitor_and_connect_endpoints(&self) { + tokio::time::sleep(Duration::from_secs(5)).await; + + self.connect_disks().await; + + let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(15 * 3)); + let cloned_token = self.ctx.clone(); + loop { + tokio::select! { + _= interval.tick()=>{ + debug!("tick..."); + self.connect_disks().await; + + interval.reset(); + }, + + _ = cloned_token.cancelled() => { + warn!("ctx cancelled"); + break; + } + } + } + + warn!("monitor_and_connect_endpoints exit"); + } + + async fn connect_disks(&self) { + debug!("start connect_disks ..."); + for set in self.disk_set.iter() { + set.connect_disks().await; + } + } + + pub fn get_disks(&self, set_idx: usize) -> Arc { self.disk_set[set_idx].clone() } - pub fn get_disks_by_key(&self, key: &str) -> SetDisks { + pub fn get_disks_by_key(&self, key: &str) -> Arc { self.get_disks(self.get_hashed_set_index(key)) } diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 44473c3b0..d3a510595 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -1,8 +1,6 @@ use crate::{ bucket_meta::BucketMetadata, - disk::{ - error::DiskError, new_disk, DeleteOptions, DiskOption, DiskStore, WalkDirOptions, BUCKET_META_PREFIX, RUSTFS_META_BUCKET, - }, + disk::{error::DiskError, new_disk, DiskOption, DiskStore, WalkDirOptions, BUCKET_META_PREFIX, RUSTFS_META_BUCKET}, endpoints::{EndpointServerPools, SetupType}, error::{Error, Result}, peer::S3PeerSys, @@ -25,7 +23,7 @@ use std::{ }; use time::OffsetDateTime; use tokio::{fs, sync::RwLock}; -use tracing::{debug, info, warn}; +use tracing::{debug, info}; use uuid::Uuid; use lazy_static::lazy_static; @@ -150,7 +148,7 @@ pub struct ECStore { pub id: uuid::Uuid, // pub disks: Vec, pub disk_map: HashMap>>, - pub pools: Vec, + pub pools: Vec>, pub peer_sys: S3PeerSys, // pub local_disks: Vec, } @@ -223,7 +221,6 @@ impl ECStore { } let sets = Sets::new(disks.clone(), pool_eps, &fm, i, partiy_count).await?; - pools.push(sets); disk_map.insert(i, disks); @@ -285,62 +282,54 @@ impl ECStore { for sets in self.pools.iter() { for set in sets.disk_set.iter() { - for disk in set.disks.iter() { - if disk.is_none() { - continue; - } - - let disk = disk.as_ref().unwrap(); - let opts = opts.clone(); - // let mut wr = &mut wr; - futures.push(disk.walk_dir(opts)); - // tokio::spawn(async move { disk.walk_dir(opts, wr).await }); - } + futures.push(set.walk_dir(&opts)); } } let results = join_all(futures).await; - let mut errs = Vec::new(); + // let mut errs = Vec::new(); let mut ress = Vec::new(); let mut uniq = HashSet::new(); - for res in results { - match res { - Ok(entrys) => { - for entry in entrys { - if !uniq.contains(&entry.name) { - uniq.insert(entry.name.clone()); - // TODO: 过滤 - if opts.limit > 0 && ress.len() as i32 >= opts.limit { - return Ok(ress); - } + for (disks_ress, _disks_errs) in results { + for (_i, disks_res) in disks_ress.iter().enumerate() { + if disks_res.is_none() { + // TODO handle errs + continue; + } + let entrys = disks_res.as_ref().unwrap(); - if entry.is_object() { - let fi = entry.to_fileinfo(&opts.bucket)?; - if fi.is_some() { - ress.push(fi.unwrap().into_object_info(&opts.bucket, &entry.name, false)); - } - continue; - } + for entry in entrys { + if !uniq.contains(&entry.name) { + uniq.insert(entry.name.clone()); + // TODO: 过滤 + if opts.limit > 0 && ress.len() as i32 >= opts.limit { + return Ok(ress); + } - if entry.is_dir() { - ress.push(ObjectInfo { - is_dir: true, - bucket: opts.bucket.clone(), - name: entry.name, - ..Default::default() - }); + if entry.is_object() { + let fi = entry.to_fileinfo(&opts.bucket)?; + if fi.is_some() { + ress.push(fi.unwrap().into_object_info(&opts.bucket, &entry.name, false)); } + continue; + } + + if entry.is_dir() { + ress.push(ObjectInfo { + is_dir: true, + bucket: opts.bucket.clone(), + name: entry.name.clone(), + ..Default::default() + }); } } - errs.push(None); } - Err(e) => errs.push(Some(e)), } } - warn!("list_merged errs {:?}", errs); + // warn!("list_merged errs {:?}", errs); Ok(ress) } @@ -349,21 +338,24 @@ impl ECStore { let mut futures = Vec::new(); for sets in self.pools.iter() { for set in sets.disk_set.iter() { - for disk in set.disks.iter() { - if disk.is_none() { - continue; - } + futures.push(set.delete_all(bucket, prefix)); + // let disks = set.disks.read().await; + // let dd = disks.clone(); + // for disk in dd { + // if disk.is_none() { + // continue; + // } - let disk = disk.as_ref().unwrap(); - futures.push(disk.delete( - bucket, - prefix, - DeleteOptions { - recursive: true, - immediate: false, - }, - )); - } + // // let disk = disk.as_ref().unwrap().clone(); + // // futures.push(disk.delete( + // // bucket, + // // prefix, + // // DeleteOptions { + // // recursive: true, + // // immediate: false, + // // }, + // // )); + // } } } let results = join_all(futures).await; From 83590cb7425a7bfdfadcc6884eeec1fe8d9da19a Mon Sep 17 00:00:00 2001 From: weisd Date: Fri, 20 Sep 2024 16:45:17 +0800 Subject: [PATCH 6/6] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E5=AE=9A?= =?UTF-8?q?=E6=97=B6=E6=A3=80=E6=B5=8B=E9=87=8D=E8=BF=9Edisk?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ecstore/src/disk/format.rs | 49 ++++++++++++++++++++++++++++++++++++++ ecstore/src/disk/mod.rs | 1 + ecstore/src/sets.rs | 2 ++ ecstore/src/store_init.rs | 22 ++++++++--------- 4 files changed, 63 insertions(+), 11 deletions(-) diff --git a/ecstore/src/disk/format.rs b/ecstore/src/disk/format.rs index 0c37e0862..dcae122be 100644 --- a/ecstore/src/disk/format.rs +++ b/ecstore/src/disk/format.rs @@ -182,6 +182,55 @@ impl FormatV3 { Err(Error::msg(format!("disk id not found {}", disk_id))) } + + pub fn check_other(&self, other: &FormatV3) -> Result<()> { + let mut tmp = other.clone(); + let this = tmp.erasure.this; + tmp.erasure.this = Uuid::nil(); + + if self.erasure.sets.len() != other.erasure.sets.len() { + return Err(Error::from_string(format!( + "Expected number of sets {}, got {}", + self.erasure.sets.len(), + other.erasure.sets.len() + ))); + } + + for i in 0..self.erasure.sets.len() { + if self.erasure.sets[i].len() != other.erasure.sets[i].len() { + return Err(Error::from_string(format!( + "Each set should be of same size, expected {}, got {}", + self.erasure.sets[i].len(), + other.erasure.sets[i].len() + ))); + } + + for j in 0..self.erasure.sets[i].len() { + if self.erasure.sets[i][j] != self.erasure.sets[i][j] { + return Err(Error::from_string(format!( + "UUID on positions {}:{} do not match with, expected {:?} got {:?}: (%w)", + i, + j, + self.erasure.sets[i][j].to_string(), + other.erasure.sets[i][j].to_string(), + ))); + } + } + } + + for i in 0..tmp.erasure.sets.len() { + for j in 0..tmp.erasure.sets[i].len() { + if this == tmp.erasure.sets[i][j] { + return Ok(()); + } + } + } + + Err(Error::msg(format!( + "DriveID {:?} not found in any drive sets {:?}", + this, other.erasure.sets + ))) + } } #[cfg(test)] diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 8c089c8ed..e5d88fc34 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -222,6 +222,7 @@ impl MetaCacheEntry { } } +#[derive(Debug, Default)] pub struct DiskOption { pub cleanup: bool, pub health_check: bool, diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index dc6dd3dda..1da14a235 100644 --- a/ecstore/src/sets.rs +++ b/ecstore/src/sets.rs @@ -102,6 +102,7 @@ impl Sets { set_index: i, pool_index: pool_idx, set_endpoints, + format: fm.clone(), }; disk_set.push(Arc::new(set_disks)); @@ -159,6 +160,7 @@ impl Sets { for set in self.disk_set.iter() { set.connect_disks().await; } + debug!("done connect_disks ..."); } pub fn get_disks(&self, set_idx: usize) -> Arc { diff --git a/ecstore/src/store_init.rs b/ecstore/src/store_init.rs index 91d580f31..eb37a20fd 100644 --- a/ecstore/src/store_init.rs +++ b/ecstore/src/store_init.rs @@ -202,14 +202,19 @@ pub fn default_partiy_count(drive: usize) -> usize { // load_format_erasure_all 读取所有foramt.json async fn load_format_erasure_all(disks: &[Option], heal: bool) -> (Vec>, Vec>) { let mut futures = Vec::with_capacity(disks.len()); - - for disk in disks.iter() { - futures.push(read_format_file(disk, heal)); - } - let mut datas = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len()); + for disk in disks.iter() { + if disk.is_none() { + datas.push(None); + errors.push(Some(Error::new(DiskError::DiskNotFound))); + } + + let disk = disk.as_ref().unwrap(); + futures.push(load_format_erasure(disk, heal)); + } + let results = join_all(futures).await; let mut i = 0; for result in results { @@ -234,12 +239,7 @@ async fn load_format_erasure_all(disks: &[Option], heal: bool) -> (Ve (datas, errors) } -async fn read_format_file(disk: &Option, _heal: bool) -> Result { - if disk.is_none() { - return Err(Error::new(DiskError::DiskNotFound)); - } - let disk = disk.as_ref().unwrap(); - +pub async fn load_format_erasure(disk: &DiskStore, _heal: bool) -> Result { let data = disk .read_all(RUSTFS_META_BUCKET, FORMAT_CONFIG_FILE) .await