mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-27 16:48:58 +00:00
weisd
Generated
+41
-10
@@ -678,7 +678,7 @@ dependencies = [
|
||||
"sha2",
|
||||
"siphasher",
|
||||
"tempfile",
|
||||
"thiserror",
|
||||
"thiserror 1.0.68",
|
||||
"time",
|
||||
"tokio",
|
||||
"tokio-stream",
|
||||
@@ -1976,7 +1976,7 @@ checksum = "a3a7c64d9bf75b1b8d981124c14c179074e8caa7dfe7b6a12e6222ddcd0c8f72"
|
||||
dependencies = [
|
||||
"once_cell",
|
||||
"protobuf-support",
|
||||
"thiserror",
|
||||
"thiserror 1.0.68",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1985,7 +1985,7 @@ version = "3.7.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b088fd20b938a875ea00843b6faf48579462630015c3788d397ad6a786663252"
|
||||
dependencies = [
|
||||
"thiserror",
|
||||
"thiserror 1.0.68",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -2018,7 +2018,7 @@ dependencies = [
|
||||
"num_cpus",
|
||||
"once_cell",
|
||||
"platforms",
|
||||
"thiserror",
|
||||
"thiserror 1.0.68",
|
||||
"unescape",
|
||||
]
|
||||
|
||||
@@ -2082,7 +2082,7 @@ dependencies = [
|
||||
"pin-project-lite",
|
||||
"s3s",
|
||||
"sha2",
|
||||
"thiserror",
|
||||
"thiserror 1.0.68",
|
||||
"tokio",
|
||||
"tracing",
|
||||
]
|
||||
@@ -2332,7 +2332,7 @@ checksum = "f3cb5ba0dc43242ce17de99c180e96db90b235b8a9fdc9543c96d2209116bd9f"
|
||||
[[package]]
|
||||
name = "s3s"
|
||||
version = "0.11.0-dev"
|
||||
source = "git+https://github.com/Nugine/s3s.git?rev=207170f526c75a8190e8f9afadf961909fd01d34#207170f526c75a8190e8f9afadf961909fd01d34"
|
||||
source = "git+https://github.com/Nugine/s3s.git?rev=3291c0ca0284971569499cbe75bd69ef7bde8321#3291c0ca0284971569499cbe75bd69ef7bde8321"
|
||||
dependencies = [
|
||||
"arrayvec",
|
||||
"async-trait",
|
||||
@@ -2364,8 +2364,9 @@ dependencies = [
|
||||
"sha1",
|
||||
"sha2",
|
||||
"smallvec",
|
||||
"std-next",
|
||||
"sync_wrapper 1.0.1",
|
||||
"thiserror",
|
||||
"thiserror 2.0.3",
|
||||
"time",
|
||||
"tokio",
|
||||
"tower 0.5.1",
|
||||
@@ -2378,12 +2379,12 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "s3s-policy"
|
||||
version = "0.11.0-dev"
|
||||
source = "git+https://github.com/Nugine/s3s.git?rev=207170f526c75a8190e8f9afadf961909fd01d34#207170f526c75a8190e8f9afadf961909fd01d34"
|
||||
source = "git+https://github.com/Nugine/s3s.git?rev=3291c0ca0284971569499cbe75bd69ef7bde8321#3291c0ca0284971569499cbe75bd69ef7bde8321"
|
||||
dependencies = [
|
||||
"indexmap 2.6.0",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"thiserror",
|
||||
"thiserror 2.0.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -2560,6 +2561,16 @@ version = "1.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3"
|
||||
|
||||
[[package]]
|
||||
name = "std-next"
|
||||
version = "0.1.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "87854dde78837ff867561b22d873c53c25938d4b152afba2fa6fc48cb406c099"
|
||||
dependencies = [
|
||||
"simdutf8",
|
||||
"thiserror 2.0.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "strsim"
|
||||
version = "0.11.1"
|
||||
@@ -2625,7 +2636,16 @@ version = "1.0.68"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "02dd99dc800bbb97186339685293e1cc5d9df1f8fae2d0aecd9ff1c77efea892"
|
||||
dependencies = [
|
||||
"thiserror-impl",
|
||||
"thiserror-impl 1.0.68",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "thiserror"
|
||||
version = "2.0.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c006c85c7651b3cf2ada4584faa36773bd07bac24acfb39f3c431b36d7e667aa"
|
||||
dependencies = [
|
||||
"thiserror-impl 2.0.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -2639,6 +2659,17 @@ dependencies = [
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "thiserror-impl"
|
||||
version = "2.0.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f077553d607adc1caf65430528a576c757a71ed73944b66ebb58ef2bbd243568"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "thread_local"
|
||||
version = "1.1.8"
|
||||
|
||||
+3
-3
@@ -10,7 +10,7 @@ members = [
|
||||
"common/protos",
|
||||
"api/admin",
|
||||
"reader",
|
||||
"common/workers",
|
||||
"common/workers",
|
||||
]
|
||||
|
||||
[workspace.package]
|
||||
@@ -55,10 +55,10 @@ protos = { path = "./common/protos" }
|
||||
rand = "0.8.5"
|
||||
rmp = "0.8.14"
|
||||
rmp-serde = "1.3.0"
|
||||
s3s = { git = "https://github.com/Nugine/s3s.git", rev = "207170f526c75a8190e8f9afadf961909fd01d34", default-features = true, features = [
|
||||
s3s = { git = "https://github.com/Nugine/s3s.git", rev = "3291c0ca0284971569499cbe75bd69ef7bde8321", default-features = true, features = [
|
||||
"tower",
|
||||
] }
|
||||
s3s-policy = { git = "https://github.com/Nugine/s3s.git", rev = "207170f526c75a8190e8f9afadf961909fd01d34" }
|
||||
s3s-policy = { git = "https://github.com/Nugine/s3s.git", rev = "3291c0ca0284971569499cbe75bd69ef7bde8321" }
|
||||
serde = { version = "1.0.214", features = ["derive"] }
|
||||
serde_json = "1.0.132"
|
||||
tempfile = "3.13.0"
|
||||
|
||||
@@ -1,24 +1,25 @@
|
||||
use s3s::dto::{BucketVersioningStatus, VersioningConfiguration};
|
||||
|
||||
use crate::utils::wildcard;
|
||||
|
||||
pub trait VersioningApi {
|
||||
fn enabled(&self) -> bool;
|
||||
fn prefix_enabled(&self, prefix: &str) -> bool;
|
||||
fn prefix_suspended(&self, prefix: &str) -> bool;
|
||||
fn versioned(&self, prefix: &str) -> bool;
|
||||
fn suspended(&self) -> bool;
|
||||
}
|
||||
|
||||
impl VersioningApi for VersioningConfiguration {
|
||||
fn enabled(&self) -> bool {
|
||||
self.status
|
||||
.as_ref()
|
||||
.is_some_and(|v| v.as_str() == BucketVersioningStatus::ENABLED)
|
||||
self.status == Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED))
|
||||
}
|
||||
fn suspended(&self) -> bool {
|
||||
self.status == Some(BucketVersioningStatus::from_static(BucketVersioningStatus::SUSPENDED))
|
||||
}
|
||||
|
||||
fn prefix_enabled(&self, prefix: &str) -> bool {
|
||||
if !self
|
||||
.status
|
||||
.as_ref()
|
||||
.is_some_and(|v| v.as_str() == BucketVersioningStatus::ENABLED)
|
||||
{
|
||||
if self.status == Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -26,27 +27,56 @@ impl VersioningApi for VersioningConfiguration {
|
||||
return true;
|
||||
}
|
||||
|
||||
// TODO: ExcludeFolders
|
||||
if let Some(exclude_folders) = self.exclude_folders {
|
||||
if exclude_folders && prefix.ends_with('/') {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(ref excluded_prefixes) = self.excluded_prefixes {
|
||||
for p in excluded_prefixes.iter() {
|
||||
if let Some(ref sprefix) = p.prefix {
|
||||
let pattern = format!("{}*", sprefix);
|
||||
if wildcard::match_simple(&pattern, prefix) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
true
|
||||
}
|
||||
fn prefix_suspended(&self, prefix: &str) -> bool {
|
||||
if self
|
||||
.status
|
||||
.as_ref()
|
||||
.is_some_and(|v| v.as_str() == BucketVersioningStatus::SUSPENDED)
|
||||
{
|
||||
if self.status == Some(BucketVersioningStatus::from_static(BucketVersioningStatus::SUSPENDED)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if let Some(status) = self.status.as_ref() {
|
||||
if status.as_str() == BucketVersioningStatus::ENABLED && prefix.is_empty() {
|
||||
if self.status == Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)) {
|
||||
if prefix.is_empty() {
|
||||
return false;
|
||||
}
|
||||
|
||||
// TODO: ExcludeFolders
|
||||
if let Some(exclude_folders) = self.exclude_folders {
|
||||
if exclude_folders && prefix.ends_with('/') {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(ref excluded_prefixes) = self.excluded_prefixes {
|
||||
for p in excluded_prefixes.iter() {
|
||||
if let Some(ref sprefix) = p.prefix {
|
||||
let pattern = format!("{}*", sprefix);
|
||||
if wildcard::match_simple(&pattern, prefix) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
false
|
||||
}
|
||||
fn versioned(&self, prefix: &str) -> bool {
|
||||
self.prefix_enabled(prefix) || self.prefix_suspended(prefix)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,6 +7,9 @@ use tracing::warn;
|
||||
pub struct BucketVersioningSys {}
|
||||
|
||||
impl BucketVersioningSys {
|
||||
pub fn new() -> Self {
|
||||
Self {}
|
||||
}
|
||||
pub async fn enabled(bucket: &str) -> bool {
|
||||
match Self::get(bucket).await {
|
||||
Ok(res) => res.enabled(),
|
||||
@@ -27,15 +30,15 @@ impl BucketVersioningSys {
|
||||
}
|
||||
}
|
||||
|
||||
// pub async fn suspended(bucket: &str) -> bool {
|
||||
// match Self::get(bucket).await {
|
||||
// Ok(res) => res.suspended(),
|
||||
// Err(err) => {
|
||||
// warn!("{:?}", err);
|
||||
// false
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
pub async fn suspended(bucket: &str) -> bool {
|
||||
match Self::get(bucket).await {
|
||||
Ok(res) => res.suspended(),
|
||||
Err(err) => {
|
||||
warn!("{:?}", err);
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn prefix_suspended(bucket: &str, prefix: &str) -> bool {
|
||||
match Self::get(bucket).await {
|
||||
|
||||
@@ -6,8 +6,7 @@ use super::{storageclass, Config, GLOBAL_StorageClass, KVS};
|
||||
use crate::config::error::is_not_found;
|
||||
use crate::disk::RUSTFS_META_BUCKET;
|
||||
use crate::error::{Error, Result};
|
||||
use crate::store::ECStore;
|
||||
use crate::store_api::{HTTPRangeSpec, ObjectIO, ObjectInfo, ObjectOptions, PutObjReader, StorageAPI};
|
||||
use crate::store_api::{HTTPRangeSpec, ObjectInfo, ObjectOptions, PutObjReader, StorageAPI};
|
||||
use crate::store_err::is_err_object_not_found;
|
||||
use crate::utils::path::SLASH_SEPARATOR;
|
||||
use http::HeaderMap;
|
||||
@@ -30,13 +29,17 @@ lazy_static! {
|
||||
h
|
||||
};
|
||||
}
|
||||
pub async fn read_config(api: Arc<ECStore>, file: &str) -> Result<Vec<u8>> {
|
||||
pub async fn read_config<S: StorageAPI>(api: Arc<S>, file: &str) -> Result<Vec<u8>> {
|
||||
let (data, _obj) = read_config_with_metadata(api, file, &ObjectOptions::default()).await?;
|
||||
|
||||
Ok(data)
|
||||
}
|
||||
|
||||
async fn read_config_with_metadata(api: Arc<ECStore>, file: &str, opts: &ObjectOptions) -> Result<(Vec<u8>, ObjectInfo)> {
|
||||
async fn read_config_with_metadata<S: StorageAPI>(
|
||||
api: Arc<S>,
|
||||
file: &str,
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<(Vec<u8>, ObjectInfo)> {
|
||||
let range = HTTPRangeSpec::nil();
|
||||
let h = HeaderMap::new();
|
||||
let mut rd = api
|
||||
@@ -59,7 +62,7 @@ async fn read_config_with_metadata(api: Arc<ECStore>, file: &str, opts: &ObjectO
|
||||
Ok((data, rd.object_info))
|
||||
}
|
||||
|
||||
pub async fn save_config(api: Arc<ECStore>, file: &str, data: &[u8]) -> Result<()> {
|
||||
pub async fn save_config<S: StorageAPI>(api: Arc<S>, file: &str, data: &[u8]) -> Result<()> {
|
||||
save_config_with_opts(
|
||||
api,
|
||||
file,
|
||||
@@ -72,7 +75,7 @@ pub async fn save_config(api: Arc<ECStore>, file: &str, data: &[u8]) -> Result<(
|
||||
.await
|
||||
}
|
||||
|
||||
async fn save_config_with_opts(api: Arc<ECStore>, file: &str, data: &[u8], opts: &ObjectOptions) -> Result<()> {
|
||||
async fn save_config_with_opts<S: StorageAPI>(api: Arc<S>, file: &str, data: &[u8], opts: &ObjectOptions) -> Result<()> {
|
||||
let _ = api
|
||||
.put_object(
|
||||
RUSTFS_META_BUCKET,
|
||||
@@ -88,7 +91,7 @@ fn new_server_config() -> Config {
|
||||
Config::new()
|
||||
}
|
||||
|
||||
async fn new_and_save_server_config(api: Arc<ECStore>) -> Result<Config> {
|
||||
async fn new_and_save_server_config<S: StorageAPI>(api: Arc<S>) -> Result<Config> {
|
||||
let mut cfg = new_server_config();
|
||||
lookup_configs(&mut cfg, api.clone()).await;
|
||||
save_server_config(api, &cfg).await?;
|
||||
@@ -96,7 +99,7 @@ async fn new_and_save_server_config(api: Arc<ECStore>) -> Result<Config> {
|
||||
Ok(cfg)
|
||||
}
|
||||
|
||||
pub async fn read_config_without_migrate(api: Arc<ECStore>) -> Result<Config> {
|
||||
pub async fn read_config_without_migrate<S: StorageAPI>(api: Arc<S>) -> Result<Config> {
|
||||
let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE);
|
||||
let data = match read_config(api.clone(), config_file.as_str()).await {
|
||||
Ok(res) => res,
|
||||
@@ -116,7 +119,7 @@ pub async fn read_config_without_migrate(api: Arc<ECStore>) -> Result<Config> {
|
||||
read_server_config(api, data.as_slice()).await
|
||||
}
|
||||
|
||||
async fn read_server_config(api: Arc<ECStore>, data: &[u8]) -> Result<Config> {
|
||||
async fn read_server_config<S: StorageAPI>(api: Arc<S>, data: &[u8]) -> Result<Config> {
|
||||
let cfg = {
|
||||
if data.is_empty() {
|
||||
let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE);
|
||||
@@ -145,7 +148,7 @@ async fn read_server_config(api: Arc<ECStore>, data: &[u8]) -> Result<Config> {
|
||||
Ok(cfg.merge())
|
||||
}
|
||||
|
||||
async fn save_server_config(api: Arc<ECStore>, cfg: &Config) -> Result<()> {
|
||||
async fn save_server_config<S: StorageAPI>(api: Arc<S>, cfg: &Config) -> Result<()> {
|
||||
let data = cfg.marshal()?;
|
||||
|
||||
let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE);
|
||||
@@ -153,14 +156,14 @@ async fn save_server_config(api: Arc<ECStore>, cfg: &Config) -> Result<()> {
|
||||
save_config(api, &config_file, data.as_slice()).await
|
||||
}
|
||||
|
||||
pub async fn lookup_configs(cfg: &mut Config, api: Arc<ECStore>) {
|
||||
pub async fn lookup_configs<S: StorageAPI>(cfg: &mut Config, api: Arc<S>) {
|
||||
// TODO: from etcd
|
||||
if let Err(err) = apply_dynamic_config(cfg, api).await {
|
||||
error!("apply_dynamic_config err {:?}", &err);
|
||||
}
|
||||
}
|
||||
|
||||
async fn apply_dynamic_config(cfg: &mut Config, api: Arc<ECStore>) -> Result<()> {
|
||||
async fn apply_dynamic_config<S: StorageAPI>(cfg: &mut Config, api: Arc<S>) -> Result<()> {
|
||||
for key in SubSystemsDynamic.iter() {
|
||||
apply_dynamic_config_for_sub_sys(cfg, api.clone(), key).await?;
|
||||
}
|
||||
@@ -168,7 +171,7 @@ async fn apply_dynamic_config(cfg: &mut Config, api: Arc<ECStore>) -> Result<()>
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn apply_dynamic_config_for_sub_sys(cfg: &mut Config, api: Arc<ECStore>, subsys: &str) -> Result<()> {
|
||||
async fn apply_dynamic_config_for_sub_sys<S: StorageAPI>(cfg: &mut Config, api: Arc<S>, subsys: &str) -> Result<()> {
|
||||
let set_drive_counts = api.set_drive_counts();
|
||||
if subsys == STORAGE_CLASS_SUB_SYS {
|
||||
let kvs = match cfg.get_value(STORAGE_CLASS_SUB_SYS, DEFAULT_KV_KEY) {
|
||||
|
||||
+451
-9
@@ -1,16 +1,22 @@
|
||||
use crate::config::common::{read_config, save_config};
|
||||
use crate::bucket::versioning_sys::BucketVersioningSys;
|
||||
use crate::config::common::{read_config, save_config, CONFIG_PREFIX};
|
||||
use crate::config::error::ConfigError;
|
||||
use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
|
||||
use crate::error::{Error, Result};
|
||||
use crate::heal::heal_commands::HealOpts;
|
||||
use crate::new_object_layer_fn;
|
||||
use crate::store_api::{StorageAPI, StorageDisk, StorageInfo};
|
||||
use crate::store_err::StorageError;
|
||||
use crate::store_api::{BucketOptions, MakeBucketOptions, StorageAPI, StorageDisk, StorageInfo};
|
||||
use crate::store_err::{is_err_bucket_exists, StorageError};
|
||||
use crate::utils::path::{path_join, SLASH_SEPARATOR};
|
||||
use crate::{sets::Sets, store::ECStore};
|
||||
use byteorder::{ByteOrder, LittleEndian, WriteBytesExt};
|
||||
use rmp_serde::{Deserializer, Serializer};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::io::{Cursor, Write};
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use time::OffsetDateTime;
|
||||
use tracing::{error, info, warn};
|
||||
|
||||
pub const POOL_META_NAME: &str = "pool.bin";
|
||||
pub const POOL_META_FORMAT: u16 = 1;
|
||||
@@ -96,7 +102,7 @@ impl PoolMeta {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn save(&self, _pools: Vec<Arc<Sets>>) -> Result<()> {
|
||||
pub async fn save(&self, pools: Vec<Arc<Sets>>) -> Result<()> {
|
||||
if self.dont_save {
|
||||
return Ok(());
|
||||
}
|
||||
@@ -107,12 +113,11 @@ impl PoolMeta {
|
||||
self.serialize(&mut Serializer::new(&mut buf))?;
|
||||
data.write_all(&buf)?;
|
||||
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
return Err(Error::from_string("errServerNotInitialized".to_string()));
|
||||
};
|
||||
for pool in pools {
|
||||
save_config(pool, &POOL_META_NAME, &data).await?;
|
||||
}
|
||||
|
||||
// FIXME:
|
||||
save_config(store, &POOL_META_NAME, &data).await
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn decommission_cancel(&mut self, idx: usize) -> bool {
|
||||
@@ -139,6 +144,150 @@ impl PoolMeta {
|
||||
false
|
||||
}
|
||||
}
|
||||
pub fn decommission_failed(&mut self, idx: usize) -> bool {
|
||||
if let Some(stats) = self.pools.get_mut(idx) {
|
||||
if let Some(d) = &stats.decommission {
|
||||
if !d.failed {
|
||||
stats.last_update = OffsetDateTime::now_utc();
|
||||
|
||||
let mut pd = d.clone();
|
||||
pd.start_time = None;
|
||||
pd.canceled = false;
|
||||
pd.failed = true;
|
||||
pd.complete = false;
|
||||
|
||||
stats.decommission = Some(pd);
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
} else {
|
||||
false
|
||||
}
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
pub fn decommission_complete(&mut self, idx: usize) -> bool {
|
||||
if let Some(stats) = self.pools.get_mut(idx) {
|
||||
if let Some(d) = &stats.decommission {
|
||||
if !d.complete {
|
||||
stats.last_update = OffsetDateTime::now_utc();
|
||||
|
||||
let mut pd = d.clone();
|
||||
pd.start_time = None;
|
||||
pd.canceled = false;
|
||||
pd.failed = false;
|
||||
pd.complete = true;
|
||||
|
||||
stats.decommission = Some(pd);
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
} else {
|
||||
false
|
||||
}
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
pub fn decommission(&mut self, idx: usize, pi: PoolSpaceInfo) -> Result<()> {
|
||||
if let Some(pool) = self.pools.get_mut(idx) {
|
||||
if let Some(ref info) = pool.decommission {
|
||||
if !info.complete && !info.failed && !info.canceled {
|
||||
return Err(Error::msg("DecommissionAlreadyRunning"));
|
||||
}
|
||||
}
|
||||
|
||||
let now = OffsetDateTime::now_utc();
|
||||
pool.last_update = now.clone();
|
||||
pool.decommission = Some(PoolDecommissionInfo {
|
||||
start_time: Some(now),
|
||||
start_size: pi.free,
|
||||
total_size: pi.total,
|
||||
current_size: pi.free,
|
||||
..Default::default()
|
||||
});
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
pub fn queue_buckets(&mut self, idx: usize, bks: Vec<DecomBucketInfo>) {
|
||||
for bk in bks.iter() {
|
||||
if let Some(dec) = self.pools[idx].decommission.as_mut() {
|
||||
dec.bucket_push(bk);
|
||||
}
|
||||
}
|
||||
}
|
||||
pub fn pending_buckets(&self, idx: usize) -> Vec<DecomBucketInfo> {
|
||||
let mut list = Vec::new();
|
||||
|
||||
if let Some(pool) = self.pools.get(idx) {
|
||||
if let Some(ref info) = pool.decommission {
|
||||
for bk in info.queued_buckets.iter() {
|
||||
let (name, prefix) = path2_bucket_object(bk);
|
||||
list.push(DecomBucketInfo { name, prefix });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
list
|
||||
}
|
||||
|
||||
pub fn is_bucket_decommissioned(&self, idx: usize, bucket: String) -> bool {
|
||||
if let Some(ref info) = self.pools[idx].decommission {
|
||||
info.is_bucket_decommissioned(&bucket)
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
pub fn bucket_done(&mut self, idx: usize, bucket: String) -> bool {
|
||||
if let Some(pool) = self.pools.get_mut(idx) {
|
||||
if let Some(info) = pool.decommission.as_mut() {
|
||||
info.bucket_pop(&bucket)
|
||||
} else {
|
||||
false
|
||||
}
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
pub fn count_item(&mut self, idx: usize, size: usize, failed: bool) {
|
||||
if let Some(pool) = self.pools.get_mut(idx) {
|
||||
if let Some(info) = pool.decommission.as_mut() {
|
||||
if failed {
|
||||
info.items_decommission_failed += 1;
|
||||
info.bytes_failed += size;
|
||||
} else {
|
||||
info.items_decommissioned += 1;
|
||||
info.bytes_done += size;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn path2_bucket_object(name: &String) -> (String, String) {
|
||||
path2_bucket_object_with_base_path("", &name)
|
||||
}
|
||||
|
||||
fn path2_bucket_object_with_base_path(base_path: &str, path: &str) -> (String, String) {
|
||||
// Trim the base path and leading slash
|
||||
let trimmed_path = path
|
||||
.strip_prefix(base_path)
|
||||
.unwrap_or(path)
|
||||
.strip_prefix(SLASH_SEPARATOR)
|
||||
.unwrap_or(path);
|
||||
// Find the position of the first '/'
|
||||
let pos = trimmed_path.find(SLASH_SEPARATOR).unwrap_or(trimmed_path.len());
|
||||
// Split into bucket and prefix
|
||||
let bucket = &trimmed_path[0..pos];
|
||||
let prefix = &trimmed_path[pos + 1..]; // +1 to skip the '/' character if it exists
|
||||
|
||||
(bucket.to_string(), prefix.to_string())
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
|
||||
@@ -162,6 +311,56 @@ pub struct PoolDecommissionInfo {
|
||||
pub bytes_failed: usize,
|
||||
}
|
||||
|
||||
impl PoolDecommissionInfo {
|
||||
pub fn bucket_push(&mut self, bucket: &DecomBucketInfo) {
|
||||
for b in self.queued_buckets.iter() {
|
||||
if self.is_bucket_decommissioned(b) {
|
||||
return;
|
||||
}
|
||||
|
||||
if b == &bucket.to_string() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
self.queued_buckets.push(bucket.to_string());
|
||||
|
||||
self.bucket = bucket.name.clone();
|
||||
self.prefix = bucket.prefix.clone();
|
||||
}
|
||||
pub fn is_bucket_decommissioned(&self, bucket: &String) -> bool {
|
||||
for b in self.decommissioned_buckets.iter() {
|
||||
if b == bucket {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
pub fn bucket_pop(&mut self, bucket: &String) -> bool {
|
||||
self.decommissioned_buckets.push(bucket.clone());
|
||||
|
||||
let mut found = None;
|
||||
for (i, b) in self.queued_buckets.iter().enumerate() {
|
||||
if b == bucket {
|
||||
found = Some(i.clone());
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(i) = found {
|
||||
self.queued_buckets.remove(i);
|
||||
if &self.bucket == bucket {
|
||||
self.bucket = "".to_owned();
|
||||
self.prefix = "".to_owned();
|
||||
self.object = "".to_owned();
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct PoolSpaceInfo {
|
||||
pub free: usize,
|
||||
@@ -169,6 +368,20 @@ pub struct PoolSpaceInfo {
|
||||
pub used: usize,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Clone)]
|
||||
pub struct DecomBucketInfo {
|
||||
pub name: String,
|
||||
pub prefix: String,
|
||||
}
|
||||
|
||||
impl ToString for DecomBucketInfo {
|
||||
fn to_string(&self) -> String {
|
||||
path_join(&[PathBuf::from(self.name.clone()), PathBuf::from(self.prefix.clone())])
|
||||
.to_string_lossy()
|
||||
.to_string()
|
||||
}
|
||||
}
|
||||
|
||||
impl ECStore {
|
||||
pub async fn status(&self, idx: usize) -> Result<PoolStatus> {
|
||||
let space_info = self.get_decommission_pool_space_info(idx).await?;
|
||||
@@ -228,6 +441,235 @@ impl ECStore {
|
||||
|
||||
Ok(())
|
||||
}
|
||||
pub async fn is_decommission_running(&self) -> bool {
|
||||
let pool_meta = self.pool_meta.read().await;
|
||||
for pool in pool_meta.pools.iter() {
|
||||
if let Some(ref info) = pool.decommission {
|
||||
if !info.complete && !info.failed && !info.canceled {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
false
|
||||
}
|
||||
|
||||
pub async fn decommission(&self, indices: Vec<usize>) -> Result<()> {
|
||||
if indices.is_empty() {
|
||||
return Err(Error::msg("errInvalidArgument"));
|
||||
}
|
||||
|
||||
if self.single_pool() {
|
||||
return Err(Error::msg("errInvalidArgument"));
|
||||
}
|
||||
|
||||
self.start_decommission(indices.clone()).await?;
|
||||
|
||||
tokio::spawn(async move {
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
error!("store not init");
|
||||
return;
|
||||
};
|
||||
for idx in indices.iter() {
|
||||
store.do_decommission_in_routine(idx.clone()).await;
|
||||
}
|
||||
});
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
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 {
|
||||
let versioning = BucketVersioningSys::get(&bi.name).await?;
|
||||
_vc = Some(versioning);
|
||||
}
|
||||
|
||||
// FIXME:
|
||||
unimplemented!()
|
||||
}
|
||||
|
||||
async fn do_decommission_in_routine(&self, idx: usize) {
|
||||
if let Err(err) = self.decommission_in_background(idx).await {
|
||||
error!("decom err {:?}", &err);
|
||||
if let Err(er) = self.decommission_failed(idx).await {
|
||||
error!("decom failed err {:?}", &er);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
let (failed, cmd_line) = {
|
||||
let pool_meta = self.pool_meta.read().await;
|
||||
let failed = {
|
||||
if let Some(info) = &pool_meta.pools[idx].decommission {
|
||||
info.items_decommission_failed > 0
|
||||
} else {
|
||||
false
|
||||
}
|
||||
};
|
||||
let cmd_line = pool_meta.pools[idx].cmd_line.clone();
|
||||
(failed, cmd_line)
|
||||
};
|
||||
|
||||
if !failed {
|
||||
warn!("Decommissioning complete for pool {}, verifying for any pending objects", cmd_line);
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn decommission_failed(&self, idx: usize) -> Result<()> {
|
||||
if self.single_pool() {
|
||||
return Err(Error::msg("errInvalidArgument"));
|
||||
}
|
||||
|
||||
let mut pool_meta = self.pool_meta.write().await;
|
||||
if pool_meta.decommission_failed(idx) {
|
||||
pool_meta.save(self.pools.clone()).await?;
|
||||
// FIXME: globalNotificationSys.ReloadPoolMeta(ctx)
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn complete_decommission(&self, idx: usize) -> Result<()> {
|
||||
if self.single_pool() {
|
||||
return Err(Error::msg("errInvalidArgument"));
|
||||
}
|
||||
|
||||
let mut pool_meta = self.pool_meta.write().await;
|
||||
if pool_meta.decommission_complete(idx) {
|
||||
pool_meta.save(self.pools.clone()).await?;
|
||||
// FIXME: globalNotificationSys.ReloadPoolMeta(ctx)
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn decommission_in_background(&self, idx: usize) -> Result<()> {
|
||||
let pool = self.pools[idx].clone();
|
||||
|
||||
let pending = {
|
||||
let pool_meta = self.pool_meta.read().await;
|
||||
pool_meta.pending_buckets(idx)
|
||||
};
|
||||
|
||||
for bucket in pending.iter() {
|
||||
let is_decommissioned = {
|
||||
let pool_meta = self.pool_meta.read().await;
|
||||
pool_meta.is_bucket_decommissioned(idx, bucket.to_string())
|
||||
};
|
||||
|
||||
if is_decommissioned {
|
||||
info!("decommission: already done, moving on {}", bucket.to_string());
|
||||
|
||||
{
|
||||
let mut pool_meta = self.pool_meta.write().await;
|
||||
if pool_meta.bucket_done(idx, bucket.to_string()) {
|
||||
if let Err(err) = pool_meta.save(self.pools.clone()).await {
|
||||
error!("decom pool_meta.save err {:?}", err);
|
||||
}
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
info!("decommission: currently on bucket {}", &bucket.name);
|
||||
|
||||
if let Err(err) = self.decommission_pool(idx, pool.clone(), bucket.clone()).await {
|
||||
error!("decommission: decommission_pool err {:?}", &err);
|
||||
return Err(err);
|
||||
}
|
||||
|
||||
{
|
||||
let mut pool_meta = self.pool_meta.write().await;
|
||||
if pool_meta.bucket_done(idx, bucket.to_string()) {
|
||||
if let Err(err) = pool_meta.save(self.pools.clone()).await {
|
||||
error!("decom pool_meta.save err {:?}", err);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn start_decommission(&self, indices: Vec<usize>) -> Result<()> {
|
||||
if indices.is_empty() {
|
||||
return Err(Error::msg("errInvalidArgument"));
|
||||
}
|
||||
|
||||
if self.single_pool() {
|
||||
return Err(Error::msg("errInvalidArgument"));
|
||||
}
|
||||
|
||||
let decom_buckets = self.get_buckets_to_decommission().await?;
|
||||
|
||||
for bk in decom_buckets.iter() {
|
||||
let _ = self.heal_bucket(&bk.name, &HealOpts::default()).await;
|
||||
}
|
||||
|
||||
let meta_buckets = vec![
|
||||
path_join(&[PathBuf::from(RUSTFS_META_BUCKET), PathBuf::from(CONFIG_PREFIX)]),
|
||||
path_join(&[PathBuf::from(RUSTFS_META_BUCKET), PathBuf::from(BUCKET_META_PREFIX)]),
|
||||
];
|
||||
|
||||
for bk in meta_buckets.iter() {
|
||||
if let Err(err) = self
|
||||
.make_bucket(bk.to_string_lossy().to_string().as_str(), &MakeBucketOptions::default())
|
||||
.await
|
||||
{
|
||||
if !is_err_bucket_exists(&err) {
|
||||
return Err(err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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?;
|
||||
|
||||
pool_meta.decommission(idx.clone(), pi)?;
|
||||
|
||||
pool_meta.queue_buckets(idx.clone(), decom_buckets.clone());
|
||||
}
|
||||
|
||||
pool_meta.save(self.pools.clone()).await?;
|
||||
|
||||
// FIXME: globalNotificationSys.ReloadPoolMeta(ctx)
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_buckets_to_decommission(&self) -> Result<Vec<DecomBucketInfo>> {
|
||||
let buckets = self.list_bucket(&BucketOptions::default()).await?;
|
||||
|
||||
let mut ret: Vec<DecomBucketInfo> = buckets
|
||||
.iter()
|
||||
.map(|v| DecomBucketInfo {
|
||||
name: v.name.clone(),
|
||||
..Default::default()
|
||||
})
|
||||
.collect();
|
||||
|
||||
ret.push(DecomBucketInfo {
|
||||
name: RUSTFS_META_BUCKET.to_owned(),
|
||||
prefix: CONFIG_PREFIX.to_owned(),
|
||||
});
|
||||
ret.push(DecomBucketInfo {
|
||||
name: RUSTFS_META_BUCKET.to_owned(),
|
||||
prefix: BUCKET_META_PREFIX.to_owned(),
|
||||
});
|
||||
|
||||
Ok(ret)
|
||||
}
|
||||
}
|
||||
|
||||
fn get_total_usable_capacity(disks: &Vec<StorageDisk>, info: &StorageInfo) -> usize {
|
||||
|
||||
@@ -308,7 +308,7 @@ impl Operation for StatusPool {
|
||||
let query = {
|
||||
if let Some(query) = req.uri.query() {
|
||||
let input: StatusPoolQuery =
|
||||
from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidRequest, "get body failed"))?;
|
||||
from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed"))?;
|
||||
input
|
||||
} else {
|
||||
StatusPoolQuery::default()
|
||||
@@ -353,10 +353,77 @@ pub struct StartDecommission {}
|
||||
#[async_trait::async_trait]
|
||||
impl Operation for StartDecommission {
|
||||
// POST <endpoint>/<admin-API>/pools/decommission?pool=http://server{1...4}/disk{1...4}
|
||||
async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
warn!("handle StartDecommission");
|
||||
|
||||
return Err(s3_error!(NotImplemented));
|
||||
let Some(endpoints) = GLOBAL_Endpoints.get() else {
|
||||
return Err(s3_error!(NotImplemented));
|
||||
};
|
||||
|
||||
if endpoints.legacy() {
|
||||
return Err(s3_error!(NotImplemented));
|
||||
}
|
||||
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
|
||||
if store.is_decommission_running().await {
|
||||
return Err(S3Error::with_message(
|
||||
S3ErrorCode::InvalidRequest,
|
||||
"DecommissionAlreadyRunning".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
// TODO: check IsRebalanceStarted
|
||||
|
||||
let query = {
|
||||
if let Some(query) = req.uri.query() {
|
||||
let input: StatusPoolQuery =
|
||||
from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed"))?;
|
||||
input
|
||||
} else {
|
||||
StatusPoolQuery::default()
|
||||
}
|
||||
};
|
||||
let is_byid = query.by_id.as_str() == "true";
|
||||
|
||||
let pools: Vec<&str> = query.pool.split(",").collect();
|
||||
let mut pools_indices = Vec::with_capacity(pools.len());
|
||||
|
||||
for pool in pools.iter() {
|
||||
let idx = {
|
||||
if is_byid {
|
||||
pool.parse::<usize>()
|
||||
.map_err(|_e| s3_error!(InvalidArgument, "pool parse failed"))?
|
||||
} else {
|
||||
let Some(idx) = endpoints.get_pool_idx(pool) else {
|
||||
return Err(s3_error!(InvalidArgument, "pool parse failed"));
|
||||
};
|
||||
idx
|
||||
}
|
||||
};
|
||||
|
||||
let mut has_found = None;
|
||||
for (i, pool) in store.pools.iter().enumerate() {
|
||||
if i == idx {
|
||||
has_found = Some(pool.clone());
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
let Some(_p) = has_found else {
|
||||
return Err(s3_error!(InvalidArgument));
|
||||
};
|
||||
|
||||
pools_indices.push(idx);
|
||||
}
|
||||
|
||||
if !pools_indices.is_empty() {
|
||||
store.decommission(pools_indices).await.map_err(to_s3_error)?;
|
||||
}
|
||||
|
||||
Ok(S3Response::new((StatusCode::OK, Body::default())))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -379,7 +446,7 @@ impl Operation for CancelDecommission {
|
||||
let query = {
|
||||
if let Some(query) = req.uri.query() {
|
||||
let input: StatusPoolQuery =
|
||||
from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidRequest, "get body failed"))?;
|
||||
from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed"))?;
|
||||
input
|
||||
} else {
|
||||
StatusPoolQuery::default()
|
||||
|
||||
Reference in New Issue
Block a user