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

heal object

Browse Source 
Signed-off-by: junxiang Mu <1948535941@qq.com>
This commit is contained in:
junxiang Mu
2024-10-21 16:38:50 +08:00
parent 3f2772c74c
commit 76fb6775d1
19 changed files with 1392 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/bucket/metadata.rs
-11
View File
@@ -19,7 +19,6 @@ use crate::config::common::{read_config, save_config};
use crate::error::{Error, Result};
use crate::disk::BUCKET_META_PREFIX;
use crate::file_meta::FileMetaShallowVersion;
use crate::store::ECStore;
pub const BUCKET_METADATA_FILE: &str = ".metadata.bin";
@@ -398,16 +397,6 @@ where
s.serialize_bytes(&buf)
}
#[derive(Debug, Default)]
pub struct MetadataResolutionParams {
pub dir_quorum: usize,
pub obj_quorum: usize,
pub requested_versions: usize,
pub bucket: String,
pub strict: bool,
pub candidates: Vec<Vec<FileMetaShallowVersion>>,
}
#[cfg(test)]
mod test {
ecstore/src/cache/cache.rs
Vendored
-119
View File
@@ -1,119 +0,0 @@
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<T> = Arc<dyn Fn() -> Result<T> + Send + Sync>;
#[derive(Clone, Debug, Default)]
pub struct Opts {
return_last_good: bool,
no_wait: bool,
}
#[derive(Deserialize, Serialize)]
pub struct Cache<T: Clone + Debug + Send> {
update_fn: UpdateFn<T>,
ttl: Duration,
opts: Opts,
val: AtomicPtr<T>,
last_update_ms: AtomicU64,
updating: Arc<Mutex<bool>>,
}
impl<T: Clone + Debug + Send + 'static> Cache<T> {
pub fn new(update_fn: UpdateFn<T>, 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<Self>) -> Result<T> {
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);
}
}
}
}
ecstore/src/cache/mod.rs
Vendored
-1
View File
@@ -1 +0,0 @@
pub mod cache;
ecstore/src/cache_value/cache.rs
+1 -1
View File
@@ -12,7 +12,7 @@ use tokio::{spawn, sync::Mutex};
use crate::error::Result;
type UpdateFn<T> = Arc<dyn Fn() -> Result<T> + Send + Sync>;
type UpdateFn<T> = Box<dyn Fn() -> Result<T> + Send + Sync>;
#[derive(Clone, Debug, Default)]
pub struct Opts {
ecstore/src/cache_value/metacache_set.rs
+230
View File
@@ -0,0 +1,230 @@
use std::sync::Arc;
use tokio::{
spawn,
sync::{
broadcast::Receiver as B_Receiver,
mpsc::{self},
RwLock,
},
};
use crate::{
disk::{DiskStore, MetaCacheEntries, MetaCacheEntry, WalkDirOptions},
error::{Error, Result},
};
#[derive(Default)]
pub struct ListPathRawOptions {
pub disks: Vec<Option<DiskStore>>,
pub fallback_disks: Vec<Option<DiskStore>>,
pub bucket: String,
pub path: String,
pub recursice: bool,
pub filter_prefix: String,
pub forward_to: String,
pub min_disks: usize,
pub report_not_found: bool,
pub per_disk_limit: i32,
pub agreed: Option<Arc<dyn Fn(MetaCacheEntry) + Send + Sync>>,
pub partial: Option<Arc<dyn Fn(MetaCacheEntries, &[Option<Error>]) + Send + Sync>>,
pub finished: Option<Arc<dyn Fn(&[Option<Error>]) + Send + Sync>>,
}
impl Clone for ListPathRawOptions {
fn clone(&self) -> Self {
Self {
disks: self.disks.clone(),
fallback_disks: self.fallback_disks.clone(),
bucket: self.bucket.clone(),
path: self.path.clone(),
recursice: self.recursice.clone(),
filter_prefix: self.filter_prefix.clone(),
forward_to: self.forward_to.clone(),
min_disks: self.min_disks.clone(),
report_not_found: self.report_not_found.clone(),
per_disk_limit: self.per_disk_limit.clone(),
..Default::default()
}
}
}
pub async fn list_path_raw(mut rx: B_Receiver<bool>, opts: ListPathRawOptions) -> Result<()> {
if opts.disks.is_empty() {
return Err(Error::from_string("list_path_raw: 0 drives provided"));
}
let mut readers = Vec::with_capacity(opts.disks.len());
let fds = Arc::new(RwLock::new(opts.fallback_disks.clone()));
for disk in opts.disks.iter() {
let disk = disk.clone();
let opts_clone = opts.clone();
let fds_clone = fds.clone();
let (m_tx, m_rx) = mpsc::channel::<MetaCacheEntry>(100);
readers.push(m_rx);
spawn(async move {
let mut need_fallback = false;
if disk.is_none() {
need_fallback = true;
} else {
match disk
.as_ref()
.unwrap()
.walk_dir(WalkDirOptions {
bucket: opts_clone.bucket.clone(),
base_dir: opts_clone.path.clone(),
recursive: opts_clone.recursice.clone(),
report_notfound: opts_clone.report_not_found,
filter_prefix: opts_clone.filter_prefix.clone(),
forward_to: opts_clone.forward_to.clone(),
limit: opts_clone.per_disk_limit,
..Default::default()
})
.await
{
Ok(r) => {
for v in r.iter() {
let _ = m_tx.send(v.to_owned()).await;
}
}
Err(_) => need_fallback = true,
}
}
while need_fallback {
let f_disk = loop {
let mut fds_w = fds_clone.write().await;
if fds_w.is_empty() {
break None;
}
let fd = fds_w.remove(0);
if fd.is_some() && fd.as_ref().unwrap().is_online().await {
break fd;
}
};
if f_disk.is_none() {
break;
}
match disk
.as_ref()
.unwrap()
.walk_dir(WalkDirOptions {
bucket: opts_clone.bucket.clone(),
base_dir: opts_clone.path.clone(),
recursive: opts_clone.recursice,
report_notfound: opts_clone.report_not_found,
filter_prefix: opts_clone.filter_prefix.clone(),
forward_to: opts_clone.forward_to.clone(),
limit: opts_clone.per_disk_limit,
..Default::default()
})
.await
{
Ok(r) => {
for v in r.iter() {
let _ = m_tx.send(v.to_owned()).await;
}
need_fallback = false;
}
Err(_) => break,
}
}
});
}
let errs: Vec<Option<Error>> = Vec::with_capacity(readers.len());
loop {
let mut current = MetaCacheEntry::default();
let (mut at_eof, mut has_err, mut agree) = (0, 0, 0);
if rx.try_recv().is_ok() {
return Err(Error::from_string("canceled"));
}
let mut top_entries: Vec<MetaCacheEntry> = Vec::with_capacity(readers.len());
// top_entries.clear();
for (i, r) in readers.iter_mut().enumerate() {
if errs[i].is_none() {
has_err += 1;
continue;
}
let entry = match r.recv().await {
Some(entry) => entry,
None => {
at_eof += 1;
continue;
},
};
// If no current, add it.
if current.name.is_empty() {
top_entries.insert(i, entry.clone());
current = entry;
agree += 1;
continue;
}
// If exact match, we agree.
if let Ok((_, true)) = current.matches(&entry, true) {
top_entries.insert(i, entry);
agree += 1;
continue;
}
// If only the name matches we didn't agree, but add it for resolution.
if entry.name == current.name {
top_entries.insert(i, entry);
continue;
}
// We got different entries
if entry.name > current.name {
continue;
}
// We got a new, better current.
// Clear existing entries.
top_entries.clear();
agree += 1;
top_entries.insert(i, entry.clone());
current = entry;
}
if has_err > 0 && has_err > opts.disks.len() - opts.min_disks {
if let Some(finished_fn) = opts.finished.clone() {
finished_fn(&errs);
}
let mut combined_err = Vec::new();
errs.iter().zip(opts.disks.iter()).for_each(|(err, disk)| {
match (err, disk) {
(Some(err), Some(disk)) => {
combined_err.push(format!("drive {} returned: {}", disk.to_string(), err));
},
(Some(err), None) => {
combined_err.push(err.to_string());
},
_ => {},
}
});
return Err(Error::from_string(combined_err.join(", ")));
}
// Break if all at EOF or error.
if at_eof + has_err == readers.len() {
if has_err > 0 {
if let Some(finished_fn) = opts.finished.clone() {
finished_fn(&errs);
}
break;
}
}
if agree == readers.len() {
if let Some(agreed_fn) = opts.agreed.clone() {
agreed_fn(current);
}
continue;
}
if let Some(partial_fn) = opts.partial.clone() {
partial_fn(MetaCacheEntries(top_entries), &errs);
}
}
Ok(())
}
ecstore/src/cache_value/mod.rs
+1
View File
@@ -1 +1,2 @@
pub mod cache;
pub mod metacache_set;
ecstore/src/disk/endpoint.rs
+1 -1
View File
@@ -16,7 +16,7 @@ pub enum EndpointType {
}
/// any type of endpoint.
#[derive(Debug, PartialEq, Eq, Clone)]
#[derive(Debug, PartialEq, Eq, Clone, Hash)]
pub struct Endpoint {
pub url: url::Url,
pub is_local: bool,
ecstore/src/disk/local.rs
+87 -40
View File
@@ -4,9 +4,9 @@ use super::{endpoint::Endpoint, error::DiskError, format::FormatV3};
use super::{
os, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskMetrics, FileInfoVersions, FileReader, FileWriter,
Info, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo,
WalkDirOptions, BUCKET_META_PREFIX, RUSTFS_META_BUCKET,
WalkDirOptions,
};
use crate::cache_value::cache::Cache;
use crate::cache_value::cache::{Cache, Opts};
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,
@@ -15,9 +15,7 @@ use crate::disk::os::check_path_length;
use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE};
use crate::error::{Error, Result};
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::fs::{lstat, 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::{
@@ -26,7 +24,11 @@ use crate::{
utils,
};
use path_absolutize::Absolutize;
use tokio::runtime::Runtime;
use std::fmt::Debug;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use std::time::Duration;
use std::{
fs::Metadata,
path::{Path, PathBuf},
@@ -61,7 +63,13 @@ pub struct LocalDisk {
pub format_path: PathBuf,
pub format_info: RwLock<FormatInfo>,
pub endpoint: Endpoint,
disk_info_cache: Cache<DiskInfo>,
pub disk_info_cache: Arc<Cache<DiskInfo>>,
pub scanning: AtomicU32,
pub rotational: bool,
pub fstype: String,
pub major: u64,
pub minor: u64,
pub nrrequests: u64,
// pub id: Mutex<Option<Uuid>>,
// pub format_data: Mutex<Vec<u8>>,
// pub format_file_info: Mutex<Option<Metadata>>,
@@ -118,46 +126,77 @@ impl LocalDisk {
file_info: format_meta,
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));
let root_clone = root.clone();
let disk_id = Arc::new(id.map_or("".to_string(), |id| id.to_string()));
let update_fn = move || {
let rt = Runtime::new().unwrap();
rt.block_on(async {
match get_disk_info(root.clone()).await {
Ok((info, root)) => {
let 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.to_string(),
..Default::default()
};
if root {
return Err(Error::new(DiskError::DriveIsRoot));
}
// disk_info.healing =
Ok(disk_info)
},
Err(err) => {
Err(err)
}
}
// disk_info.healing =
} else {
return Ok(DiskInfo::default());
}
})
};
let cache = Cache::new(Box::new(update_fn), Duration::from_secs(1), Opts::default());
// TODO: DIRECT suport
// TODD: DiskInfo
let disk = Self {
root,
let mut disk = Self {
root: root_clone.clone(),
endpoint: ep.clone(),
format_path,
format_info: RwLock::new(format_info),
disk_info_cache: Arc::new(cache),
scanning: AtomicU32::new(0),
rotational: Default::default(),
fstype: Default::default(),
minor: Default::default(),
major: Default::default(),
nrrequests: Default::default(),
// // format_legacy,
// format_file_info: Mutex::new(format_meta),
// format_data: Mutex::new(format_data),
// format_last_check: Mutex::new(format_last_check),
};
let (info, root) = get_disk_info(root_clone).await?;
disk.major = info.major;
disk.minor = info.minor;
disk.fstype = info.fstype;
if root {
return Err(Error::new(DiskError::DriveIsRoot));
}
if info.nrrequests > 0 {
disk.nrrequests = info.nrrequests;
}
if info.rotational {
disk.rotational = true;
}
disk.make_meta_volumes().await?;
@@ -607,6 +646,7 @@ impl LocalDisk {
Ok(f)
}
#[allow(dead_code)]
fn get_metrics(&self) -> DiskMetrics {
DiskMetrics::default()
}
@@ -1650,23 +1690,30 @@ impl DiskAPI for LocalDisk {
Ok(())
}
async fn disk_info(&self, opts: &DiskInfoOptions) -> Result<DiskInfo> {
let mut info = DiskInfo::default();
async fn disk_info(&self, _: &DiskInfoOptions) -> Result<DiskInfo> {
let mut info = Cache::get(self.disk_info_cache.clone()).await?;
// TODO: nr_requests, rotational
info.nr_requests = self.nrrequests;
info.rotational = self.rotational;
info.mount_path = self.path().to_str().unwrap().to_string();
info.endpoint = self.endpoint.to_string();
info.scanning = self.scanning.load(Ordering::SeqCst) == 1;
Ok(info)
}
}
async fn get_disk_info(drive_path: &str) -> Result<(Info, bool)> {
check_path_length(drive_path)?;
async fn get_disk_info(drive_path: PathBuf) -> Result<(Info, bool)> {
let drive_path = drive_path.to_string_lossy().to_string();
check_path_length(&drive_path)?;
let disk_info = get_info(drive_path, false)?;
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) {
match is_root_disk(&drive_path, SLASH_SEPARATOR) {
Ok(result) => result,
Err(_) => false,
}
ecstore/src/disk/mod.rs
+194 -3
View File
@@ -14,7 +14,10 @@ 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, heal::heal_commands::HealingTracker, store_api::{FileInfo, RawFileInfo}
erasure::{ReadAt, Write},
error::{Error, Result},
file_meta::{merge_file_meta_versions, FileMeta, FileMetaShallowVersion},
store_api::{FileInfo, RawFileInfo},
};
use endpoint::Endpoint;
@@ -23,7 +26,7 @@ use protos::proto_gen::node_service::{
node_service_client::NodeServiceClient, ReadAtRequest, ReadAtResponse, WriteRequest, WriteResponse,
};
use serde::{Deserialize, Serialize};
use std::{collections::HashMap, fmt::Debug, io::SeekFrom, path::PathBuf, sync::Arc, usize};
use std::{cmp::Ordering, collections::HashMap, fmt::Debug, io::SeekFrom, path::PathBuf, sync::Arc, usize};
use time::OffsetDateTime;
use tokio::{
fs::File,
@@ -247,7 +250,17 @@ pub struct WalkDirOptions {
pub disk_id: String,
}
#[derive(Debug, Default, Serialize, Deserialize)]
#[derive(Clone, Debug, Default)]
pub struct MetadataResolutionParams {
pub dir_quorum: usize,
pub obj_quorum: usize,
pub requested_versions: usize,
pub bucket: String,
pub strict: bool,
pub candidates: Vec<Vec<FileMetaShallowVersion>>,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct MetaCacheEntry {
// name is the full name of the object including prefixes
pub name: String,
@@ -335,6 +348,184 @@ impl MetaCacheEntry {
Ok(fm.into_file_info_versions(bucket, self.name.as_str(), false)?)
}
pub fn matches(&self, other: &MetaCacheEntry, strict: bool) -> Result<(Option<MetaCacheEntry>, bool)> {
let mut prefer = None;
if self.name != other.name {
if self.name < other.name {
return Ok((Some(self.clone()), false));
}
return Ok((Some(other.clone()), false));
}
if other.is_dir() || self.is_dir() {
if self.is_dir() {
return Ok((Some(self.clone()), other.is_dir()));
}
return Ok((Some(other.clone()), other.is_dir() == self.is_dir()));
}
let self_vers = match &self.cached {
Some(file_meta) => file_meta.clone(),
None => FileMeta::load(&self.metadata)?,
};
let other_vers = match &other.cached {
Some(file_meta) => file_meta.clone(),
None => FileMeta::load(&other.metadata)?,
};
if self_vers.versions.len() != other_vers.versions.len() {
match self_vers.lastest_mod_time().cmp(&other_vers.lastest_mod_time()) {
Ordering::Greater => {
return Ok((Some(self.clone()), false));
}
Ordering::Less => {
return Ok((Some(self.clone()), false));
}
_ => {}
}
if self_vers.versions.len() > other_vers.versions.len() {
return Ok((Some(self.clone()), false));
}
return Ok((Some(self.clone()), false));
}
for (s_version, o_version) in self_vers.versions.iter().zip(other_vers.versions.iter()) {
if s_version.header != o_version.header {
if s_version.header.has_ec() != o_version.header.has_ec() {
// One version has EC and the other doesn't - may have been written later.
// Compare without considering EC.
let (mut a, mut b) = (s_version.header.clone(), o_version.header.clone());
(a.ec_n, a.ec_m, b.ec_n, b.ec_m) = (0, 0, 0, 0);
if a == b {
continue;
}
}
if !strict && s_version.header.matches_not_strict(&o_version.header) {
if prefer.is_none() {
if s_version.header.sorts_before(&o_version.header) {
prefer = Some(self.clone());
} else {
prefer = Some(other.clone());
}
}
continue;
}
if prefer.is_some() {
return Ok((prefer, false));
}
if s_version.header.sorts_before(&o_version.header) {
return Ok((Some(self.clone()), false));
}
return Ok((Some(other.clone()), false));
}
}
if prefer.is_none() {
prefer = Some(self.clone());
}
Ok((prefer, true))
}
}
pub struct MetaCacheEntries(pub Vec<MetaCacheEntry>);
impl MetaCacheEntries {
pub fn resolve(&self, mut params: MetadataResolutionParams) -> Result<Option<MetaCacheEntry>> {
if self.0.is_empty() {
return Ok(None);
}
let mut dir_exists = 0;
let mut selected = None;
params.candidates.clear();
let mut objs_agree = 0;
let mut objs_valid = 0;
for entry in self.0.iter() {
if entry.name.is_empty() {
continue;
}
if entry.is_dir() {
dir_exists += 1;
selected = Some(entry.clone());
continue;
}
objs_valid += 1;
match &entry.cached {
Some(file_meta) => {
params.candidates.push(file_meta.versions.clone());
}
None => {
params.candidates.push(FileMeta::load(&entry.metadata)?.versions);
}
}
if selected.is_none() {
selected = Some(entry.clone());
objs_agree = 1;
continue;
}
if let (Some(prefer), true) = entry.matches(selected.as_ref().unwrap(), params.strict)? {
selected = Some(prefer);
objs_agree += 1;
continue;
}
}
// Return dir entries, if enough...
if selected.is_some() && selected.as_ref().unwrap().is_dir() && dir_exists >= params.dir_quorum {
return Ok(selected);
}
// If we would never be able to reach read quorum.
if objs_valid < params.obj_quorum {
return Ok(None);
}
// If all objects agree.
if selected.is_some() && objs_agree == objs_valid {
return Ok(selected);
}
// If cached is nil we shall skip the entry.
if selected.is_none() || (selected.is_some() && selected.as_ref().unwrap().cached.is_none()) {
return Ok(None);
}
// Merge if we have disagreement.
// Create a new merged result.
selected = Some(MetaCacheEntry {
name: selected.as_ref().unwrap().name.clone(),
cached: Some(FileMeta {
meta_ver: selected.as_ref().unwrap().cached.as_ref().unwrap().meta_ver.clone(),
..Default::default()
}),
_reusable: true,
..Default::default()
});
selected.as_mut().unwrap().cached.as_mut().unwrap().versions =
merge_file_meta_versions(params.obj_quorum, params.strict, params.requested_versions, &params.candidates);
if selected.as_ref().unwrap().cached.as_ref().unwrap().versions.is_empty() {
return Ok(None);
}
selected.as_mut().unwrap().metadata = selected.as_ref().unwrap().cached.as_ref().unwrap().marshal_msg()?;
Ok(selected)
}
pub fn first_found(&self) -> (Option<MetaCacheEntry>, usize) {
(self.0.iter().find(|x| !x.name.is_empty()).cloned(), self.0.len())
}
}
#[derive(Debug, Default)]
ecstore/src/disk/remote.rs
+3 -1
View File
@@ -20,7 +20,9 @@ use super::{
RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
};
use crate::{
disk::error::DiskError, error::{Error, Result}, heal::heal_commands::HealingTracker, store_api::{FileInfo, RawFileInfo}
disk::error::DiskError,
error::{Error, Result},
store_api::{FileInfo, RawFileInfo},
};
use protos::proto_gen::node_service::RenamePartRequst;
ecstore/src/global.rs
+2 -2
View File
@@ -11,8 +11,8 @@ use crate::{
lazy_static! {
pub static ref GLOBAL_OBJECT_API: Arc<RwLock<Option<ECStore>>> = Arc::new(RwLock::new(None));
pub static ref GLOBAL_LOCAL_DISK: Arc<RwLock<Vec<Option<DiskStore>>>> = Arc::new(RwLock::new(Vec::new()));
static ref GLOBAL_IsErasure: RwLock<bool> = RwLock::new(false);
static ref GLOBAL_IsDistErasure: RwLock<bool> = RwLock::new(false);
pub static ref GLOBAL_IsErasure: RwLock<bool> = RwLock::new(false);
pub static ref GLOBAL_IsDistErasure: RwLock<bool> = RwLock::new(false);
pub static ref GLOBAL_IsErasureSD: RwLock<bool> = RwLock::new(false);
pub static ref GLOBAL_LOCAL_DISK_MAP: Arc<RwLock<HashMap<String, Option<DiskStore>>>> = Arc::new(RwLock::new(HashMap::new()));
pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc<RwLock<TypeLocalDiskSetDrives>> = Arc::new(RwLock::new(Vec::new()));
ecstore/src/heal/background_heal_ops.rs
+89
View File
@@ -0,0 +1,89 @@
use std::sync::Arc;
use tokio::{
select,
sync::{
broadcast::Receiver as B_Receiver,
mpsc::{self, Receiver, Sender},
},
};
use crate::{
error::Error,
heal::heal_ops::NOP_HEAL,
utils::path::SLASH_SEPARATOR,
};
use super::{
heal_commands::{HealOpts, HealResultItem},
heal_ops::HealSequence,
};
#[derive(Clone, Debug)]
pub struct HealTask {
pub bucket: String,
pub object: String,
pub version_id: String,
pub opts: HealOpts,
pub resp_tx: Arc<Sender<HealResult>>,
pub resp_rx: Arc<Receiver<HealResult>>,
}
impl HealTask {
pub fn new(bucket: &str, object: &str, version_id: &str, opts: &HealOpts) -> Self {
let (tx, rx) = mpsc::channel(10);
Self {
bucket: bucket.to_string(),
object: object.to_string(),
version_id: version_id.to_string(),
opts: opts.clone(),
resp_tx: tx.into(),
resp_rx: rx.into(),
}
}
}
pub struct HealResult {
pub result: HealResultItem,
err: Error,
}
pub struct HealRoutine {
tasks_tx: Sender<HealTask>,
tasks_rx: Receiver<HealTask>,
workers: usize,
}
impl HealRoutine {
pub async fn add_worker(&mut self, mut ctx: B_Receiver<bool>, bgseq: &HealSequence) {
loop {
select! {
task = self.tasks_rx.recv() => {
let mut res = HealResultItem::default();
let mut err: Error;
match task {
Some(task) => {
if task.bucket == NOP_HEAL {
err = Error::from_string("skip file");
} else if task.bucket == SLASH_SEPARATOR {
(res, err) = heal_disk_format(task.opts).await;
}
},
None => return,
}
}
_ = ctx.recv() => {
return;
}
}
}
}
}
// pub fn active_listeners() -> Result<usize> {
// }
async fn heal_disk_format(opts: HealOpts) -> (HealResultItem, Error) {
todo!()
}
ecstore/src/heal/heal_commands.rs
+236 -61
View File
@@ -1,16 +1,27 @@
use std::{path::Path, time::{SystemTime, UNIX_EPOCH}};
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,
disk::{DeleteOptions, DiskStore, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
error::{Error, Result},
heal::heal_ops::HEALING_TRACKER_FILENAME,
new_object_layer_fn,
store_api::{BucketInfo, StorageAPI},
utils::fs::read_file,
};
pub type HealScanMode = usize;
pub type HealItemType = String;
pub const HEAL_UNKNOWN_SCAN: HealScanMode = 0;
pub const HEAL_NORMAL_SCAN: HealScanMode = 1;
pub const HEAL_DEEP_SCAN: HealScanMode = 2;
#[derive(Clone, Copy, Debug, Default)]
pub struct HealOpts {
pub recursive: bool,
@@ -24,14 +35,14 @@ pub struct HealOpts {
pub set: Option<usize>,
}
#[derive(Debug)]
#[derive(Clone, Debug)]
struct HealDriveInfo {
uuid: String,
endpoint: String,
state: String,
}
#[derive(Debug)]
#[derive(Clone, Debug, Default)]
pub struct HealResultItem {
pub result_index: usize,
pub heal_item_type: HealItemType,
@@ -48,54 +59,78 @@ pub struct HealResultItem {
pub object_size: usize,
}
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct HealStartSuccess {
pub client_token: String,
pub client_address: String,
pub start_time: u64,
}
pub type HealStopSuccess = HealStartSuccess;
pub struct HealingDisk {
pub id: String,
pub heal_id: String,
pub pool_index: Option<usize>,
pub set_index: Option<usize>,
pub disk_index: Option<usize>,
pub endpoint: String,
pub path: String,
pub started: u64,
pub last_update: u64,
pub retry_attempts: u64,
pub objects_total_count: u64,
pub objects_total_size: u64,
pub items_healed: u64,
pub items_failed: u64,
pub item_skipped: u64,
pub bytes_done: u64,
pub bytes_failed: u64,
pub bytes_skipped: u64,
pub objects_healed: u64,
pub objects_failed: u64,
pub bucket: String,
pub object: String,
pub queue_buckets: Vec<String>,
pub healed_buckets: Vec<String>,
pub finished: bool,
}
#[derive(Debug, Default, Deserialize, Serialize)]
pub struct HealingTracker {
#[serde(skip_serializing, skip_deserializing)]
disk: Option<DiskStore>,
id: String,
pool_index: Option<usize>,
set_index: Option<usize>,
disk_index: Option<usize>,
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<String>,
healed_buckets: Vec<String>,
heal_id: String,
item_skipped: u64,
bytes_skipped: u64,
retry_attempts: u64,
finished: bool,
pub disk: Option<DiskStore>,
pub id: String,
pub pool_index: Option<usize>,
pub set_index: Option<usize>,
pub disk_index: Option<usize>,
pub path: String,
pub endpoint: String,
pub started: u64,
pub last_update: u64,
pub objects_total_count: u64,
pub objects_total_size: u64,
pub items_healed: u64,
pub items_failed: u64,
pub item_skipped: u64,
pub bytes_done: u64,
pub bytes_failed: u64,
pub bytes_skipped: u64,
pub bucket: String,
pub object: String,
pub resume_items_healed: u64,
pub resume_items_failed: u64,
pub resume_items_skipped: u64,
pub resume_bytes_done: u64,
pub resume_bytes_failed: u64,
pub resume_bytes_skipped: u64,
pub queue_buckets: Vec<String>,
pub healed_buckets: Vec<String>,
pub heal_id: String,
pub retry_attempts: u64,
pub finished: bool,
#[serde(skip_serializing, skip_deserializing)]
mu: RwLock<bool>,
pub mu: RwLock<bool>,
}
impl HealingTracker {
@@ -106,8 +141,7 @@ impl HealingTracker {
}
pub fn unmarshal_msg(data: &[u8]) -> Result<Self> {
serde_json::from_slice::<HealingTracker>(data)
.map_err(|err| Error::from_string(err.to_string()))
serde_json::from_slice::<HealingTracker>(data).map_err(|err| Error::from_string(err.to_string()))
}
pub async fn reset_healing(&mut self) {
@@ -204,18 +238,157 @@ impl HealingTracker {
}
self.last_update = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("Time went backwards")
.as_secs();
.duration_since(UNIX_EPOCH)
.expect("Time went backwards")
.as_secs();
let htracker_bytes = self.marshal_msg()?;
// TODO: globalBackgroundHealState
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;
return disk
.write_all(RUSTFS_META_BUCKET, file_path.to_str().unwrap(), htracker_bytes)
.await;
}
Ok(())
}
async fn delete(&self) -> Result<()> {
if let Some(disk) = &self.disk {
let file_path = Path::new(BUCKET_META_PREFIX).join(HEALING_TRACKER_FILENAME);
return disk
.delete(
RUSTFS_META_BUCKET,
file_path.to_str().unwrap(),
DeleteOptions {
recursive: false,
immediate: false,
..Default::default()
},
)
.await;
}
Ok(())
}
async fn is_healed(&self, bucket: &str) -> bool {
self.mu.read().await;
for v in self.healed_buckets.iter() {
if v == bucket {
return true;
}
}
false
}
async fn resume(&mut self) {
self.mu.write().await;
self.items_healed = self.resume_items_healed;
self.items_failed = self.resume_items_failed;
self.item_skipped = self.resume_items_skipped;
self.bytes_done = self.resume_bytes_done;
self.bytes_failed = self.resume_bytes_failed;
self.bytes_skipped = self.resume_bytes_skipped;
}
async fn bucket_done(&mut self, bucket: &str) {
self.mu.write().await;
self.resume_items_healed = self.items_healed;
self.resume_items_failed = self.items_failed;
self.resume_items_skipped = self.item_skipped;
self.resume_bytes_done = self.bytes_done;
self.resume_bytes_failed = self.bytes_failed;
self.resume_bytes_skipped = self.bytes_skipped;
self.healed_buckets.push(bucket.to_string());
self.queue_buckets.retain(|x| x != bucket);
}
async fn set_queue_buckets(&mut self, buckets: &[BucketInfo]) {
self.mu.write().await;
buckets.iter().for_each(|bucket| {
if !self.healed_buckets.contains(&bucket.name) {
self.queue_buckets.push(bucket.name.clone());
}
});
}
pub async fn to_healing_disk(&self) -> HealingDisk {
self.mu.read().await;
HealingDisk {
id: self.id.clone(),
heal_id: self.heal_id.clone(),
pool_index: self.pool_index,
set_index: self.set_index,
disk_index: self.disk_index,
endpoint: self.endpoint.clone(),
path: self.path.clone(),
started: self.started,
last_update: self.last_update,
retry_attempts: self.retry_attempts,
objects_total_count: self.objects_total_count,
objects_total_size: self.objects_total_size,
items_healed: self.items_healed,
items_failed: self.items_failed,
item_skipped: self.item_skipped,
bytes_done: self.bytes_done,
bytes_failed: self.bytes_failed,
bytes_skipped: self.bytes_skipped,
objects_healed: self.items_healed,
objects_failed: self.items_failed,
bucket: self.bucket.clone(),
object: self.object.clone(),
queue_buckets: self.queue_buckets.clone(),
healed_buckets: self.healed_buckets.clone(),
finished: self.finished,
}
}
}
impl Clone for HealingTracker {
fn clone(&self) -> Self {
Self {
disk: self.disk.clone(),
id: self.id.clone(),
pool_index: self.pool_index.clone(),
set_index: self.set_index.clone(),
disk_index: self.disk_index.clone(),
path: self.path.clone(),
endpoint: self.endpoint.clone(),
started: self.started.clone(),
last_update: self.last_update.clone(),
objects_total_count: self.objects_total_count.clone(),
objects_total_size: self.objects_total_size.clone(),
items_healed: self.items_healed.clone(),
items_failed: self.items_failed.clone(),
item_skipped: self.item_skipped.clone(),
bytes_done: self.bytes_done.clone(),
bytes_failed: self.bytes_failed.clone(),
bytes_skipped: self.bytes_skipped.clone(),
bucket: self.bucket.clone(),
object: self.object.clone(),
resume_items_healed: self.resume_items_healed.clone(),
resume_items_failed: self.resume_items_failed.clone(),
resume_items_skipped: self.resume_items_skipped.clone(),
resume_bytes_done: self.resume_bytes_done.clone(),
resume_bytes_failed: self.resume_bytes_failed.clone(),
resume_bytes_skipped: self.resume_bytes_skipped.clone(),
queue_buckets: self.queue_buckets.clone(),
healed_buckets: self.healed_buckets.clone(),
heal_id: self.heal_id.clone(),
retry_attempts: self.retry_attempts.clone(),
finished: self.finished.clone(),
mu: RwLock::new(false),
}
}
}
async fn load_healing_tracker(disk: &Option<DiskStore>) -> Result<HealingTracker> {
@@ -227,14 +400,16 @@ async fn load_healing_tracker(disk: &Option<DiskStore>) -> Result<HealingTracker
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)));
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"));
}
@@ -247,9 +422,9 @@ async fn init_healing_tracker(disk: DiskStore, heal_id: String) -> Result<Healin
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();
.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;
ecstore/src/heal/heal_ops.rs
+434 -60
View File
@@ -1,64 +1,124 @@
use crate::{
disk::MetaCacheEntry,
disk::{endpoint::Endpoint, MetaCacheEntry},
endpoints::Endpoints,
error::{Error, Result},
global::GLOBAL_IsDistErasure,
heal::heal_commands::HEAL_UNKNOWN_SCAN,
utils::path::has_profix,
};
use lazy_static::lazy_static;
use std::{collections::HashMap, time::Instant};
use std::{
collections::HashMap,
future::Future,
path::Path,
pin::Pin,
sync::Arc,
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use tokio::{
select, spawn,
sync::{
broadcast::{self, Receiver, Sender},
mpsc::{self, Receiver as M_Receiver, Sender as M_Sender},
RwLock,
},
time::{interval, sleep},
};
use uuid::Uuid;
use super::heal_commands::{HealItemType, HealOpts, HealResultItem, HealScanMode};
use super::{
background_heal_ops::HealTask,
heal_commands::{HealItemType, HealOpts, HealResultItem, HealScanMode, HealStopSuccess, HealingDisk, HealingTracker},
};
type HealStatusSummary = String;
type ItemsMap = HashMap<HealItemType, usize>;
pub type HealObjectFn = Box<dyn Fn(&str, &str, &str, HealScanMode) -> Result<()> + Send>;
pub type HealEntryFn = Box<dyn Fn(String, MetaCacheEntry, HealScanMode) -> Result<()> + Send>;
pub type HealObjectFn = Arc<dyn Fn(&str, &str, &str, HealScanMode) -> Result<()> + Send + Sync>;
pub type HealEntryFn =
Box<dyn Fn(String, MetaCacheEntry, HealScanMode) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> + Send>;
lazy_static! {
static ref HEAL_NOT_STARTED_STATUS: HealStatusSummary = String::from("not started");
}
pub const BG_HEALING_UUID: &str = "0000-0000-0000-0000";
pub const HEALING_TRACKER_FILENAME: &str = ".healing.bin";
const KEEP_HEAL_SEQ_STATE_DURATION: std::time::Duration = Duration::from_secs(10 * 60);
const HEAL_NOT_STARTED_STATUS: &str = "not started";
const HEAL_RUNNING_STATUS: &str = "running";
const HEAL_STOPPED_STATUS: &str = "stopped";
const HEAL_FINISHED_STATUS: &str = "finished";
const MAX_UNCONSUMED_HEAL_RESULT_ITEMS: usize = 1000;
const HEAL_UNCONSUMED_TIMEOUT: std::time::Duration = Duration::from_secs(24 * 60 * 60);
pub const NOP_HEAL: &str = "";
lazy_static! {}
#[derive(Debug)]
#[derive(Clone, Debug, Default)]
pub struct HealSequenceStatus {
summary: HealStatusSummary,
failure_detail: String,
start_time: Instant,
heal_setting: HealOpts,
items: Vec<HealResultItem>,
pub summary: HealStatusSummary,
pub failure_detail: String,
pub start_time: u64,
pub heal_setting: HealOpts,
pub items: Vec<HealResultItem>,
}
impl Default for HealSequenceStatus {
fn default() -> Self {
Self {
summary: Default::default(),
failure_detail: Default::default(),
start_time: Instant::now(),
heal_setting: Default::default(),
items: Default::default(),
}
}
pub struct HealSource {
pub bucket: String,
pub object: String,
pub version_id: String,
pub no_wait: bool,
opts: Option<HealOpts>,
}
#[derive(Debug)]
#[derive(Clone, Debug)]
pub struct HealSequence {
pub bucket: String,
pub object: String,
pub report_progress: bool,
pub start_time: Instant,
pub end_time: Instant,
pub start_time: u64,
pub end_time: Arc<RwLock<u64>>,
pub client_token: String,
pub client_address: String,
pub force_started: bool,
pub setting: HealOpts,
pub current_status: HealSequenceStatus,
pub current_status: Arc<RwLock<HealSequenceStatus>>,
pub last_sent_result_index: usize,
pub scanned_items_map: ItemsMap,
pub healed_items_map: ItemsMap,
pub heal_failed_items_map: ItemsMap,
pub last_heal_activity: Instant,
pub last_heal_activity: u64,
traverse_and_heal_done_tx: Arc<RwLock<M_Sender<Option<Error>>>>,
traverse_and_heal_done_rx: Arc<RwLock<M_Receiver<Option<Error>>>>,
tx: Arc<RwLock<Sender<bool>>>,
rx: Arc<RwLock<Receiver<bool>>>,
}
impl Default for HealSequence {
fn default() -> Self {
let (h_tx, h_rx) = mpsc::channel(1);
let (tx, rx) = broadcast::channel(1);
Self {
bucket: Default::default(),
object: Default::default(),
report_progress: Default::default(),
start_time: Default::default(),
end_time: Default::default(),
client_token: Default::default(),
client_address: Default::default(),
force_started: Default::default(),
setting: Default::default(),
current_status: Default::default(),
last_sent_result_index: Default::default(),
scanned_items_map: Default::default(),
healed_items_map: Default::default(),
heal_failed_items_map: Default::default(),
last_heal_activity: Default::default(),
traverse_and_heal_done_tx: Arc::new(RwLock::new(h_tx)),
traverse_and_heal_done_rx: Arc::new(RwLock::new(h_rx)),
tx: Arc::new(RwLock::new(tx)),
rx: Arc::new(RwLock::new(rx)),
}
}
}
impl HealSequence {
@@ -73,11 +133,11 @@ impl HealSequence {
client_address: client_addr.to_string(),
force_started: force_start,
setting: hs,
current_status: HealSequenceStatus {
current_status: Arc::new(RwLock::new(HealSequenceStatus {
summary: HEAL_NOT_STARTED_STATUS.to_string(),
heal_setting: hs,
..Default::default()
},
})),
..Default::default()
}
}
@@ -102,36 +162,100 @@ impl HealSequence {
fn count_failed(&mut self, heal_type: HealItemType) {
*self.heal_failed_items_map.entry(heal_type).or_insert(0) += 1;
self.last_heal_activity = Instant::now();
self.last_heal_activity = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("Time went backwards")
.as_secs();
}
fn count_scanned(&mut self, heal_type: HealItemType) {
*self.scanned_items_map.entry(heal_type).or_insert(0) += 1;
self.last_heal_activity = Instant::now()
self.last_heal_activity = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("Time went backwards")
.as_secs();
}
fn count_healed(&mut self, heal_type: HealItemType) {
*self.healed_items_map.entry(heal_type).or_insert(0) += 1;
self.last_heal_activity = Instant::now()
self.last_heal_activity = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("Time went backwards")
.as_secs();
}
fn is_quitting(&self) -> bool {
todo!()
async fn is_quitting(&self) -> bool {
let mut w = self.rx.write().await;
if w.try_recv().is_ok() {
return true;
}
false
}
fn has_ended(&self) -> bool {
async fn has_ended(&self) -> bool {
if self.client_token == BG_HEALING_UUID.to_string() {
return false;
}
!(self.end_time == self.start_time)
!(*(self.end_time.read().await) == self.start_time)
}
fn stop(&self) {
todo!()
async fn stop(&self) {
let w = self.tx.write().await;
w.send(true);
}
fn push_heal_result_item(&self, r: HealResultItem) -> Result<()> {
async fn push_heal_result_item(&self, r: &HealResultItem) -> Result<()> {
let mut r = r.clone();
let mut interval_timer = interval(HEAL_UNCONSUMED_TIMEOUT);
let mut items_len = 0;
loop {
{
let current_status_r = self.current_status.read().await;
items_len = current_status_r.items.len();
}
if items_len == MAX_UNCONSUMED_HEAL_RESULT_ITEMS {
select! {
_ = sleep(Duration::from_secs(1)) => {
}
_ = self.is_done() => {
return Err(Error::from_string("stopped"));
}
_ = interval_timer.tick() => {
return Err(Error::from_string("timeout"));
}
}
} else {
break;
}
}
let mut current_status_w = self.current_status.write().await;
if items_len > 0 {
r.result_index = 1 + current_status_w.items[items_len - 1].result_index;
} else {
r.result_index = 1 + self.last_sent_result_index;
}
current_status_w.items.push(r);
Ok(())
}
async fn queue_heal_task(&mut self, source: HealSource, heal_type: HealItemType) -> Result<()> {
let mut task = HealTask::new(&source.bucket, &source.object, &source.version_id, &self.setting);
if let Some(opts) = source.opts {
task.opts = opts;
} else {
task.opts.scan_mode = HEAL_UNKNOWN_SCAN;
}
self.count_scanned(heal_type);
if source.no_wait {}
todo!()
}
@@ -147,33 +271,283 @@ impl HealSequence {
todo!()
}
fn traverse_and_heal(&self) {
async fn traverse_and_heal(&self) {
let buckets_only = false;
}
fn heal_rustfs_sys_meta(&self, meta_prefix: String) -> Result<()> {
todo!()
}
async fn is_done(&self) -> bool {
let mut rx_w = self.rx.write().await;
if let Ok(true) = rx_w.recv().await {
return true;
}
false
}
}
impl Default for HealSequence {
fn default() -> Self {
Self {
bucket: Default::default(),
object: Default::default(),
report_progress: Default::default(),
start_time: Instant::now(),
end_time: Instant::now(),
client_token: Default::default(),
client_address: Default::default(),
force_started: Default::default(),
setting: Default::default(),
current_status: Default::default(),
last_sent_result_index: Default::default(),
scanned_items_map: Default::default(),
healed_items_map: Default::default(),
heal_failed_items_map: Default::default(),
last_heal_activity: Instant::now(),
pub async fn heal_sequence_start(h: Arc<HealSequence>) {
{
let mut current_status_w = h.current_status.write().await;
(*current_status_w).summary = HEAL_RUNNING_STATUS.to_string();
(*current_status_w).start_time = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("Time went backwards")
.as_secs();
}
let h_clone = h.clone();
spawn(async move {
h_clone.traverse_and_heal().await;
});
let h_clone_1 = h.clone();
let mut x = h.traverse_and_heal_done_rx.write().await;
select! {
_ = h.is_done() => {
*(h.end_time.write().await) = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("Time went backwards")
.as_secs();
let mut current_status_w = h.current_status.write().await;
(*current_status_w).summary = HEAL_FINISHED_STATUS.to_string();
spawn(async move {
let mut rx_w = h_clone_1.traverse_and_heal_done_rx.write().await;
rx_w.recv().await;
});
}
result = x.recv() => {
match result {
Some(err) => {
match err {
Some(err) => {
let mut current_status_w = h.current_status.write().await;
(*current_status_w).summary = HEAL_STOPPED_STATUS.to_string();
(*current_status_w).failure_detail = err.to_string();
},
None => {
let mut current_status_w = h.current_status.write().await;
(*current_status_w).summary = HEAL_FINISHED_STATUS.to_string();
}
}
},
None => {
return;
}
}
}
}
}
#[derive(Debug, Default)]
pub struct AllHealState {
mu: RwLock<bool>,
heal_seq_map: HashMap<String, HealSequence>,
heal_local_disks: HashMap<Endpoint, bool>,
heal_status: HashMap<String, HealingTracker>,
}
impl AllHealState {
pub fn new(cleanup: bool) -> Self {
let hstate = AllHealState::default();
if cleanup {
// spawn(f);
}
hstate
}
async fn pop_heal_local_disks(&mut self, heal_local_disks: &[Endpoint]) {
self.mu.write().await;
self.heal_local_disks.retain(|k, _| {
if heal_local_disks.contains(k) {
return false;
}
true
});
let heal_local_disks = heal_local_disks.iter().map(|s| s.to_string()).collect::<Vec<_>>();
self.heal_status.retain(|_, v| {
if heal_local_disks.contains(&v.endpoint) {
return false;
}
true
});
}
async fn update_heal_status(&mut self, tracker: &HealingTracker) {
self.mu.write().await;
tracker.mu.read().await;
self.heal_status.insert(tracker.id.clone(), tracker.clone());
}
async fn get_local_healing_disks(&self) -> HashMap<String, HealingDisk> {
self.mu.read().await;
let mut dst = HashMap::new();
for v in self.heal_status.values() {
dst.insert(v.endpoint.clone(), v.to_healing_disk().await);
}
dst
}
async fn get_heal_local_disk_endpoints(&self) -> Endpoints {
self.mu.read().await;
let mut endpoints = Vec::new();
self.heal_local_disks.iter().for_each(|(k, v)| {
if !v {
endpoints.push(k.clone());
}
});
Endpoints::from(endpoints)
}
async fn set_disk_healing_status(&mut self, ep: Endpoint, healing: bool) {
self.mu.write().await;
self.heal_local_disks.insert(ep, healing);
}
async fn push_heal_local_disks(&mut self, heal_local_disks: &[Endpoint]) {
self.mu.write().await;
heal_local_disks.iter().for_each(|heal_local_disk| {
self.heal_local_disks.insert(heal_local_disk.clone(), false);
});
}
async fn periodic_heal_seqs_clean(&mut self, mut rx: Receiver<bool>) {
loop {
select! {
result = rx.recv() =>{
if let Ok(true) = result {
return;
}
}
_ = sleep(Duration::from_secs(5 * 60)) => {
self.mu.write().await;
let now = SystemTime::now();
let mut keys_to_reomve = Vec::new();
for (k, v) in self.heal_seq_map.iter() {
if v.has_ended().await && (UNIX_EPOCH + Duration::from_secs(*(v.end_time.read().await)) + KEEP_HEAL_SEQ_STATE_DURATION) < now {
keys_to_reomve.push(k.clone())
}
}
for key in keys_to_reomve.iter() {
self.heal_seq_map.remove(key);
}
}
}
}
}
async fn get_heal_sequence_by_token(&self, token: &str) -> (Option<HealSequence>, bool) {
self.mu.read().await;
for v in self.heal_seq_map.values() {
if v.client_token == token {
return (Some(v.clone()), true);
}
}
return (None, false);
}
async fn get_heal_sequence(&self, path: &str) -> Option<HealSequence> {
self.mu.read().await;
self.heal_seq_map.get(path).cloned()
}
async fn stop_heal_sequence(&mut self, path: &str) -> Result<Vec<u8>> {
let mut hsp = HealStopSuccess::default();
if let Some(he) = self.get_heal_sequence(path).await {
let client_token = he.client_token.clone();
if *GLOBAL_IsDistErasure.read().await {
// TODO: proxy
}
hsp.client_token = client_token;
hsp.client_address = he.client_address.clone();
hsp.start_time = he.start_time;
he.stop();
loop {
if he.has_ended().await {
break;
}
sleep(Duration::from_secs(1)).await;
}
self.mu.write().await;
self.heal_seq_map.remove(path);
} else {
hsp.client_token = "unknown".to_string();
}
let b = serde_json::to_string(&hsp)?;
Ok(b.as_bytes().to_vec())
}
// LaunchNewHealSequence - launches a background routine that performs
// healing according to the healSequence argument. For each heal
// sequence, state is stored in the `globalAllHealState`, which is a
// map of the heal path to `healSequence` which holds state about the
// heal sequence.
//
// Heal results are persisted in server memory for
// `keepHealSeqStateDuration`. This function also launches a
// background routine to clean up heal results after the
// aforementioned duration.
pub async fn launch_new_heal_sequence(&mut self, heal_sequence: &HealSequence) -> Result<Vec<u8>> {
let path = Path::new(&heal_sequence.bucket).join(heal_sequence.object.clone());
let path_s = path.to_str().unwrap();
if heal_sequence.force_started {
self.stop_heal_sequence(path_s).await?;
} else {
if let Some(hs) = self.get_heal_sequence(path_s).await {
if !hs.has_ended().await {
return Err(Error::from_string(format!("Heal is already running on the given path (use force-start option to stop and start afresh). The heal was started by IP {} at {}, token is {}", heal_sequence.client_address, heal_sequence.start_time, heal_sequence.client_token)));
}
}
}
self.mu.write().await;
for (k, v) in self.heal_seq_map.iter() {
if !v.has_ended().await && (has_profix(&k, path_s) || has_profix(path_s, &k)) {
return Err(Error::from_string(format!(
"The provided heal sequence path overlaps with an existing heal path: {}",
k
)));
}
}
self.heal_seq_map.insert(path_s.to_string(), heal_sequence.clone());
let client_token = heal_sequence.client_token.clone();
if *GLOBAL_IsDistErasure.read().await {
// TODO: proxy
}
if heal_sequence.client_token == BG_HEALING_UUID {
// For background heal do nothing, do not spawn an unnecessary goroutine.
} else {
}
todo!()
}
}
ecstore/src/heal/mod.rs
+1
View File
@@ -1,2 +1,3 @@
pub mod background_heal_ops;
pub mod heal_commands;
pub mod heal_ops;
ecstore/src/sets.rs
+20 -8
View File
@@ -12,17 +12,11 @@ use crate::{
disk::{
format::{DistributionAlgoVersion, FormatV3},
DiskStore,
},
endpoints::PoolEndpoints,
error::{Error, Result},
global::{is_dist_erasure, GLOBAL_LOCAL_DISK_SET_DRIVES},
set_disk::SetDisks,
store_api::{
}, endpoints::PoolEndpoints, error::{Error, Result}, global::{is_dist_erasure, GLOBAL_LOCAL_DISK_SET_DRIVES}, heal::{heal_commands::{HealOpts, HealResultItem}, heal_ops::HealObjectFn}, set_disk::SetDisks, store_api::{
BucketInfo, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, HTTPRangeSpec,
ListObjectsV2Info, MakeBucketOptions, MultipartUploadResult, ObjectIO, ObjectInfo, ObjectOptions, ObjectToDelete,
PartInfo, PutObjReader, StorageAPI,
},
utils::hash,
}, utils::hash
};
use tokio::time::Duration;
@@ -440,4 +434,22 @@ impl StorageAPI for Sets {
async fn delete_bucket(&self, _bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> {
unimplemented!()
}
async fn heal_format(&self, dry_run: bool) -> Result<HealResultItem> {
unimplemented!()
}
async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
unimplemented!()
}
async fn heal_object(&self, bucket: &str, object: &str, version_id: &str, opts: &HealOpts) -> Result<HealResultItem> {
unimplemented!()
}
async fn heal_objects(&self, bucket: &str, prefix: &str, opts: &HealOpts, func: HealObjectFn) -> Result<()> {
unimplemented!()
}
async fn get_pool_and_set(&self, id: &str) -> Result<(Option<usize>, Option<usize>, Option<usize>)> {
unimplemented!()
}
async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()> {
unimplemented!()
}
}
ecstore/src/store.rs
+76 -52
View File
@@ -7,6 +7,7 @@ use crate::disk::MetaCacheEntry;
use crate::global::{is_dist_erasure, set_object_layer, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES};
use crate::heal::heal_commands::{HealOpts, HealResultItem, HealScanMode};
use crate::heal::heal_ops::HealObjectFn;
use crate::new_object_layer_fn;
use crate::store_api::ObjectIO;
use crate::{
bucket::metadata::BucketMetadata,
@@ -827,54 +828,7 @@ impl StorageAPI for ECStore {
Err(Error::new(DiskError::FileNotFound))
}
async fn heal_objects(&self, bucket: &str, prefix: &str, opts: &HealOpts, func: HealObjectFn) -> Result<()> {
let heal_entry = |bucket: String, entry: MetaCacheEntry, scan_mode: HealScanMode| async move {
if entry.is_dir() {
return Ok(());
}
// We might land at .metacache, .trash, .multipart
// no need to heal them skip, only when bucket
// is '.rustfs.sys'
if bucket == RUSTFS_META_BUCKET {
if Pattern::new("buckets/*/.metacache/*")
.map(|p| p.matches(&entry.name))
.unwrap_or(false)
|| Pattern::new("tmp/*").map(|p| p.matches(&entry.name)).unwrap_or(false)
|| Pattern::new("multipart/*").map(|p| p.matches(&entry.name)).unwrap_or(false)
|| Pattern::new("tmp-old/*").map(|p| p.matches(&entry.name)).unwrap_or(false)
{
return Ok(());
}
}
match entry.file_info_versions(&bucket) {
Ok(fivs) => {
if opts.remove && !opts.dry_run {
if let Err(err) = self.check_abandoned_parts(&bucket, &entry.name, opts).await {
return Err(Error::from_string(format!(
"unable to check object {}/{} for abandoned data: {}",
bucket, entry.name, err
)));
}
}
for version in fivs.versions.iter() {
let version_id = version.version_id.map_or("".to_string(), |version_id| version_id.to_string());
if let Err(err) = func(&bucket, &entry.name, &version_id, scan_mode) {
match err.downcast_ref::<DiskError>() {
Some(DiskError::FileNotFound) | Some(DiskError::FileVersionNotFound) => {}
_ => return Err(err),
}
}
}
}
Err(_) => {
return func(&bucket, &entry.name, "", scan_mode);
}
}
Ok(())
};
let mut first_err = None;
for (idx, pool) in self.pools.iter().enumerate() {
if opts.pool.is_some() && opts.pool.unwrap() != idx {
continue;
@@ -886,14 +840,23 @@ impl StorageAPI for ECStore {
continue;
}
set.list
if let Err(err) = set.list_and_heal(bucket, prefix, opts, func.clone()).await {
if first_err.is_none() {
first_err = Some(err)
}
}
}
}
todo!()
if first_err.is_some() {
return Err(first_err.unwrap());
}
Ok(())
}
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 (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 {
@@ -921,7 +884,7 @@ impl StorageAPI for ECStore {
}
if !errs.is_empty() {
return Err(errs[0]);
return Err(errs[0].clone());
}
Ok(())
@@ -950,3 +913,64 @@ async fn init_local_peer(endpoint_pools: &EndpointServerPools, host: &String, po
*GLOBAL_Local_Node_Name.write().await = peer_set[0].clone();
}
pub async fn heal_entry(
bucket: String,
entry: MetaCacheEntry,
scan_mode: HealScanMode,
opts: HealOpts,
func: HealObjectFn,
) -> Result<()> {
if entry.is_dir() {
return Ok(());
}
// We might land at .metacache, .trash, .multipart
// no need to heal them skip, only when bucket
// is '.rustfs.sys'
if bucket == RUSTFS_META_BUCKET {
if Pattern::new("buckets/*/.metacache/*")
.map(|p| p.matches(&entry.name))
.unwrap_or(false)
|| Pattern::new("tmp/*").map(|p| p.matches(&entry.name)).unwrap_or(false)
|| Pattern::new("multipart/*").map(|p| p.matches(&entry.name)).unwrap_or(false)
|| Pattern::new("tmp-old/*").map(|p| p.matches(&entry.name)).unwrap_or(false)
{
return Ok(());
}
}
let layer = new_object_layer_fn();
let lock = layer.read().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(Error::msg("errServerNotInitialized")),
};
match entry.file_info_versions(&bucket) {
Ok(fivs) => {
if opts.remove && !opts.dry_run {
if let Err(err) = store.check_abandoned_parts(&bucket, &entry.name, &opts).await {
return Err(Error::from_string(format!(
"unable to check object {}/{} for abandoned data: {}",
bucket, entry.name, err
)));
}
}
for version in fivs.versions.iter() {
let version_id = version.version_id.map_or("".to_string(), |version_id| version_id.to_string());
if let Err(err) = func(&bucket, &entry.name, &version_id, scan_mode) {
match err.downcast_ref::<DiskError>() {
Some(DiskError::FileNotFound) | Some(DiskError::FileVersionNotFound) => {}
_ => return Err(err),
}
}
}
}
Err(_) => {
return func(&bucket, &entry.name, "", scan_mode);
}
}
Ok(())
}
ecstore/src/utils/path.rs
+12
View File
@@ -38,6 +38,18 @@ pub fn retain_slash(s: &str) -> String {
}
}
pub fn strings_has_prefix_fold(s: &str, prefix: &str) -> bool {
s.len() >= prefix.len() && (s[..prefix.len()] == *prefix || s[..prefix.len()].eq_ignore_ascii_case(prefix))
}
pub fn has_profix(s: &str, prefix: &str) -> bool {
if cfg!(target_os = "windows") {
return strings_has_prefix_fold(s, prefix);
}
s.starts_with(prefix)
}
pub struct LazyBuf {
s: String,
buf: Option<Vec<u8>>,
rustfs/src/grpc.rs
+5 -15
View File
@@ -1,7 +1,7 @@
use std::{error::Error, io::ErrorKind, pin::Pin};
use ecstore::{
disk::{DeleteOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, UpdateMetadataOpts, WalkDirOptions},
disk::{DeleteOptions, DiskInfoOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, UpdateMetadataOpts, WalkDirOptions},
erasure::{ReadAt, Write},
peer::{LocalPeerS3Client, PeerS3Client},
store::{all_local_disk_path, find_local_disk},
@@ -13,17 +13,7 @@ use lock::{lock_args::LockArgs, Locker, GLOBAL_LOCAL_SERVER};
use protos::{
models::{PingBody, PingBodyBuilder},
proto_gen::node_service::{
node_service_server::NodeService as Node, DeleteBucketRequest, DeleteBucketResponse, DeletePathsRequest,
DeletePathsResponse, DeleteRequest, DeleteResponse, DeleteVersionRequest, DeleteVersionResponse, DeleteVersionsRequest,
DeleteVersionsResponse, DeleteVolumeRequest, DeleteVolumeResponse, GenerallyLockRequest, GenerallyLockResponse,
GetBucketInfoRequest, GetBucketInfoResponse, ListBucketRequest, ListBucketResponse, ListDirRequest, ListDirResponse,
ListVolumesRequest, ListVolumesResponse, MakeBucketRequest, MakeBucketResponse, MakeVolumeRequest, MakeVolumeResponse,
MakeVolumesRequest, MakeVolumesResponse, PingRequest, PingResponse, ReadAllRequest, ReadAllResponse, ReadAtRequest,
ReadAtResponse, ReadMultipleRequest, ReadMultipleResponse, ReadVersionRequest, ReadVersionResponse, ReadXlRequest,
ReadXlResponse, RenameDataRequest, RenameDataResponse, RenameFileRequst, RenameFileResponse, RenamePartRequst,
RenamePartResponse, StatVolumeRequest, StatVolumeResponse, UpdateMetadataRequest, UpdateMetadataResponse, WalkDirRequest,
WalkDirResponse, WriteAllRequest, WriteAllResponse, WriteMetadataRequest, WriteMetadataResponse, WriteRequest,
WriteResponse,
node_service_server::NodeService as Node, DeleteBucketRequest, DeleteBucketResponse, DeletePathsRequest, DeletePathsResponse, DeleteRequest, DeleteResponse, DeleteVersionRequest, DeleteVersionResponse, DeleteVersionsRequest, DeleteVersionsResponse, DeleteVolumeRequest, DeleteVolumeResponse, DiskInfoRequest, DiskInfoResponse, GenerallyLockRequest, GenerallyLockResponse, GetBucketInfoRequest, GetBucketInfoResponse, ListBucketRequest, ListBucketResponse, ListDirRequest, ListDirResponse, ListVolumesRequest, ListVolumesResponse, MakeBucketRequest, MakeBucketResponse, MakeVolumeRequest, MakeVolumeResponse, MakeVolumesRequest, MakeVolumesResponse, PingRequest, PingResponse, ReadAllRequest, ReadAllResponse, ReadAtRequest, ReadAtResponse, ReadMultipleRequest, ReadMultipleResponse, ReadVersionRequest, ReadVersionResponse, ReadXlRequest, ReadXlResponse, RenameDataRequest, RenameDataResponse, RenameFileRequst, RenameFileResponse, RenamePartRequst, RenamePartResponse, StatVolumeRequest, StatVolumeResponse, UpdateMetadataRequest, UpdateMetadataResponse, WalkDirRequest, WalkDirResponse, WriteAllRequest, WriteAllResponse, WriteMetadataRequest, WriteMetadataResponse, WriteRequest, WriteResponse
},
};
use tokio::sync::mpsc;
@@ -1131,10 +1121,10 @@ impl Node for NodeService {
let opts = match serde_json::from_str::<DiskInfoOptions>(&request.opts) {
Ok(opts) => opts,
Err(_) => {
return Ok(tonic::Response::new(ReadMultipleResponse {
return Ok(tonic::Response::new(DiskInfoResponse {
success: false,
read_multiple_resps: Vec::new(),
error_info: Some("can not decode ReadMultipleReq".to_string()),
disk_info: "".to_string(),
error_info: Some("can not decode DiskInfoOptions".to_string()),
}));
}
};