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

support mc admin heal command

Browse Source 
Signed-off-by: mujunxiang <1948535941@qq.com>
This commit is contained in:
mujunxiang
2024-11-30 21:29:37 +08:00
parent 26c5ebea53
commit 1a418c74b4
19 changed files with 388 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+13
View File
@@ -52,6 +52,12 @@ dependencies = [
"memchr",
]
[[package]]
name = "android-tzdata"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e999941b234f3131b00bc13c22d06e8c5ff726d1b6318ac7eb276997bbb4fef0"
[[package]]
name = "android_system_properties"
version = "0.1.5"
@@ -374,7 +380,13 @@ version = "0.4.38"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a21f936df1771bf62b77f047b726c4625ff2e8aa607c01ec06e5a05bd8463401"
dependencies = [
"android-tzdata",
"iana-time-zone",
"js-sys",
"num-traits",
"serde",
"wasm-bindgen",
"windows-targets",
]
[[package]]
@@ -645,6 +657,7 @@ dependencies = [
"byteorder",
"bytes",
"bytesize",
"chrono",
"common",
"crc32fast",
"flatbuffers",
ecstore/Cargo.toml
+1
View File
@@ -14,6 +14,7 @@ backon.workspace = true
blake2 = "0.10.6"
bytes.workspace = true
common.workspace = true
chrono = { version = "0.4.38", features = ["serde"] }
reader.workspace = true
glob = "0.3.1"
thiserror.workspace = true
ecstore/src/admin_server_info.rs
+3 -3
View File
@@ -119,11 +119,11 @@ pub async fn get_local_server_property() -> ServerProperties {
network.insert(node_name, ITEM_ONLINE.to_string());
continue;
}
if !network.contains_key(&node_name) {
if let std::collections::hash_map::Entry::Vacant(e) = network.entry(node_name) {
if is_server_resolvable(endpoint).await.is_err() {
network.insert(node_name, ITEM_OFFLINE.to_string());
e.insert(ITEM_OFFLINE.to_string());
} else {
network.insert(node_name, ITEM_ONLINE.to_string());
e.insert(ITEM_ONLINE.to_string());
}
}
}
ecstore/src/bucket/versioning_sys.rs
+6
View File
@@ -6,6 +6,12 @@ use tracing::warn;
pub struct BucketVersioningSys {}
impl Default for BucketVersioningSys {
fn default() -> Self {
Self::new()
}
}
impl BucketVersioningSys {
pub fn new() -> Self {
Self {}
ecstore/src/disk/local.rs
+1 -1
View File
@@ -1842,7 +1842,7 @@ impl DiskAPI for LocalDisk {
if let Some(old_data_dir) = opts.old_data_dir {
if opts.undo_write {
let src_path = file_path.join(Path::new(
format!("{}{}{}", old_data_dir.to_string(), SLASH_SEPARATOR, STORAGE_FORMAT_FILE_BACKUP).as_str(),
format!("{}{}{}", old_data_dir, SLASH_SEPARATOR, STORAGE_FORMAT_FILE_BACKUP).as_str(),
));
let dst_path = file_path.join(Path::new(format!("{}{}{}", path, SLASH_SEPARATOR, STORAGE_FORMAT_FILE).as_str()));
return rename_all(src_path, dst_path, file_path).await;
ecstore/src/erasure.rs
+1 -1
View File
@@ -10,8 +10,8 @@ use std::fmt::Debug;
use std::io::ErrorKind;
use tokio::io::DuplexStream;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tracing::info;
use tracing::warn;
use tracing::{debug, info};
// use tracing::debug;
use uuid::Uuid;
ecstore/src/global.rs
+3 -6
View File
@@ -29,7 +29,8 @@ lazy_static! {
pub static ref GLOBAL_Endpoints: OnceLock<EndpointServerPools> = OnceLock::new();
pub static ref GLOBAL_RootDiskThreshold: RwLock<u64> = RwLock::new(0);
pub static ref GLOBAL_BackgroundHealRoutine: Arc<HealRoutine> = HealRoutine::new();
pub static ref GLOBAL_BackgroundHealState: Arc<RwLock<AllHealState>> = AllHealState::new(false);
pub static ref GLOBAL_BackgroundHealState: Arc<AllHealState> = AllHealState::new(false);
pub static ref GLOBAL_ALlHealState: Arc<AllHealState> = AllHealState::new(false);
static ref globalDeploymentIDPtr: RwLock<Uuid> = RwLock::new(Uuid::nil());
}
@@ -50,11 +51,7 @@ pub fn set_global_endpoints(eps: Vec<PoolEndpoints>) {
}
pub fn new_object_layer_fn() -> Option<Arc<ECStore>> {
if let Some(ec) = GLOBAL_OBJECT_API.get() {
Some(ec.clone())
} else {
None
}
GLOBAL_OBJECT_API.get().map(|ec| ec.clone())
}
pub async fn set_object_layer(o: Arc<ECStore>) {
ecstore/src/heal/background_heal_ops.rs
+3 -17
View File
@@ -39,8 +39,6 @@ pub async fn init_auto_heal() {
if v == "on" {
info!("start monitor local disks and heal");
GLOBAL_BackgroundHealState
.write()
.await
.push_heal_local_disks(&get_local_disks_to_heal().await)
.await;
tokio::spawn(async {
@@ -58,11 +56,7 @@ async fn init_background_healing() {
GLOBAL_BackgroundHealRoutine.add_worker(bg_seq_clone).await;
});
}
let _ = GLOBAL_BackgroundHealState
.write()
.await
.launch_new_heal_sequence(bg_seq)
.await;
let _ = GLOBAL_BackgroundHealState.launch_new_heal_sequence(bg_seq).await;
}
pub async fn get_local_disks_to_heal() -> Vec<Endpoint> {
@@ -95,7 +89,7 @@ async fn monitor_local_disks_and_heal() {
loop {
interval.tick().await;
let heal_disks = GLOBAL_BackgroundHealState.read().await.get_heal_local_disk_endpoints().await;
let heal_disks = GLOBAL_BackgroundHealState.get_heal_local_disk_endpoints().await;
if heal_disks.is_empty() {
interval.reset();
continue;
@@ -115,23 +109,15 @@ async fn monitor_local_disks_and_heal() {
let disk_clone = disk.clone();
tokio::spawn(async move {
GLOBAL_BackgroundHealState
.write()
.await
.set_disk_healing_status(disk_clone.clone(), true)
.await;
if heal_fresh_disk(&disk_clone).await.is_err() {
GLOBAL_BackgroundHealState
.write()
.await
.set_disk_healing_status(disk_clone.clone(), false)
.await;
return;
}
GLOBAL_BackgroundHealState
.write()
.await
.pop_heal_local_disks(&[disk_clone])
.await;
GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await;
});
}
interval.reset();
ecstore/src/heal/data_scanner.rs
+1 -5
View File
@@ -719,11 +719,7 @@ impl FolderScanner {
break 'outer;
}
let (bg_seq, found) = GLOBAL_BackgroundHealState
.read()
.await
.get_heal_sequence_by_token(BG_HEALING_UUID)
.await;
let (bg_seq, found) = GLOBAL_BackgroundHealState.get_heal_sequence_by_token(BG_HEALING_UUID).await;
if !found {
break 'outer;
}
ecstore/src/heal/heal_commands.rs
+34 -7
View File
@@ -1,5 +1,6 @@
use std::{path::Path, time::SystemTime};
use chrono::{DateTime, Utc};
use lazy_static::lazy_static;
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
@@ -44,45 +45,71 @@ lazy_static! {
#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize)]
pub struct HealOpts {
pub recursive: bool,
#[serde(rename = "dryRun")]
pub dry_run: bool,
pub remove: bool,
pub recreate: bool,
#[serde(rename = "scanMode")]
pub scan_mode: HealScanMode,
#[serde(rename = "updateParity")]
pub update_parity: bool,
#[serde(rename = "nolock")]
pub no_lock: bool,
pub pool: Option<usize>,
pub set: Option<usize>,
}
#[derive(Clone, Debug, Default)]
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct HealDriveInfo {
pub uuid: String,
pub endpoint: String,
pub state: String,
}
#[derive(Clone, Debug, Default)]
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct Infos {
#[serde(rename = "drives")]
pub drives: Vec<HealDriveInfo>,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct HealResultItem {
#[serde(rename = "resultId")]
pub result_index: usize,
#[serde(rename = "type")]
pub heal_item_type: HealItemType,
#[serde(rename = "bucket")]
pub bucket: String,
#[serde(rename = "object")]
pub object: String,
#[serde(rename = "versionId")]
pub version_id: String,
#[serde(rename = "detail")]
pub detail: String,
#[serde(rename = "parityBlocks")]
pub parity_blocks: usize,
#[serde(rename = "dataBlocks")]
pub data_blocks: usize,
#[serde(rename = "diskCount")]
pub disk_count: usize,
#[serde(rename = "setCount")]
pub set_count: usize,
pub before: Vec<HealDriveInfo>,
pub after: Vec<HealDriveInfo>,
#[serde(rename = "before")]
pub before: Infos,
#[serde(rename = "after")]
pub after: Infos,
#[serde(rename = "objectSize")]
pub object_size: usize,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct HealStartSuccess {
#[serde(rename = "clientToken")]
pub client_token: String,
#[serde(rename = "clientAddress")]
pub client_address: String,
pub start_time: SystemTime,
#[serde(rename = "startTime")]
pub start_time: DateTime<Utc>,
}
impl Default for HealStartSuccess {
@@ -90,7 +117,7 @@ impl Default for HealStartSuccess {
Self {
client_token: Default::default(),
client_address: Default::default(),
start_time: SystemTime::now(),
start_time: Utc::now(),
}
}
}
@@ -265,7 +292,7 @@ impl HealingTracker {
let htracker_bytes = self.marshal_msg()?;
GLOBAL_BackgroundHealState.write().await.update_heal_status(self).await;
GLOBAL_BackgroundHealState.update_heal_status(self).await;
if let Some(disk) = &self.disk {
let file_path = Path::new(BUCKET_META_PREFIX).join(HEALING_TRACKER_FILENAME);
ecstore/src/heal/heal_ops.rs
+90 -39
View File
@@ -7,7 +7,6 @@ use super::{
HEAL_ITEM_BUCKET_METADATA,
},
};
use crate::heal::heal_commands::{HEAL_ITEM_BUCKET, HEAL_ITEM_OBJECT};
use crate::store_api::StorageAPI;
use crate::{
config::common::CONFIG_PREFIX,
@@ -27,12 +26,18 @@ use crate::{
new_object_layer_fn,
utils::path::has_profix,
};
use crate::{
heal::heal_commands::{HEAL_ITEM_BUCKET, HEAL_ITEM_OBJECT},
utils::path::path_join,
};
use chrono::Utc;
use futures::join;
use lazy_static::lazy_static;
use serde::{Deserialize, Serialize};
use std::{
collections::HashMap,
future::Future,
path::Path,
path::PathBuf,
pin::Pin,
sync::Arc,
time::{Duration, SystemTime, UNIX_EPOCH},
@@ -73,7 +78,7 @@ pub const NOP_HEAL: &str = "";
lazy_static! {}
#[derive(Clone, Debug, Default)]
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct HealSequenceStatus {
pub summary: HealStatusSummary,
pub failure_detail: String,
@@ -103,7 +108,7 @@ pub struct HealSequence {
pub force_started: bool,
pub setting: HealOpts,
pub current_status: Arc<RwLock<HealSequenceStatus>>,
pub last_sent_result_index: usize,
pub last_sent_result_index: RwLock<usize>,
pub scanned_items_map: RwLock<ItemsMap>,
pub healed_items_map: RwLock<ItemsMap>,
pub heal_failed_items_map: RwLock<ItemsMap>,
@@ -151,7 +156,7 @@ pub fn new_heal_sequence(bucket: &str, obj_prefix: &str, client_addr: &str, hs:
client_token,
client_address: client_addr.to_string(),
force_started: force_start,
setting: hs.clone(),
setting: hs,
current_status: Arc::new(RwLock::new(HealSequenceStatus {
summary: HEAL_NOT_STARTED_STATUS.to_string(),
heal_setting: hs,
@@ -300,7 +305,7 @@ impl HealSequence {
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;
r.result_index = 1 + *self.last_sent_result_index.read().await;
}
current_status_w.items.push(r);
@@ -369,7 +374,7 @@ impl HealSequence {
}
if res.result.parity_blocks > 0 && res.result.data_blocks > 0 && res.result.data_blocks > res.result.parity_blocks
{
let got = count_ok_drives(&res.result.after);
let got = count_ok_drives(&res.result.after.drives);
if got < res.result.parity_blocks {
res.result.detail = format!(
"quorum loss - expected {} minimum, got drive states in OK {}",
@@ -563,14 +568,14 @@ pub async fn heal_sequence_start(h: Arc<HealSequence>) {
pub struct AllHealState {
mu: RwLock<bool>,
heal_seq_map: HashMap<String, Arc<HealSequence>>,
heal_local_disks: HashMap<Endpoint, bool>,
heal_status: HashMap<String, HealingTracker>,
heal_seq_map: RwLock<HashMap<String, Arc<HealSequence>>>,
heal_local_disks: RwLock<HashMap<Endpoint, bool>>,
heal_status: RwLock<HashMap<String, HealingTracker>>,
}
impl AllHealState {
pub fn new(cleanup: bool) -> Arc<RwLock<Self>> {
let hstate = Arc::new(RwLock::new(AllHealState::default()));
pub fn new(cleanup: bool) -> Arc<Self> {
let hstate = Arc::new(AllHealState::default());
let (_, mut rx) = broadcast::channel(1);
if cleanup {
let hstate_clone = hstate.clone();
@@ -583,7 +588,7 @@ impl AllHealState {
}
}
_ = sleep(Duration::from_secs(5 * 60)) => {
hstate_clone.write().await.periodic_heal_seqs_clean().await;
hstate_clone.periodic_heal_seqs_clean().await;
}
}
}
@@ -593,10 +598,10 @@ impl AllHealState {
hstate
}
pub async fn pop_heal_local_disks(&mut self, heal_local_disks: &[Endpoint]) {
pub async fn pop_heal_local_disks(&self, heal_local_disks: &[Endpoint]) {
let _ = self.mu.write().await;
self.heal_local_disks.retain(|k, _| {
self.heal_local_disks.write().await.retain(|k, _| {
if heal_local_disks.contains(k) {
return false;
}
@@ -604,7 +609,7 @@ impl AllHealState {
});
let heal_local_disks = heal_local_disks.iter().map(|s| s.to_string()).collect::<Vec<_>>();
self.heal_status.retain(|_, v| {
self.heal_status.write().await.retain(|_, v| {
if heal_local_disks.contains(&v.endpoint) {
return false;
}
@@ -613,18 +618,57 @@ impl AllHealState {
});
}
pub async fn update_heal_status(&mut self, tracker: &HealingTracker) {
pub async fn pop_heal_status_json(&self, heal_path: &str, client_token: &str) -> Result<Vec<u8>> {
match self.get_heal_sequence(heal_path).await {
Some(h) => {
if client_token != h.client_token {
info!("err heal invalid client token");
return Err(Error::from_string("err heal invalid client token"));
}
let num_items = h.current_status.read().await.items.len();
let mut last_result_index = *h.last_sent_result_index.read().await;
if num_items > 0 {
if let Some(item) = h.current_status.read().await.items.last() {
last_result_index = item.result_index;
}
}
*h.last_sent_result_index.write().await = last_result_index;
let data = h.current_status.read().await.clone();
match serde_json::to_vec(&data) {
Ok(b) => {
h.current_status.write().await.items.clear();
Ok(b)
}
Err(e) => {
h.current_status.write().await.items.clear();
info!("json encode err, e: {}", e);
Err(Error::msg(e.to_string()))
}
}
}
None => serde_json::to_vec(&HealSequenceStatus {
summary: HEAL_FINISHED_STATUS.to_string(),
..Default::default()
})
.map_err(|e| {
info!("json encode err, e: {}", e);
Error::msg(e.to_string())
}),
}
}
pub async fn update_heal_status(&self, tracker: &HealingTracker) {
let _ = self.mu.write().await;
let _ = tracker.mu.read().await;
self.heal_status.insert(tracker.id.clone(), tracker.clone());
self.heal_status.write().await.insert(tracker.id.clone(), tracker.clone());
}
pub async fn get_local_healing_disks(&self) -> HashMap<String, HealingDisk> {
let _ = self.mu.read().await;
let mut dst = HashMap::new();
for v in self.heal_status.values() {
for v in self.heal_status.read().await.values() {
dst.insert(v.endpoint.clone(), v.to_healing_disk().await);
}
@@ -635,7 +679,7 @@ impl AllHealState {
let _ = self.mu.read().await;
let mut endpoints = Vec::new();
self.heal_local_disks.iter().for_each(|(k, v)| {
self.heal_local_disks.read().await.iter().for_each(|(k, v)| {
if !v {
endpoints.push(k.clone());
}
@@ -644,39 +688,39 @@ impl AllHealState {
Endpoints::from(endpoints)
}
pub async fn set_disk_healing_status(&mut self, ep: Endpoint, healing: bool) {
pub async fn set_disk_healing_status(&self, ep: Endpoint, healing: bool) {
let _ = self.mu.write().await;
self.heal_local_disks.insert(ep, healing);
self.heal_local_disks.write().await.insert(ep, healing);
}
pub async fn push_heal_local_disks(&mut self, heal_local_disks: &[Endpoint]) {
pub async fn push_heal_local_disks(&self, heal_local_disks: &[Endpoint]) {
let _ = self.mu.write().await;
heal_local_disks.iter().for_each(|heal_local_disk| {
self.heal_local_disks.insert(heal_local_disk.clone(), false);
});
for heal_local_disk in heal_local_disks.iter() {
self.heal_local_disks.write().await.insert(heal_local_disk.clone(), false);
}
}
pub async fn periodic_heal_seqs_clean(&mut self) {
pub async fn periodic_heal_seqs_clean(&self) {
let _ = self.mu.write().await;
let now = SystemTime::now();
let mut keys_to_reomve = Vec::new();
for (k, v) in self.heal_seq_map.iter() {
for (k, v) in self.heal_seq_map.read().await.iter() {
if v.has_ended().await && now.duration_since(*(v.end_time.read().await)).unwrap() > KEEP_HEAL_SEQ_STATE_DURATION {
keys_to_reomve.push(k.clone())
}
}
for key in keys_to_reomve.iter() {
self.heal_seq_map.remove(key);
self.heal_seq_map.write().await.remove(key);
}
}
pub async fn get_heal_sequence_by_token(&self, token: &str) -> (Option<Arc<HealSequence>>, bool) {
let _ = self.mu.read().await;
for v in self.heal_seq_map.values() {
for v in self.heal_seq_map.read().await.values() {
if v.client_token == token {
return (Some(v.clone()), true);
}
@@ -688,10 +732,10 @@ impl AllHealState {
pub async fn get_heal_sequence(&self, path: &str) -> Option<Arc<HealSequence>> {
let _ = self.mu.read().await;
self.heal_seq_map.get(path).cloned()
self.heal_seq_map.read().await.get(path).cloned()
}
pub async fn stop_heal_sequence(&mut self, path: &str) -> Result<Vec<u8>> {
pub async fn stop_heal_sequence(&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();
@@ -701,7 +745,7 @@ impl AllHealState {
hsp.client_token = client_token;
hsp.client_address = he.client_address.clone();
hsp.start_time = he.start_time;
hsp.start_time = Utc::now();
he.stop().await;
@@ -714,7 +758,7 @@ impl AllHealState {
}
let _ = self.mu.write().await;
self.heal_seq_map.remove(path);
self.heal_seq_map.write().await.remove(path);
} else {
hsp.client_token = "unknown".to_string();
}
@@ -733,8 +777,11 @@ impl AllHealState {
// `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: Arc<HealSequence>) -> Result<Vec<u8>> {
let path = Path::new(&heal_sequence.bucket).join(heal_sequence.object.clone());
pub async fn launch_new_heal_sequence(&self, heal_sequence: Arc<HealSequence>) -> Result<Vec<u8>> {
let path = path_join(&[
PathBuf::from(heal_sequence.bucket.clone()),
PathBuf::from(heal_sequence.object.clone()),
]);
let path_s = path.to_str().unwrap();
if heal_sequence.force_started {
self.stop_heal_sequence(path_s).await?;
@@ -746,7 +793,7 @@ impl AllHealState {
let _ = self.mu.write().await;
for (k, v) in self.heal_seq_map.iter() {
for (k, v) in self.heal_seq_map.read().await.iter() {
if (has_profix(k, path_s) || has_profix(path_s, k)) && !v.has_ended().await {
return Err(Error::from_string(format!(
"The provided heal sequence path overlaps with an existing heal path: {}",
@@ -755,7 +802,10 @@ impl AllHealState {
}
}
self.heal_seq_map.insert(path_s.to_string(), heal_sequence.clone());
self.heal_seq_map
.write()
.await
.insert(path_s.to_string(), heal_sequence.clone());
let client_token = heal_sequence.client_token.clone();
if *GLOBAL_IsDistErasure.read().await {
@@ -774,7 +824,8 @@ impl AllHealState {
let b = serde_json::to_vec(&HealStartSuccess {
client_token,
client_address: heal_sequence.client_address.clone(),
start_time: heal_sequence.start_time,
// start_time: Utc::now(),
start_time: heal_sequence.start_time.into(),
})?;
Ok(b)
}
ecstore/src/peer.rs
+4 -4
View File
@@ -8,9 +8,9 @@ use regex::Regex;
use std::{collections::HashMap, fmt::Debug, sync::Arc};
use tokio::sync::RwLock;
use tonic::Request;
use tracing::{error, info, warn};
use tracing::{info, warn};
use crate::disk::error::{is_all_buckets_not_found, is_all_not_found};
use crate::disk::error::is_all_buckets_not_found;
use crate::disk::{DiskAPI, DiskStore};
use crate::global::GLOBAL_LOCAL_DISK_MAP;
use crate::heal::heal_commands::{
@@ -716,7 +716,7 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result<HealResu
}
for (disk, state) in disks.iter().zip(before_state.read().await.iter()) {
res.before.push(HealDriveInfo {
res.before.drives.push(HealDriveInfo {
uuid: "".to_string(),
endpoint: disk.clone().map(|s| s.to_string()).unwrap_or_default(),
state: state.to_string(),
@@ -773,7 +773,7 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result<HealResu
}
for (disk, state) in disks.iter().zip(after_state.read().await.iter()) {
res.before.push(HealDriveInfo {
res.before.drives.push(HealDriveInfo {
uuid: "".to_string(),
endpoint: disk.clone().map(|s| s.to_string()).unwrap_or_default(),
state: state.to_string(),
ecstore/src/pools.rs
+14 -16
View File
@@ -114,7 +114,7 @@ impl PoolMeta {
data.write_all(&buf)?;
for pool in pools {
save_config(pool, &POOL_META_NAME, &data).await?;
save_config(pool, POOL_META_NAME, &data).await?;
}
Ok(())
@@ -201,7 +201,7 @@ impl PoolMeta {
}
let now = OffsetDateTime::now_utc();
pool.last_update = now.clone();
pool.last_update = now;
pool.decommission = Some(PoolDecommissionInfo {
start_time: Some(now),
start_size: pi.free,
@@ -271,7 +271,7 @@ impl PoolMeta {
}
fn path2_bucket_object(name: &String) -> (String, String) {
path2_bucket_object_with_base_path("", &name)
path2_bucket_object_with_base_path("", name)
}
fn path2_bucket_object_with_base_path(base_path: &str, path: &str) -> (String, String) {
@@ -342,7 +342,7 @@ impl PoolDecommissionInfo {
let mut found = None;
for (i, b) in self.queued_buckets.iter().enumerate() {
if b == bucket {
found = Some(i.clone());
found = Some(i);
break;
}
}
@@ -471,7 +471,7 @@ impl ECStore {
return;
};
for idx in indices.iter() {
store.do_decommission_in_routine(idx.clone()).await;
store.do_decommission_in_routine(*idx).await;
}
});
@@ -481,7 +481,7 @@ impl ECStore {
async fn decommission_pool<S: StorageAPI>(&self, _idx: usize, _pool: Arc<S>, bi: DecomBucketInfo) -> Result<()> {
let mut _vc = None;
if &bi.name == RUSTFS_META_BUCKET {
if bi.name == RUSTFS_META_BUCKET {
let versioning = BucketVersioningSys::get(&bi.name).await?;
_vc = Some(versioning);
}
@@ -518,10 +518,8 @@ impl ECStore {
if let Err(er) = self.decommission_failed(idx).await {
error!("decom failed err {:?}", &er);
}
} else {
if let Err(er) = self.complete_decommission(idx).await {
error!("decom complete err {:?}", &er);
}
} else if let Err(er) = self.complete_decommission(idx).await {
error!("decom complete err {:?}", &er);
}
}
@@ -616,7 +614,7 @@ impl ECStore {
let _ = self.heal_bucket(&bk.name, &HealOpts::default()).await;
}
let meta_buckets = vec![
let meta_buckets = [
path_join(&[PathBuf::from(RUSTFS_META_BUCKET), PathBuf::from(CONFIG_PREFIX)]),
path_join(&[PathBuf::from(RUSTFS_META_BUCKET), PathBuf::from(BUCKET_META_PREFIX)]),
];
@@ -634,11 +632,11 @@ impl ECStore {
let mut pool_meta = self.pool_meta.write().await;
for idx in indices.iter() {
let pi = self.get_decommission_pool_space_info(idx.clone()).await?;
let pi = self.get_decommission_pool_space_info(*idx).await?;
pool_meta.decommission(idx.clone(), pi)?;
pool_meta.decommission(*idx, pi)?;
pool_meta.queue_buckets(idx.clone(), decom_buckets.clone());
pool_meta.queue_buckets(*idx, decom_buckets.clone());
}
pool_meta.save(self.pools.clone()).await?;
@@ -672,7 +670,7 @@ impl ECStore {
}
}
fn get_total_usable_capacity(disks: &Vec<StorageDisk>, info: &StorageInfo) -> usize {
fn get_total_usable_capacity(disks: &[StorageDisk], info: &StorageInfo) -> usize {
let mut capacity = 0;
for disk in disks.iter() {
if disk.pool_index < 0 || info.backend.standard_sc_data.len() <= disk.pool_index as usize {
@@ -685,7 +683,7 @@ fn get_total_usable_capacity(disks: &Vec<StorageDisk>, info: &StorageInfo) -> us
capacity
}
fn get_total_usable_capacity_free(disks: &Vec<StorageDisk>, info: &StorageInfo) -> usize {
fn get_total_usable_capacity_free(disks: &[StorageDisk], info: &StorageInfo) -> usize {
let mut capacity = 0;
for disk in disks.iter() {
if disk.pool_index < 0 || info.backend.standard_sc_data.len() <= disk.pool_index as usize {
ecstore/src/sets.rs
+6 -6
View File
@@ -587,12 +587,12 @@ impl StorageAPI for Sets {
..Default::default()
};
let before_derives = formats_to_drives_info(&self.endpoints.endpoints, &formats, &errs);
res.before = vec![HealDriveInfo::default(); before_derives.len()];
res.after = vec![HealDriveInfo::default(); before_derives.len()];
res.before.drives = vec![HealDriveInfo::default(); before_derives.len()];
res.after.drives = vec![HealDriveInfo::default(); before_derives.len()];
for v in before_derives.iter() {
res.before.push(v.clone());
res.after.push(v.clone());
res.before.drives.push(v.clone());
res.after.drives.push(v.clone());
}
if DiskError::UnformattedDisk.count_errs(&errs) == 0 {
return Ok((res, Some(Error::new(DiskError::NoHealRequired))));
@@ -609,8 +609,8 @@ impl StorageAPI for Sets {
for (i, set) in new_format_sets.iter().enumerate() {
for (j, fm) in set.iter().enumerate() {
if let Some(fm) = fm {
res.after[i * self.set_drive_count + j].uuid = fm.erasure.this.to_string();
res.after[i * self.set_drive_count + j].state = DRIVE_STATE_OK.to_string();
res.after.drives[i * self.set_drive_count + j].uuid = fm.erasure.this.to_string();
res.after.drives[i * self.set_drive_count + j].state = DRIVE_STATE_OK.to_string();
tmp_new_formats[i * self.set_drive_count + j] = Some(fm.clone());
}
}
ecstore/src/store.rs
+9 -13
View File
@@ -635,7 +635,7 @@ impl ECStore {
Err(to_object_err(Error::new(DiskError::FileNotFound), vec![bucket, object]))
}
async fn pools_with_object(&self, pools: &Vec<PoolObjInfo>, opts: &ObjectOptions) -> Vec<PoolErr> {
async fn pools_with_object(&self, pools: &[PoolObjInfo], opts: &ObjectOptions) -> Vec<PoolErr> {
let mut errs = Vec::new();
let pool_meta = self.pool_meta.read().await;
for pool in pools.iter() {
@@ -1183,7 +1183,7 @@ impl StorageAPI for ECStore {
}
let backend = self.backend_info().await;
StorageInfo { backend: backend, disks }
StorageInfo { backend, disks }
}
async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
@@ -1323,7 +1323,7 @@ impl StorageAPI for ECStore {
for obj in objects.iter() {
futures.push(async move {
self.internal_get_pool_info_existing_with_opts(
&bucket,
bucket,
&obj.object_name,
&ObjectOptions {
no_lock: true,
@@ -1381,18 +1381,14 @@ impl StorageAPI for ECStore {
if let Some(obj) = objects.get(i) {
if !pool_obj_idx_map.contains_key(&pinfo.index) {
pool_obj_idx_map.insert(pinfo.index, vec![obj.clone()]);
} else {
if let Some(val) = pool_obj_idx_map.get_mut(&pinfo.index) {
val.push(obj.clone());
}
} else if let Some(val) = pool_obj_idx_map.get_mut(&pinfo.index) {
val.push(obj.clone());
}
if !orig_index_map.contains_key(&pinfo.index) {
orig_index_map.insert(pinfo.index, vec![i]);
} else {
if let Some(val) = orig_index_map.get_mut(&pinfo.index) {
val.push(i);
}
} else if let Some(val) = orig_index_map.get_mut(&pinfo.index) {
val.push(i);
}
}
}
@@ -1934,8 +1930,8 @@ impl StorageAPI for ECStore {
}
r.disk_count += result.disk_count;
r.set_count += result.set_count;
r.before.append(&mut result.before);
r.after.append(&mut result.after);
r.before.drives.append(&mut result.before.drives);
r.after.drives.append(&mut result.after.drives);
}
if count_no_heal == self.pools.len() {
return Ok((r, Some(Error::new(DiskError::NoHealRequired))));
ecstore/src/store_err.rs
+13 -13
View File
@@ -84,14 +84,14 @@ pub fn to_object_err(err: Error, params: Vec<&str>) -> Error {
}
DiskError::FileNotFound => {
let bucket = params.get(0).cloned().unwrap_or_default().to_owned();
let object = params.get(1).cloned().map(|v| decode_dir_object(v)).unwrap_or_default();
let bucket = params.first().cloned().unwrap_or_default().to_owned();
let object = params.get(1).cloned().map(decode_dir_object).unwrap_or_default();
return Error::new(StorageError::ObjectNotFound(bucket, object));
}
DiskError::FileVersionNotFound => {
let bucket = params.get(0).cloned().unwrap_or_default().to_owned();
let object = params.get(1).cloned().map(|v| decode_dir_object(v)).unwrap_or_default();
let bucket = params.first().cloned().unwrap_or_default().to_owned();
let object = params.get(1).cloned().map(decode_dir_object).unwrap_or_default();
let version = params.get(2).cloned().unwrap_or_default().to_owned();
return Error::new(StorageError::VersionNotFound(bucket, object, version));
@@ -100,34 +100,34 @@ pub fn to_object_err(err: Error, params: Vec<&str>) -> Error {
return Error::new(StorageError::SlowDown);
}
DiskError::FileNameTooLong => {
let bucket = params.get(0).cloned().unwrap_or_default().to_owned();
let object = params.get(1).cloned().map(|v| decode_dir_object(v)).unwrap_or_default();
let bucket = params.first().cloned().unwrap_or_default().to_owned();
let object = params.get(1).cloned().map(decode_dir_object).unwrap_or_default();
return Error::new(StorageError::ObjectNameInvalid(bucket, object));
}
DiskError::VolumeExists => {
let bucket = params.get(0).cloned().unwrap_or_default().to_owned();
let bucket = params.first().cloned().unwrap_or_default().to_owned();
return Error::new(StorageError::BucketExists(bucket));
}
DiskError::IsNotRegular => {
let bucket = params.get(0).cloned().unwrap_or_default().to_owned();
let object = params.get(1).cloned().map(|v| decode_dir_object(v)).unwrap_or_default();
let bucket = params.first().cloned().unwrap_or_default().to_owned();
let object = params.get(1).cloned().map(decode_dir_object).unwrap_or_default();
return Error::new(StorageError::ObjectExistsAsDirectory(bucket, object));
}
DiskError::VolumeNotFound => {
let bucket = params.get(0).cloned().unwrap_or_default().to_owned();
let bucket = params.first().cloned().unwrap_or_default().to_owned();
return Error::new(StorageError::BucketNotFound(bucket));
}
DiskError::VolumeNotEmpty => {
let bucket = params.get(0).cloned().unwrap_or_default().to_owned();
let bucket = params.first().cloned().unwrap_or_default().to_owned();
return Error::new(StorageError::BucketNotEmpty(bucket));
}
DiskError::FileAccessDenied => {
let bucket = params.get(0).cloned().unwrap_or_default().to_owned();
let object = params.get(1).cloned().map(|v| decode_dir_object(v)).unwrap_or_default();
let bucket = params.first().cloned().unwrap_or_default().to_owned();
let object = params.get(1).cloned().map(decode_dir_object).unwrap_or_default();
return Error::new(StorageError::PrefixAccessDenied(bucket, object));
}
madmin/src/heal_command.rs
+2 -6
View File
@@ -37,11 +37,7 @@ pub struct BgHealState {
}
pub async fn get_local_background_heal_status() -> (BgHealState, bool) {
let (bg_seq, ok) = GLOBAL_BackgroundHealState
.read()
.await
.get_heal_sequence_by_token(BG_HEALING_UUID)
.await;
let (bg_seq, ok) = GLOBAL_BackgroundHealState.get_heal_sequence_by_token(BG_HEALING_UUID).await;
if !ok {
return (BgHealState::default(), false);
}
@@ -56,7 +52,7 @@ pub async fn get_local_background_heal_status() -> (BgHealState, bool) {
}
let Some(store) = new_object_layer_fn() else {
let healing = GLOBAL_BackgroundHealState.read().await.get_local_healing_disks().await;
let healing = GLOBAL_BackgroundHealState.get_local_healing_disks().await;
for disk in healing.values() {
status.heal_disks.push(disk.endpoint.clone());
}
rustfs/src/admin/handlers.rs
+183 -4
View File
@@ -1,13 +1,22 @@
use super::router::Operation;
use crate::storage::error::to_s3_error;
use bytes::Bytes;
use ecstore::bucket::policy::action::{Action, ActionSet};
use ecstore::bucket::policy::bucket_policy::{BPStatement, BucketPolicy};
use ecstore::bucket::policy::effect::Effect;
use ecstore::bucket::policy::resource::{Resource, ResourceSet};
use ecstore::error::Error as ec_Error;
use ecstore::global::GLOBAL_ALlHealState;
use ecstore::heal::heal_commands::HealOpts;
use ecstore::heal::heal_ops::new_heal_sequence;
use ecstore::peer::is_reserved_or_invalid_bucket;
use ecstore::store::is_valid_object_prefix;
use ecstore::store_api::StorageAPI;
use ecstore::utils::path::path_join;
use ecstore::utils::xml;
use ecstore::GLOBAL_Endpoints;
use ecstore::{new_object_layer_fn, store_api::BackendInfo};
use http::Uri;
use hyper::StatusCode;
use matchit::Params;
use s3s::S3ErrorCode;
@@ -18,8 +27,12 @@ use s3s::{
use serde::{Deserialize, Serialize};
use serde_urlencoded::from_bytes;
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use time::{Duration, OffsetDateTime};
use tracing::warn;
use tokio::spawn;
use tokio::sync::mpsc;
use tracing::{info, warn};
#[derive(Deserialize, Debug, Default)]
#[serde(rename_all = "PascalCase", default)]
@@ -224,14 +237,168 @@ impl Operation for MetricsHandler {
}
}
#[derive(Debug, Default, Serialize, Deserialize)]
struct HealInitParams {
bucket: String,
obj_prefix: String,
hs: HealOpts,
client_token: String,
force_start: bool,
force_stop: bool,
}
fn extract_heal_init_params(body: &Bytes, uri: &Uri, params: Params<'_, '_>) -> S3Result<HealInitParams> {
let mut hip = HealInitParams {
bucket: params.get("bucket").map(|s| s.to_string()).unwrap_or_default(),
obj_prefix: params.get("prefix").map(|s| s.to_string()).unwrap_or_default(),
..Default::default()
};
if hip.bucket.is_empty() && !hip.obj_prefix.is_empty() {
return Err(s3_error!(InvalidRequest, "invalid bucket name"));
}
if is_reserved_or_invalid_bucket(&hip.bucket, false) {
return Err(s3_error!(InvalidRequest, "invalid bucket name"));
}
if !is_valid_object_prefix(&hip.obj_prefix) {
return Err(s3_error!(InvalidRequest, "invalid object name"));
}
if let Some(query) = uri.query() {
let params: Vec<&str> = query.split('&').collect();
for param in params {
let mut parts = param.split('=');
if let Some(key) = parts.next() {
if key == "clientToken" {
if let Some(value) = parts.next() {
hip.client_token = value.to_string();
}
}
if key == "forceStart" {
if parts.next().is_some() {
hip.force_start = true;
}
}
if key == "forceStop" {
if parts.next().is_some() {
hip.force_stop = true;
}
}
}
}
}
if (hip.force_start && hip.force_stop) || (!hip.client_token.is_empty() && (hip.force_start || hip.force_stop)) {
return Err(s3_error!(InvalidRequest, ""));
}
if hip.client_token.is_empty() {
hip.hs = serde_json::from_slice(body).map_err(|e| {
info!("err request body parse, err: {:?}", e);
s3_error!(InvalidRequest, "err request body parse")
})?;
}
Ok(hip)
}
pub struct HealHandler {}
#[async_trait::async_trait]
impl Operation for HealHandler {
async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
warn!("handle HealHandler");
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
warn!("handle HealHandler, req: {:?}, params: {:?}", req, params);
let Some(cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")) };
info!("cred: {:?}", cred);
let mut input = req.input;
let bytes = match input.store_all_unlimited().await {
Ok(b) => b,
Err(e) => {
warn!("get body failed, e: {:?}", e);
return Err(s3_error!(InvalidRequest, "get body failed"));
}
};
info!("bytes: {:?}", bytes);
let hip = extract_heal_init_params(&bytes, &req.uri, params)?;
info!("body: {:?}", hip);
return Err(s3_error!(NotImplemented));
#[derive(Default)]
struct HealResp {
resp_bytes: Vec<u8>,
_api_err: Option<ec_Error>,
_err_body: String,
}
let heap_path = path_join(&[PathBuf::from(hip.bucket.clone()), PathBuf::from(hip.obj_prefix.clone())]);
if !hip.client_token.is_empty() && !hip.force_start && !hip.force_stop {
match GLOBAL_ALlHealState
.pop_heal_status_json(heap_path.to_str().unwrap_or_default(), &hip.client_token)
.await
{
Ok(b) => {
info!("pop_heal_status_json success");
return Ok(S3Response::new((StatusCode::OK, Body::from(b))));
}
Err(_e) => {
info!("pop_heal_status_json failed");
return Ok(S3Response::new((StatusCode::INTERNAL_SERVER_ERROR, Body::from(vec![]))));
}
}
}
let (tx, mut rx) = mpsc::channel(1);
if hip.force_stop {
let tx_clone = tx.clone();
spawn(async move {
match GLOBAL_ALlHealState
.stop_heal_sequence(heap_path.to_str().unwrap_or_default())
.await
{
Ok(b) => {
let _ = tx_clone
.send(HealResp {
resp_bytes: b,
..Default::default()
})
.await;
}
Err(e) => {
let _ = tx_clone
.send(HealResp {
_api_err: Some(e),
..Default::default()
})
.await;
}
}
});
} else if hip.client_token.is_empty() {
let nh = Arc::new(new_heal_sequence(&hip.bucket, &hip.obj_prefix, "", hip.hs, hip.force_start));
let tx_clone = tx.clone();
spawn(async move {
match GLOBAL_ALlHealState.launch_new_heal_sequence(nh).await {
Ok(b) => {
let _ = tx_clone
.send(HealResp {
resp_bytes: b,
..Default::default()
})
.await;
}
Err(e) => {
let _ = tx_clone
.send(HealResp {
_api_err: Some(e),
..Default::default()
})
.await;
}
}
});
}
match rx.recv().await {
Some(result) => Ok(S3Response::new((StatusCode::OK, Body::from(result.resp_bytes)))),
None => Ok(S3Response::new((StatusCode::INTERNAL_SERVER_ERROR, Body::from(vec![])))),
}
}
}
@@ -516,3 +683,15 @@ impl Operation for RebalanceStop {
return Err(s3_error!(NotImplemented));
}
}
#[cfg(test)]
mod test {
use ecstore::heal::heal_commands::HealOpts;
#[test]
fn test_decode() {
let b = b"{\"recursive\":false,\"dryRun\":false,\"remove\":false,\"recreate\":false,\"scanMode\":1,\"updateParity\":false,\"nolock\":false}";
let s: HealOpts = serde_urlencoded::from_bytes(b).unwrap();
println!("{:?}", s);
}
}
rustfs/src/admin/mod.rs
+1 -1
View File
@@ -2,7 +2,7 @@ pub mod handlers;
pub mod router;
use common::error::Result;
use ecstore::global::{is_dist_erasure, is_erasure};
// use ecstore::global::{is_dist_erasure, is_erasure};
use hyper::Method;
use router::{AdminOperation, S3Router};
use s3s::route::S3Route;