diff --git a/Cargo.lock b/Cargo.lock index c908b52d8..aba59a7fe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -327,6 +327,12 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" +[[package]] +name = "cfg_aliases" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" + [[package]] name = "chrono" version = "0.4.38" @@ -484,6 +490,7 @@ dependencies = [ "lazy_static", "lock", "netif", + "nix", "num_cpus", "openssl", "path-absolutize", @@ -1105,6 +1112,18 @@ dependencies = [ "winapi", ] +[[package]] +name = "nix" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71e2746dc3a24dd78b3cfcb7be93368c6de9963d30f43a6a73998a9cf4b17b46" +dependencies = [ + "bitflags 2.6.0", + "cfg-if", + "cfg_aliases", + "libc", +] + [[package]] name = "nom" version = "7.1.3" diff --git a/ecstore/Cargo.toml b/ecstore/Cargo.toml index 9a6a3f6c7..89eef4670 100644 --- a/ecstore/Cargo.toml +++ b/ecstore/Cargo.toml @@ -31,6 +31,7 @@ lazy_static.workspace = true lock.workspace = true regex = "1.11.0" netif = "0.1.6" +nix = { version = "0.29.0", features = ["fs"] } path-absolutize = "3.1.1" protos.workspace = true rmp-serde = "1.3.0" diff --git a/ecstore/src/cache/cache.rs b/ecstore/src/cache/cache.rs new file mode 100644 index 000000000..b5e0059b0 --- /dev/null +++ b/ecstore/src/cache/cache.rs @@ -0,0 +1,119 @@ +use std::{ + fmt::Debug, + ptr, + sync::{ + atomic::{AtomicPtr, AtomicU64, Ordering}, + Arc, + }, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use serde::{Deserialize, Serialize}; +use tokio::{spawn, sync::Mutex}; + +use crate::error::Result; + +type UpdateFn = Arc Result + Send + Sync>; + +#[derive(Clone, Debug, Default)] +pub struct Opts { + return_last_good: bool, + no_wait: bool, +} + +#[derive(Deserialize, Serialize)] +pub struct Cache { + update_fn: UpdateFn, + ttl: Duration, + opts: Opts, + val: AtomicPtr, + last_update_ms: AtomicU64, + updating: Arc>, +} + +impl Cache { + pub fn new(update_fn: UpdateFn, ttl: Duration, opts: Opts) -> Self { + let val = AtomicPtr::new(ptr::null_mut()); + Self { + update_fn, + ttl, + opts, + val, + last_update_ms: AtomicU64::new(0), + updating: Arc::new(Mutex::new(false)), + } + } + + pub async fn get(self: Arc) -> Result { + let v_ptr = self.val.load(Ordering::SeqCst); + let v = if v_ptr.is_null() { + None + } else { + Some(unsafe { (*v_ptr).clone() }) + }; + + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("Time went backwards") + .as_secs(); + if v.is_some() && now - self.last_update_ms.load(Ordering::SeqCst) < self.ttl.as_secs() { + return Ok(v.unwrap()); + } + + if self.opts.no_wait && v.is_some() && now - self.last_update_ms.load(Ordering::SeqCst) < self.ttl.as_secs() * 2 { + if self.updating.try_lock().is_ok() { + let this = Arc::clone(&self); + spawn(async move { + let _ = this.update().await; + }); + } + + return Ok(v.unwrap()); + } + + let _ = self.updating.lock().await; + + if let Ok(duration) = + SystemTime::now().duration_since(UNIX_EPOCH + Duration::from_secs(self.last_update_ms.load(Ordering::SeqCst))) + { + if duration < self.ttl { + return Ok(v.unwrap()); + } + } + + match self.update().await { + Ok(_) => { + let v_ptr = self.val.load(Ordering::SeqCst); + let v = if v_ptr.is_null() { + None + } else { + Some(unsafe { (*v_ptr).clone() }) + }; + Ok(v.unwrap()) + } + Err(err) => Err(err), + } + } + + async fn update(&self) -> Result<()> { + match (self.update_fn)() { + Ok(val) => { + self.val.store(Box::into_raw(Box::new(val)), Ordering::SeqCst); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("Time went backwards") + .as_secs(); + self.last_update_ms.store(now, Ordering::SeqCst); + Ok(()) + } + Err(err) => { + let v_ptr = self.val.load(Ordering::SeqCst); + if self.opts.return_last_good && !v_ptr.is_null() { + return Ok(()); + } + + return Err(err); + } + } + } +} diff --git a/ecstore/src/cache/mod.rs b/ecstore/src/cache/mod.rs new file mode 100644 index 000000000..a5c08fdb0 --- /dev/null +++ b/ecstore/src/cache/mod.rs @@ -0,0 +1 @@ +pub mod cache; diff --git a/ecstore/src/cache_value/cache.rs b/ecstore/src/cache_value/cache.rs index 231263b61..77dd104b1 100644 --- a/ecstore/src/cache_value/cache.rs +++ b/ecstore/src/cache_value/cache.rs @@ -1,14 +1,18 @@ use std::{ + fmt::Debug, ptr, - sync::{atomic::{AtomicPtr, AtomicU64, Ordering}, Arc, Mutex}, + sync::{ + atomic::{AtomicPtr, AtomicU64, Ordering}, + Arc, + }, time::{Duration, SystemTime, UNIX_EPOCH}, }; -use tokio::spawn; +use tokio::{spawn, sync::Mutex}; -use crate::error::{Error, Result}; +use crate::error::Result; -type UpdateFn = Box Result>; +type UpdateFn = Arc Result + Send + Sync>; #[derive(Clone, Debug, Default)] pub struct Opts { @@ -16,7 +20,7 @@ pub struct Opts { no_wait: bool, } -pub struct Cache { +pub struct Cache { update_fn: UpdateFn, ttl: Duration, opts: Opts, @@ -25,7 +29,7 @@ pub struct Cache { updating: Arc>, } -impl Cache { +impl Cache { pub fn new(update_fn: UpdateFn, ttl: Duration, opts: Opts) -> Self { let val = AtomicPtr::new(ptr::null_mut()); Self { @@ -38,9 +42,13 @@ impl Cache { } } - pub fn get(&self) -> Result { + pub async fn get(self: Arc) -> Result { let v_ptr = self.val.load(Ordering::SeqCst); - let v = if v_ptr.is_null() { None } else { Some(unsafe { *v_ptr }) }; + let v = if v_ptr.is_null() { + None + } else { + Some(unsafe { (*v_ptr).clone() }) + }; let now = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -52,37 +60,58 @@ impl Cache { if self.opts.no_wait && v.is_some() && now - self.last_update_ms.load(Ordering::SeqCst) < self.ttl.as_secs() * 2 { if self.updating.try_lock().is_ok() { - let this = self.clone(); - spawn(async { + let this = Arc::clone(&self); + spawn(async move { let _ = this.update().await; }); - self.update(); } return Ok(v.unwrap()); } - match self.updating.lock() { - Ok(_) => { - if let Ok(duration) = SystemTime::now().duration_since(UNIX_EPOCH + Duration::from_secs(self.last_update_ms.load(Ordering::SeqCst))) { - if duration < self.ttl { - return Ok(v.unwrap()); - } - } + let _ = self.updating.lock().await; - match self.update() { - - } - }, - Err(err) => { - return Err(Error::from_string(err.to_string())); + if let Ok(duration) = + SystemTime::now().duration_since(UNIX_EPOCH + Duration::from_secs(self.last_update_ms.load(Ordering::SeqCst))) + { + if duration < self.ttl { + return Ok(v.unwrap()); } } - todo!() + match self.update().await { + Ok(_) => { + let v_ptr = self.val.load(Ordering::SeqCst); + let v = if v_ptr.is_null() { + None + } else { + Some(unsafe { (*v_ptr).clone() }) + }; + Ok(v.unwrap()) + } + Err(err) => Err(err), + } } async fn update(&self) -> Result<()> { - todo!() + match (self.update_fn)() { + Ok(val) => { + self.val.store(Box::into_raw(Box::new(val)), Ordering::SeqCst); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("Time went backwards") + .as_secs(); + self.last_update_ms.store(now, Ordering::SeqCst); + Ok(()) + } + Err(err) => { + let v_ptr = self.val.load(Ordering::SeqCst); + if self.opts.return_last_good && !v_ptr.is_null() { + return Ok(()); + } + + return Err(err); + } + } } } diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 00adf0b5c..9fda27fb9 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -1,10 +1,12 @@ use super::error::{is_sys_err_io, is_sys_err_not_empty, is_sys_err_too_many_files, os_is_not_exist, os_is_permission}; +use super::os::is_root_disk; use super::{endpoint::Endpoint, error::DiskError, format::FormatV3}; use super::{ os, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskMetrics, FileInfoVersions, FileReader, FileWriter, - MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, - WalkDirOptions, + Info, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, + WalkDirOptions, BUCKET_META_PREFIX, RUSTFS_META_BUCKET, }; +use crate::cache_value::cache::Cache; use crate::disk::error::{ convert_access_error, is_sys_err_handle_invalid, is_sys_err_invalid_arg, is_sys_err_is_dir, is_sys_err_not_dir, map_err_not_exists, os_err_to_file_err, @@ -12,14 +14,19 @@ use crate::disk::error::{ use crate::disk::os::check_path_length; use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE}; use crate::error::{Error, Result}; -use crate::utils::fs::{lstat, O_APPEND, O_CREATE, O_RDONLY, O_WRONLY}; +use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold}; +use crate::heal::heal_commands::HealingTracker; +use crate::heal::heal_ops::HEALING_TRACKER_FILENAME; +use crate::utils::fs::{lstat, read_file, O_APPEND, O_CREATE, O_RDONLY, O_WRONLY}; use crate::utils::path::{clean, has_suffix, SLASH_SEPARATOR}; +use crate::utils::stat_linux::get_info; use crate::{ file_meta::FileMeta, store_api::{FileInfo, RawFileInfo}, utils, }; use path_absolutize::Absolutize; +use std::fmt::Debug; use std::{ fs::Metadata, path::{Path, PathBuf}, @@ -49,18 +56,29 @@ impl FormatInfo { } } -#[derive(Debug)] pub struct LocalDisk { pub root: PathBuf, pub format_path: PathBuf, pub format_info: RwLock, pub endpoint: Endpoint, + disk_info_cache: Cache, // pub id: Mutex>, // pub format_data: Mutex>, // pub format_file_info: Mutex>, // pub format_last_check: Mutex>, } +impl Debug for LocalDisk { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("LocalDisk") + .field("root", &self.root) + .field("format_path", &self.format_path) + .field("format_info", &self.format_info) + .field("endpoint", &self.endpoint) + .finish() + } +} + impl LocalDisk { pub async fn new(ep: &Endpoint, cleanup: bool) -> Result { let root = fs::canonicalize(ep.url.path()).await?; @@ -101,6 +119,33 @@ impl LocalDisk { last_check: format_last_check, }; + let derive_path = root.to_string_lossy().to_string(); + let disk_id = id.map_or("".to_string(), |id| id.to_string()); + let update_fn = || async move { + if let Ok((info, root)) = get_disk_info(&derive_path).await { + let mut disk_info = DiskInfo { + total: info.total, + free: info.free, + used: info.used, + used_inodes: info.files - info.ffree, + free_inodes: info.ffree, + major: info.major, + minor: info.minor, + fs_type: info.fstype, + root_disk: root, + id: disk_id, + ..Default::default() + }; + if root { + return Err(Error::new(DiskError::DriveIsRoot)); + } + + // disk_info.healing = + } else { + return Ok(DiskInfo::default()); + } + }; + // TODO: DIRECT suport // TODD: DiskInfo let disk = Self { @@ -1612,6 +1657,27 @@ impl DiskAPI for LocalDisk { } } +async fn get_disk_info(drive_path: &str) -> Result<(Info, bool)> { + check_path_length(drive_path)?; + + let disk_info = get_info(drive_path, false)?; + let root_drive = if !*GLOBAL_IsErasureSD.read().await { + let root_disk_threshold = *GLOBAL_RootDiskThreshold.read().await; + if root_disk_threshold > 0 { + disk_info.total <= root_disk_threshold + } else { + match is_root_disk(drive_path, SLASH_SEPARATOR) { + Ok(result) => result, + Err(_) => false, + } + } + } else { + false + }; + + Ok((disk_info, root_drive)) +} + #[cfg(test)] mod test { diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 9ef049e3d..9ab82343c 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -14,10 +14,7 @@ pub const FORMAT_CONFIG_FILE: &str = "format.json"; const STORAGE_FORMAT_FILE: &str = "xl.meta"; use crate::{ - erasure::{ReadAt, Write}, - error::{Error, Result}, - file_meta::FileMeta, - store_api::{FileInfo, RawFileInfo}, + erasure::{ReadAt, Write}, error::{Error, Result}, file_meta::FileMeta, heal::heal_commands::HealingTracker, store_api::{FileInfo, RawFileInfo} }; use endpoint::Endpoint; @@ -167,8 +164,8 @@ pub struct DiskInfo { pub used: u64, pub used_inodes: u64, pub free_inodes: u64, - pub major: u32, - pub minor: u32, + pub major: u64, + pub minor: u64, pub nr_requests: u64, pub fs_type: String, pub root_disk: bool, @@ -192,6 +189,21 @@ pub struct DiskMetrics { total_deletes: u64, } +#[derive(Clone, Debug, Default)] +pub struct Info { + pub total: u64, + pub free: u64, + pub used: u64, + pub files: u64, + pub ffree: u64, + pub fstype: String, + pub major: u64, + pub minor: u64, + pub name: String, + pub rotational: bool, + pub nrrequests: u64, +} + #[derive(Debug, Default, Clone, Serialize, Deserialize)] pub struct FileInfoVersions { // Name of the volume. diff --git a/ecstore/src/disk/os.rs b/ecstore/src/disk/os.rs index 367a36ccd..75f43166e 100644 --- a/ecstore/src/disk/os.rs +++ b/ecstore/src/disk/os.rs @@ -8,7 +8,7 @@ use tokio::fs; use crate::{ disk::error::{is_sys_err_not_dir, is_sys_err_path_not_found, os_is_not_exist}, error::{Error, Result}, - utils, + utils::{self, stat_linux::same_disk}, }; use super::error::{os_err_to_file_err, os_is_exist, DiskError}; @@ -51,6 +51,14 @@ pub fn check_path_length(path_name: &str) -> Result<()> { Ok(()) } +pub fn is_root_disk(disk_path: &str, root_disk: &str) -> Result { + if cfg!(target_os = "windows") { + return Ok(false); + } + + same_disk(disk_path, root_disk) +} + pub async fn make_dir_all(path: impl AsRef, base_dir: impl AsRef) -> Result<()> { check_path_length(path.as_ref().to_string_lossy().to_string().as_str())?; diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index 4b82c43cd..d66d9b947 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -20,9 +20,7 @@ use super::{ RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, }; use crate::{ - disk::error::DiskError, - error::{Error, Result}, - store_api::{FileInfo, RawFileInfo}, + disk::error::DiskError, error::{Error, Result}, heal::heal_commands::HealingTracker, store_api::{FileInfo, RawFileInfo} }; use protos::proto_gen::node_service::RenamePartRequst; diff --git a/ecstore/src/global.rs b/ecstore/src/global.rs index a5735544e..30c94bdb7 100644 --- a/ecstore/src/global.rs +++ b/ecstore/src/global.rs @@ -13,10 +13,11 @@ lazy_static! { pub static ref GLOBAL_LOCAL_DISK: Arc>>> = Arc::new(RwLock::new(Vec::new())); static ref GLOBAL_IsErasure: RwLock = RwLock::new(false); static ref GLOBAL_IsDistErasure: RwLock = RwLock::new(false); - static ref GLOBAL_IsErasureSD: RwLock = RwLock::new(false); + pub static ref GLOBAL_IsErasureSD: RwLock = RwLock::new(false); pub static ref GLOBAL_LOCAL_DISK_MAP: Arc>>> = Arc::new(RwLock::new(HashMap::new())); pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc> = Arc::new(RwLock::new(Vec::new())); pub static ref GLOBAL_Endpoints: RwLock = RwLock::new(EndpointServerPools(Vec::new())); + pub static ref GLOBAL_RootDiskThreshold: RwLock = RwLock::new(0); } pub async fn set_global_endpoints(eps: Vec) { diff --git a/ecstore/src/heal/heal_commands.rs b/ecstore/src/heal/heal_commands.rs index e6cf29b6b..827543690 100644 --- a/ecstore/src/heal/heal_commands.rs +++ b/ecstore/src/heal/heal_commands.rs @@ -1,3 +1,13 @@ +use std::{path::Path, time::{SystemTime, UNIX_EPOCH}}; + +use serde::{Deserialize, Serialize}; +use tokio::sync::RwLock; + +use crate::{ + disk::{DiskStore, BUCKET_META_PREFIX, RUSTFS_META_BUCKET}, + error::{Error, Result}, heal::heal_ops::HEALING_TRACKER_FILENAME, new_object_layer_fn, store_api::StorageAPI, utils::fs::read_file, +}; + pub type HealScanMode = usize; pub type HealItemType = String; @@ -37,3 +47,230 @@ pub struct HealResultItem { pub after: Vec, pub object_size: usize, } + +#[derive(Debug, Default, Deserialize, Serialize)] +pub struct HealingTracker { + #[serde(skip_serializing, skip_deserializing)] + disk: Option, + + id: String, + pool_index: Option, + set_index: Option, + disk_index: Option, + path: String, + endpoint: String, + started: u64, + last_update: u64, + + objects_total_count: u64, + objects_total_size: u64, + + items_healed: u64, + items_failed: u64, + + bytes_done: u64, + bytes_failed: u64, + + bucket: String, + object: String, + + resume_items_healed: u64, + resume_items_failed: u64, + resume_items_skipped: u64, + resume_bytes_done: u64, + resume_bytes_failed: u64, + resume_bytes_skipped: u64, + + queue_buckets: Vec, + + healed_buckets: Vec, + + heal_id: String, + + item_skipped: u64, + bytes_skipped: u64, + + retry_attempts: u64, + + finished: bool, + + #[serde(skip_serializing, skip_deserializing)] + mu: RwLock, +} + +impl HealingTracker { + pub fn marshal_msg(&self) -> Result> { + serde_json::to_string(self) + .map(|s| s.as_bytes().to_vec()) + .map_err(|err| Error::from_string(err.to_string())) + } + + pub fn unmarshal_msg(data: &[u8]) -> Result { + serde_json::from_slice::(data) + .map_err(|err| Error::from_string(err.to_string())) + } + + pub async fn reset_healing(&mut self) { + self.mu.write().await; + self.items_healed = 0; + self.items_failed = 0; + self.bytes_done = 0; + self.bytes_failed = 0; + self.resume_items_healed = 0; + self.resume_items_failed = 0; + self.resume_bytes_done = 0; + self.resume_bytes_failed = 0; + self.item_skipped = 0; + self.bytes_skipped = 0; + + self.healed_buckets = Vec::new(); + self.bucket = String::new(); + self.object = String::new(); + } + + pub async fn get_last_update(&self) -> u64 { + self.mu.read().await; + + self.last_update + } + + pub async fn get_bucket(&self) -> String { + self.mu.read().await; + + self.bucket.clone() + } + + pub async fn set_bucket(&mut self, bucket: &str) { + self.mu.write().await; + + self.bucket = bucket.to_string(); + } + + pub async fn get_object(&self) -> String { + self.mu.read().await; + + self.object.clone() + } + + pub async fn set_object(&mut self, object: &str) { + self.mu.write().await; + + self.object = object.to_string(); + } + + pub async fn update_progress(&mut self, success: bool, skipped: bool, by: u64) { + self.mu.write().await; + + if success { + self.items_healed += 1; + self.bytes_done += by; + } else if skipped { + self.item_skipped += 1; + self.bytes_skipped += by; + } else { + self.items_failed += 1; + self.bytes_failed += by; + } + } + + pub async fn update(&mut self) -> Result<()> { + if let Some(disk) = &self.disk { + if healing(&disk.path().to_string_lossy().to_string()).await?.is_none() { + return Err(Error::from_string(format!("healingTracker: drive {} is not marked as healing", self.id))); + } + self.mu.write().await; + if self.id.is_empty() || self.pool_index.is_none() || self.set_index.is_none() || self.disk_index.is_none() { + self.id = disk.get_disk_id().await?.map_or("".to_string(), |id| id.to_string()); + let disk_location = disk.get_disk_location(); + self.pool_index = disk_location.pool_idx; + self.set_index = disk_location.set_idx; + self.disk_index = disk_location.disk_idx; + } + } + + self.save().await + } + + pub async fn save(&mut self) -> Result<()> { + self.mu.write().await; + if self.pool_index.is_none() || self.set_index.is_none() || self.disk_index.is_none() { + let layer = new_object_layer_fn(); + let lock = layer.read().await; + let store = match lock.as_ref() { + Some(s) => s, + None => return Err(Error::from_string("Not init".to_string())), + }; + (self.pool_index, self.set_index, self.disk_index) = store.get_pool_and_set(&self.id).await?; + } + + self.last_update = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("Time went backwards") + .as_secs(); + + let htracker_bytes = self.marshal_msg()?; + + if let Some(disk) = &self.disk { + let file_path = Path::new(BUCKET_META_PREFIX).join(HEALING_TRACKER_FILENAME); + return disk.write_all(RUSTFS_META_BUCKET, file_path.to_str().unwrap(), htracker_bytes).await; + } + Ok(()) + } +} + +async fn load_healing_tracker(disk: &Option) -> Result { + if let Some(disk) = disk { + let disk_id = disk.get_disk_id().await?; + if let Some(disk_id) = disk_id { + let disk_id = disk_id.to_string(); + let file_path = Path::new(BUCKET_META_PREFIX).join(HEALING_TRACKER_FILENAME); + let data = disk.read_all(RUSTFS_META_BUCKET, file_path.to_str().unwrap()).await?; + let mut healing_tracker = HealingTracker::unmarshal_msg(&data)?; + if healing_tracker.id != disk_id && !healing_tracker.id.is_empty() { + return Err(Error::from_string(format!("loadHealingTracker: drive id mismatch expected {}, got {}", healing_tracker.id, disk_id))); + } + healing_tracker.id = disk_id; + return Ok(healing_tracker); + } else { + return Err(Error::from_string("loadHealingTracker: disk not have id")); + } + + } else { + return Err(Error::from_string("loadHealingTracker: nil drive given")); + } +} + +async fn init_healing_tracker(disk: DiskStore, heal_id: String) -> Result { + let mut healing_tracker = HealingTracker::default(); + healing_tracker.id = disk.get_disk_id().await?.map_or("".to_string(), |id| id.to_string()); + healing_tracker.heal_id = heal_id; + healing_tracker.path = disk.to_string(); + healing_tracker.endpoint = disk.endpoint().to_string(); + healing_tracker.started = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("Time went backwards") + .as_secs(); + let disk_location = disk.get_disk_location(); + healing_tracker.pool_index = disk_location.pool_idx; + healing_tracker.set_index = disk_location.set_idx; + healing_tracker.disk_index = disk_location.disk_idx; + healing_tracker.disk = Some(disk); + + Ok(healing_tracker) +} + +pub async fn healing(derive_path: &str) -> Result> { + let healing_file = Path::new(derive_path) + .join(RUSTFS_META_BUCKET) + .join(BUCKET_META_PREFIX) + .join(HEALING_TRACKER_FILENAME); + + let b = read_file(healing_file).await?; + if b.is_empty() { + return Ok(None); + } + + let healing_tracker = HealingTracker::unmarshal_msg(&b)?; + + Ok(Some(healing_tracker)) +} diff --git a/ecstore/src/heal/heal_ops.rs b/ecstore/src/heal/heal_ops.rs index 05877e634..67a2d5b4a 100644 --- a/ecstore/src/heal/heal_ops.rs +++ b/ecstore/src/heal/heal_ops.rs @@ -15,8 +15,9 @@ pub type HealEntryFn = Box Resul lazy_static! { static ref HEAL_NOT_STARTED_STATUS: HealStatusSummary = String::from("not started"); - static ref bgHealingUUID: String = String::from("0000-0000-0000-0000"); } +pub const BG_HEALING_UUID: &str = "0000-0000-0000-0000"; +pub const HEALING_TRACKER_FILENAME: &str = ".healing.bin"; lazy_static! {} @@ -119,7 +120,7 @@ impl HealSequence { } fn has_ended(&self) -> bool { - if self.client_token == bgHealingUUID.to_string() { + if self.client_token == BG_HEALING_UUID.to_string() { return false; } @@ -139,7 +140,7 @@ impl HealSequence { } fn heal_items(&self, buckets_only: bool) -> Result<()> { - if self.client_token == bgHealingUUID.to_string() { + if self.client_token == BG_HEALING_UUID.to_string() { return Ok(()); } diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index b767a0412..dd320e8b0 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -892,6 +892,20 @@ impl StorageAPI for ECStore { todo!() } + async fn get_pool_and_set(&self, id: &str) -> Result<(Option, Option, Option)> { + for (pool_idx, pool) in self.pools.iter().enumerate(){ + for (set_idx, set) in pool.format.erasure.sets.iter().enumerate() { + for (disk_idx, disk_id) in set.iter().enumerate() { + if disk_id.to_string() == id { + return Ok((Some(pool_idx), Some(set_idx), Some(disk_idx))); + } + } + } + } + + Err(Error::new(DiskError::DiskNotFound)) + } + 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() { diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 514a7742f..1df9741d6 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -611,5 +611,6 @@ pub trait StorageAPI: ObjectIO { async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result; async fn heal_object(&self, bucket: &str, object: &str, version_id: &str, opts: &HealOpts) -> Result; async fn heal_objects(&self, bucket: &str, prefix: &str, opts: &HealOpts, func: HealObjectFn) -> Result<()>; + async fn get_pool_and_set(&self, id: &str) -> Result<(Option, Option, Option)>; async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()>; } diff --git a/ecstore/src/utils/mod.rs b/ecstore/src/utils/mod.rs index b5ca93bd3..bf3b6e4fd 100644 --- a/ecstore/src/utils/mod.rs +++ b/ecstore/src/utils/mod.rs @@ -4,4 +4,5 @@ pub mod fs; pub mod hash; pub mod net; pub mod path; +pub mod stat_linux; pub mod wildcard; diff --git a/ecstore/src/utils/stat_linux.rs b/ecstore/src/utils/stat_linux.rs new file mode 100644 index 000000000..254c71e6b --- /dev/null +++ b/ecstore/src/utils/stat_linux.rs @@ -0,0 +1,84 @@ +use nix::sys::{ + stat::{major, minor, stat}, + statfs::{statfs, FsType}, +}; + +use crate::{ + disk::Info, + error::{Error, Result}, +}; + +use lazy_static::lazy_static; +use std::collections::HashMap; + +lazy_static! { + static ref FS_TYPE_TO_STRING_MAP: HashMap<&'static str, &'static str> = { + let mut m = HashMap::new(); + m.insert("1021994", "TMPFS"); + m.insert("137d", "EXT"); + m.insert("4244", "HFS"); + m.insert("4d44", "MSDOS"); + m.insert("52654973", "REISERFS"); + m.insert("5346544e", "NTFS"); + m.insert("58465342", "XFS"); + m.insert("61756673", "AUFS"); + m.insert("6969", "NFS"); + m.insert("ef51", "EXT2OLD"); + m.insert("ef53", "EXT4"); + m.insert("f15f", "ecryptfs"); + m.insert("794c7630", "overlayfs"); + m.insert("2fc12fc1", "zfs"); + m.insert("ff534d42", "cifs"); + m.insert("53464846", "wslfs"); + m + }; +} + +fn get_fs_type(ftype: FsType) -> String { + let binding = format!("{:?}", ftype); + let fs_type_hex = binding.as_str(); + match FS_TYPE_TO_STRING_MAP.get(fs_type_hex) { + Some(fs_type_string) => fs_type_string.to_string(), + None => "UNKNOWN".to_string(), + } +} + +pub fn get_info(path: &str, first_time: bool) -> Result { + let statfs = statfs(path)?; + let reserved_blocks = statfs.blocks_free() - statfs.blocks_available(); + let mut info = Info { + total: statfs.block_size() as u64 * (statfs.blocks() - reserved_blocks), + free: statfs.blocks() as u64 * statfs.blocks_available(), + files: statfs.files(), + ffree: statfs.files_free(), + fstype: get_fs_type(statfs.filesystem_type()), + ..Default::default() + }; + + let stat = stat(path)?; + let dev_id = stat.st_dev as u64; + info.major = major(dev_id); + info.minor = minor(dev_id); + + if info.free > info.total { + return Err(Error::from_string(format!( + "detected free space {} > total drive space {}, fs corruption at {}. please run 'fsck'", + info.free, info.total, path + ))); + } + + info.used = info.total - info.free; + + if first_time { + // todo + } + + Ok(info) +} + +pub fn same_disk(disk1: &str, disk2: &str) -> Result { + let stat1 = stat(disk1)?; + let stat2 = stat(disk2)?; + + Ok(stat1.st_dev == stat2.st_dev) +}