feat(admin): restore config admin compatibility (#3133)

* feat(admin): restore config admin compatibility

Co-authored-by: weisd <im@weisd.in>

* fix(admin): align config admin clean rebuild

Co-authored-by: weisd <im@weisd.in>

* fix(admin): align config history and peer signals

* fix(admin): harden config admin mutations

* fix(admin): tighten config review follow-ups

* perf(admin): reuse env snapshot in config render

* fix(ecstore): clean up config admin and listing error handling

Remove redundant is_all_volume_not_found check in list_merged, add
storage class encode/decode roundtrip tests, fresh boot integration
test, and config admin clean rebuild improvements.

Co-authored-by: hehutu <heihutu@gmail.com>

* fix(admin): sync global server config on mutation and reload

Change GLOBAL_SERVER_CONFIG from OnceLock to RwLock so config mutations
(set/del/restore/reload) are visible to readers without restart. Call
set_global_server_config after every store save and on snapshot reload.
Register storage_class as a dynamic config subsystem.

Co-authored-by: hehutu <heihutu@gmail.com>

* style: apply rustfmt to config and admin tests

Co-authored-by: hehutu <heihutu@gmail.com>

* fix(test): update signal_service test for storage_class dynamic subsystem

storage_class is now a valid dynamic config subsystem, so the
"requires object layer" test should expect "storage layer not initialized"
instead of "unsupported dynamic config subsystem".

Co-authored-by: hehutu <heihutu@gmail.com>

* fix(config): publish storage_class runtime config on dynamic reload

Change GLOBAL_STORAGE_CLASS from OnceLock to RwLock so runtime updates
are possible. apply_storage_class_runtime_config now actually publishes
the parsed config via set_global_storage_class instead of dropping it.

Addresses review feedback: storage_class was marked as dynamically
applied but the parsed result was discarded, so mc admin config set
returned config_applied=true while the runtime kept using stale parity
settings until restart.

Co-authored-by: hehutu <heihutu@gmail.com>

* fix(admin): harden config init, history ordering, and env redaction

- Change GLOBAL_SERVER_CONFIG from RwLock<Config> to RwLock<Option<Config>>
  initialized with None, preserving "not initialized" detection via None
- Move save_server_config_history before save_server_config_to_store in
  SetConfigKVHandler, DelConfigKVHandler, and SetConfigHandler so a
  restore point exists before mutations are persisted
- Redact sensitive env override values with *redacted* instead of
  silently omitting the line, improving admin visibility
- Add code comment explaining VolumeNotFound removal rationale in
  list_merged for listing paths

Co-authored-by: hehutu <heihutu@gmail.com>

* fix(config): keep in-memory config in sync after set/restore/reload

GLOBAL_SERVER_CONFIG was a OnceLock set once at startup and never
updated. After mc admin config set writes to the store, any fallback
to get_global_server_config() returned stale init-time data. Similarly,
reload_runtime_config_snapshot read from the store but discarded the
result.

- Replace OnceLock with RwLock for GLOBAL_SERVER_CONFIG and
  GLOBAL_STORAGE_CLASS so they can be updated at runtime
- Add set_global_server_config / set_global_storage_class setters
- Call set_global_server_config after every config save (set-kv,
  del-kv, set-config, restore-history)
- Re-apply dynamic subsystems (storage_class, audit_webhook,
  audit_mqtt) and signal peers in reload_runtime_config_snapshot
  and full-config operations
- Fix render_selected_config scope boundary check: track per-scope
  line count instead of checking global lines.is_empty()
- Include STORAGE_CLASS_SUB_SYS in is_dynamic_config_subsystem so
  apply_storage_class_runtime_config is reachable

Co-authored-by: hehutu <heihutu@gmail.com>

* fix(storageclass): use CLASS_RRS key in lookup_config for RRS parity

lookup_config used kvs.get(RRS) where RRS="REDUCED_REDUNDANCY", but the
admin config path writes the key as CLASS_RRS="rrs". This caused RRS
values to never be read back, always falling back to default parity.

- Changed kvs.get(RRS) to kvs.get(CLASS_RRS) in lookup_config
- Added regression tests verifying RRS read/write consistency

Co-authored-by: hehutu <heihutu@gmail.com>

* fix(config): add peer-side logging and don't swallow apply errors

- Add tracing::warn! in reload_dynamic_config_runtime_state and
  reload_runtime_config_snapshot when config read or subsystem apply
  fails, so on-host diagnostics show which signal failed and why
- Change `let _ = apply_dynamic_config_for_subsystem(...)` to
  `if let Err(err) = ... { warn!(...) }` in reload_runtime_config_snapshot
  so per-subsystem failures are logged instead of silently swallowed
- Remove weak test global_server_config_returns_none_before_init that
  had no meaningful assertion due to shared global state

Co-authored-by: hehutu <heihutu@gmail.com>

* style: apply rustfmt to config and storageclass tests

Co-authored-by: hehutu <heihutu@gmail.com>

---------

Co-authored-by: weisd <im@weisd.in>
Co-authored-by: hehutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-05-31 19:50:13 +08:00
committed by GitHub
parent 92104cb354
commit 8577bd825e
19 changed files with 3027 additions and 39 deletions
@@ -3043,6 +3043,51 @@ mod tests {
);
}
#[tokio::test]
#[serial]
async fn ecstore_new_succeeds_on_fresh_local_volumes() {
let test_base_dir = format!("/tmp/rustfs_ecstore_empty_boot_{}", Uuid::new_v4());
let temp_dir = PathBuf::from(&test_base_dir);
if temp_dir.exists() {
fs::remove_dir_all(&temp_dir).await.ok();
}
fs::create_dir_all(&temp_dir).await.unwrap();
let disk_paths = vec![
temp_dir.join("disk1"),
temp_dir.join("disk2"),
temp_dir.join("disk3"),
temp_dir.join("disk4"),
];
for disk_path in &disk_paths {
fs::create_dir_all(disk_path).await.unwrap();
}
let mut endpoints = Vec::new();
for (i, disk_path) in disk_paths.iter().enumerate() {
let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap();
endpoint.set_pool_index(0);
endpoint.set_set_index(0);
endpoint.set_disk_index(i);
endpoints.push(endpoint);
}
let endpoint_pools = EndpointServerPools(vec![PoolEndpoints {
legacy: false,
set_count: 1,
drives_per_set: 4,
endpoints: Endpoints::from(endpoints),
cmd_line: "fresh-boot-test".to_string(),
platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
}]);
crate::store::init_local_disks(endpoint_pools.clone()).await.unwrap();
let ecstore = ECStore::new("127.0.0.1:0".parse().unwrap(), endpoint_pools, CancellationToken::new()).await;
assert!(ecstore.is_ok(), "fresh local ECStore boot should succeed, got {ecstore:?}");
}
#[tokio::test]
#[serial]
async fn stale_multipart_cleanup_uses_default_expiry_without_lifecycle() {
+86 -6
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::config::{Config, GLOBAL_STORAGE_CLASS, KVS, audit, notify, oidc, storageclass};
use crate::config::{Config, KVS, audit, notify, oidc, set_global_storage_class, storageclass};
use crate::disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET};
use crate::error::{Error, Result};
use crate::global::is_first_cluster_node_local;
@@ -1164,11 +1164,8 @@ async fn apply_dynamic_config_for_sub_sys<S: StorageAPI>(cfg: &mut Config, api:
for (i, count) in set_drive_counts.iter().enumerate() {
match storageclass::lookup_config(&kvs, *count) {
Ok(res) => {
if i == 0
&& GLOBAL_STORAGE_CLASS.get().is_none()
&& let Err(r) = GLOBAL_STORAGE_CLASS.set(res)
{
error!("GLOBAL_STORAGE_CLASS.set failed {:?}", r);
if i == 0 {
set_global_storage_class(res);
}
}
Err(err) => {
@@ -2638,4 +2635,87 @@ mod tests {
assert_eq!(object_info.bucket, crate::disk::RUSTFS_META_BUCKET);
assert_eq!(object_info.name, "config/test.json");
}
#[test]
fn test_encode_decode_roundtrip_preserves_storage_class() {
use crate::config::STORAGE_CLASS_SUB_SYS;
// Create a config with custom storage class values
let mut cfg = Config::new();
let kvs = storage_class_kvs_mut(&mut cfg);
kvs.insert("standard".to_string(), "EC:4".to_string());
kvs.insert("rrs".to_string(), "EC:2".to_string());
kvs.insert("optimize".to_string(), "availability".to_string());
// Encode to external format (what save_server_config produces)
let encoded = encode_server_config_blob(&cfg, None).expect("encode should succeed");
// Verify the encoded data uses "storageclass" (no underscore)
let v: serde_json::Value = serde_json::from_slice(&encoded).expect("should be valid json");
assert!(v.get("storageclass").is_some(), "encoded should have 'storageclass' key");
assert!(v.get("storage_class").is_none(), "encoded should NOT have 'storage_class' key");
// Decode back (what load_server_config_from_store produces)
let decoded = decode_server_config_blob(&encoded).expect("decode should succeed");
// Verify the decoded config has "storage_class" (with underscore) subsystem
let kvs = decoded
.get_value(STORAGE_CLASS_SUB_SYS, crate::config::DEFAULT_DELIMITER)
.expect("decoded config should have storage_class subsystem");
assert_eq!(kvs.get("standard"), "EC:4", "standard should be EC:4");
assert_eq!(kvs.get("rrs"), "EC:2", "rrs should be EC:2");
assert_eq!(kvs.get("optimize"), "availability", "optimize should be availability");
}
#[test]
fn test_set_then_load_preserves_storage_class_values() {
use crate::config::STORAGE_CLASS_SUB_SYS;
// Step 1: Start with a fresh config (simulates initial server state)
let mut cfg = Config::new();
// Step 2: Apply set directives (simulates "mc admin config set storage_class standard=EC:4 rrs=EC:2")
let kvs = cfg
.0
.entry(STORAGE_CLASS_SUB_SYS.to_string())
.or_default()
.entry(DEFAULT_DELIMITER.to_string())
.or_default();
kvs.insert("standard".to_string(), "EC:4".to_string());
kvs.insert("rrs".to_string(), "EC:2".to_string());
cfg.set_defaults();
// Step 3: Save (encode to external format)
let encoded = encode_server_config_blob(&cfg, None).expect("encode should succeed");
// Step 4: Load (decode from external format)
let loaded = decode_server_config_blob(&encoded).expect("decode should succeed");
// Step 5: Verify the values are preserved
let loaded_kvs = loaded
.get_value(STORAGE_CLASS_SUB_SYS, DEFAULT_DELIMITER)
.expect("should have storage_class");
assert_eq!(loaded_kvs.get("standard"), "EC:4");
assert_eq!(loaded_kvs.get("rrs"), "EC:2");
// Step 6: Verify the subsystem is accessible by name (what render_selected_config does)
let targets = loaded
.0
.get(STORAGE_CLASS_SUB_SYS)
.expect("storage_class subsystem should exist");
let target_kvs = targets.get(DEFAULT_DELIMITER).expect("default target should exist");
assert_eq!(target_kvs.get("standard"), "EC:4");
assert_eq!(target_kvs.get("rrs"), "EC:2");
}
#[test]
fn test_config_unmarshal_fails_on_external_format() {
// This verifies that the external "storageclass" format cannot be
// mistakenly parsed by Config::unmarshal (which expects internal format).
// If unmarshal succeeds, the config would have "storageclass" instead of
// "storage_class", causing render_selected_config to fail.
let external_json = r#"{"version":"33","storageclass":{"standard":"EC:4","rrs":"EC:2"}}"#;
let result = Config::unmarshal(external_json.as_bytes());
assert!(result.is_err(), "Config::unmarshal should fail on external format, but got: {:?}", result);
}
}
+44 -5
View File
@@ -37,11 +37,12 @@ use rustfs_config::oidc::IDENTITY_OPENID_SUB_SYS;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::LazyLock;
use std::sync::{Arc, OnceLock};
use std::sync::{Arc, OnceLock, RwLock};
pub static GLOBAL_STORAGE_CLASS: LazyLock<OnceLock<storageclass::Config>> = LazyLock::new(OnceLock::new);
pub static GLOBAL_STORAGE_CLASS: LazyLock<RwLock<storageclass::Config>> =
LazyLock::new(|| RwLock::new(storageclass::Config::default()));
pub static DEFAULT_KVS: LazyLock<OnceLock<HashMap<String, KVS>>> = LazyLock::new(OnceLock::new);
pub static GLOBAL_SERVER_CONFIG: LazyLock<OnceLock<Config>> = LazyLock::new(OnceLock::new);
pub static GLOBAL_SERVER_CONFIG: LazyLock<RwLock<Option<Config>>> = LazyLock::new(|| RwLock::new(None));
pub static GLOBAL_CONFIG_SYS: LazyLock<ConfigSys> = LazyLock::new(ConfigSys::new);
pub static RUSTFS_CONFIG_PREFIX: &str = "config";
@@ -63,14 +64,30 @@ impl ConfigSys {
lookup_configs(&mut cfg, api).await;
let _ = GLOBAL_SERVER_CONFIG.set(cfg);
set_global_server_config(cfg);
Ok(())
}
}
pub fn get_global_server_config() -> Option<Config> {
GLOBAL_SERVER_CONFIG.get().cloned()
GLOBAL_SERVER_CONFIG.read().ok().and_then(|guard| (*guard).clone())
}
pub fn set_global_server_config(cfg: Config) {
if let Ok(mut guard) = GLOBAL_SERVER_CONFIG.write() {
*guard = Some(cfg);
}
}
pub fn get_global_storage_class() -> Option<storageclass::Config> {
GLOBAL_STORAGE_CLASS.read().ok().map(|guard| (*guard).clone())
}
pub fn set_global_storage_class(cfg: storageclass::Config) {
if let Ok(mut guard) = GLOBAL_STORAGE_CLASS.write() {
*guard = cfg;
}
}
pub async fn init_global_config_sys(api: Arc<ECStore>) -> Result<()> {
@@ -262,3 +279,25 @@ pub fn init() {
// Register all default configurations
register_default_kvs(kvs)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn global_server_config_set_and_get_roundtrip() {
init();
let mut cfg = Config::new();
let mut kvs = KVS::new();
kvs.insert("standard".to_string(), "EC:4".to_string());
cfg.0
.insert(STORAGE_CLASS_SUB_SYS.to_string(), HashMap::from([("_".to_string(), kvs)]));
set_global_server_config(cfg.clone());
let loaded = get_global_server_config().expect("global config should be set");
let sc_kvs = loaded
.get_value(STORAGE_CLASS_SUB_SYS, "_")
.expect("storage_class should exist");
assert_eq!(sc_kvs.get("standard"), "EC:4");
}
}
+36 -3
View File
@@ -101,13 +101,13 @@ pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
});
// StorageClass - holds storage class information
#[derive(Serialize, Deserialize, Debug, Default)]
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
pub struct StorageClass {
parity: usize,
}
// Config storage class configuration
#[derive(Serialize, Deserialize, Debug, Default)]
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
pub struct Config {
standard: StorageClass,
rrs: StorageClass,
@@ -205,7 +205,7 @@ pub fn lookup_config(kvs: &KVS, set_drive_count: usize) -> Result<Config> {
if let Ok(ssc_str) = env::var(RRS_ENV) {
ssc_str
} else {
kvs.get(RRS)
kvs.get(CLASS_RRS)
}
};
@@ -336,3 +336,36 @@ pub fn validate_parity_inner(ss_parity: usize, rrs_parity: usize, set_drive_coun
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn lookup_config_reads_rrs_from_class_rrs_key() {
// Regression: kvs.get(RRS) used RRS="REDUCED_REDUNDANCY" instead of
// CLASS_RRS="rrs", so admin-written RRS values were never read back.
let mut kvs = KVS::new();
kvs.insert(CLASS_STANDARD.to_string(), "EC:4".to_string());
kvs.insert(CLASS_RRS.to_string(), "EC:2".to_string());
let cfg = lookup_config(&kvs, 8).expect("lookup should succeed");
assert_eq!(cfg.standard.parity, 4, "standard parity should be 4");
assert_eq!(cfg.rrs.parity, 2, "rrs parity should be 2");
}
#[test]
fn lookup_config_ignores_redundancy_key_name() {
// Ensure the old key name "REDUCED_REDUNDANCY" is NOT read.
let mut kvs = KVS::new();
kvs.insert(CLASS_STANDARD.to_string(), "EC:4".to_string());
kvs.insert(RRS.to_string(), "EC:2".to_string());
let cfg = lookup_config(&kvs, 8).expect("lookup should succeed");
assert_eq!(cfg.standard.parity, 4);
assert_eq!(
cfg.rrs.parity, DEFAULT_RRS_PARITY,
"rrs should fall back to default when CLASS_RRS key is absent"
);
}
}
+59
View File
@@ -189,6 +189,65 @@ impl NotificationSys {
join_all(futures).await
}
pub async fn reload_dynamic_config(&self, sub_sys: &str) -> Vec<NotificationPeerErr> {
let mut futures = Vec::with_capacity(self.peer_clients.len());
for client in self.peer_clients.iter() {
let sub_sys = sub_sys.to_string();
futures.push(async move {
if let Some(client) = client {
match client
.signal_service(crate::rpc::SERVICE_SIGNAL_RELOAD_DYNAMIC, &sub_sys, false, SystemTime::UNIX_EPOCH)
.await
{
Ok(_) => NotificationPeerErr {
host: client.host.to_string(),
err: None,
},
Err(e) => NotificationPeerErr {
host: client.host.to_string(),
err: Some(e),
},
}
} else {
NotificationPeerErr {
host: "".to_string(),
err: Some(Error::other("peer is not reachable")),
}
}
});
}
join_all(futures).await
}
pub async fn refresh_config_snapshot(&self) -> Vec<NotificationPeerErr> {
let mut futures = Vec::with_capacity(self.peer_clients.len());
for client in self.peer_clients.iter() {
futures.push(async move {
if let Some(client) = client {
match client
.signal_service(crate::rpc::SERVICE_SIGNAL_REFRESH_CONFIG, "", false, SystemTime::UNIX_EPOCH)
.await
{
Ok(_) => NotificationPeerErr {
host: client.host.to_string(),
err: None,
},
Err(e) => NotificationPeerErr {
host: client.host.to_string(),
err: Some(e),
},
}
} else {
NotificationPeerErr {
host: "".to_string(),
err: Some(Error::other("peer is not reachable")),
}
}
});
}
join_all(futures).await
}
pub async fn storage_info<S: StorageAPI>(&self, api: &S) -> rustfs_madmin::StorageInfo {
let mut futures = Vec::with_capacity(self.peer_clients.len());
let endpoints = get_global_endpoints();
+4 -1
View File
@@ -29,7 +29,10 @@ pub use internode_data_transport::{
InternodeDataTransport, InternodeDataTransportCapabilities, ReadStreamRequest, TcpHttpInternodeDataTransport,
WalkDirStreamRequest, WriteStreamRequest, build_internode_data_transport, build_internode_data_transport_from_env,
};
pub use peer_rest_client::PeerRestClient;
pub use peer_rest_client::{
PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, SERVICE_SIGNAL_REFRESH_CONFIG,
SERVICE_SIGNAL_RELOAD_DYNAMIC,
};
pub use peer_s3_client::{LocalPeerS3Client, PeerS3Client, RemotePeerS3Client, S3PeerSys};
pub use remote_disk::RemoteDisk;
pub use remote_locker::RemoteClient;
@@ -57,6 +57,8 @@ use tracing::warn;
pub const PEER_RESTSIGNAL: &str = "signal";
pub const PEER_RESTSUB_SYS: &str = "sub-sys";
pub const PEER_RESTDRY_RUN: &str = "dry-run";
pub const SERVICE_SIGNAL_REFRESH_CONFIG: u64 = 1;
pub const SERVICE_SIGNAL_RELOAD_DYNAMIC: u64 = 2;
const PEER_REST_RECOVERY_MAX_ATTEMPTS: u32 = 60;
const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30);
+5 -6
View File
@@ -41,7 +41,7 @@ use crate::{
LifecycleOps, gen_transition_objname, get_transitioned_object_reader, put_restore_opts,
},
cache_value::metacache_set::{ListPathRawOptions, list_path_raw},
config::{GLOBAL_STORAGE_CLASS, storageclass},
config::{get_global_storage_class, storageclass},
disk::{
CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskOption, DiskStore, FileInfoVersions,
RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions,
@@ -282,8 +282,7 @@ fn build_tiered_decommission_file_info(
default_parity_count: usize,
storage_class: Option<&str>,
) -> (FileInfo, usize) {
let parity_drives = GLOBAL_STORAGE_CLASS
.get()
let parity_drives = get_global_storage_class()
.and_then(|sc| sc.get_parity_for_sc(storage_class.unwrap_or_default()))
.unwrap_or(default_parity_count);
let data_drives = disk_count - parity_drives;
@@ -783,7 +782,7 @@ impl ObjectIO for SetDisks {
}
}
let sc_parity_drives = {
if let Some(sc) = GLOBAL_STORAGE_CLASS.get() {
if let Some(sc) = get_global_storage_class() {
sc.get_parity_for_sc(user_defined.get(AMZ_STORAGE_CLASS).cloned().unwrap_or_default().as_str())
} else {
None
@@ -837,7 +836,7 @@ impl ObjectIO for SetDisks {
let erasure = erasure_coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let is_inline_buffer = {
if let Some(sc) = GLOBAL_STORAGE_CLASS.get() {
if let Some(sc) = get_global_storage_class() {
sc.should_inline(erasure.shard_file_size(data.size()), opts.versioned)
} else {
false
@@ -3108,7 +3107,7 @@ impl MultipartOperations for SetDisks {
}
let sc_parity_drives = {
if let Some(sc) = GLOBAL_STORAGE_CLASS.get() {
if let Some(sc) = get_global_storage_class() {
sc.get_parity_for_sc(user_defined.get(AMZ_STORAGE_CLASS).cloned().unwrap_or_default().as_str())
} else {
None
+1 -1
View File
@@ -417,7 +417,7 @@ impl SetDisks {
}
let is_inline_buffer = {
if let Some(sc) = GLOBAL_STORAGE_CLASS.get() {
if let Some(sc) = get_global_storage_class() {
sc.should_inline(erasure.shard_file_size(latest_meta.size), false)
} else {
false
-1
View File
@@ -31,7 +31,6 @@ use crate::bucket::utils::check_object_args;
use crate::bucket::utils::check_put_object_args;
use crate::bucket::utils::check_put_object_part_args;
use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname};
use crate::config::GLOBAL_STORAGE_CLASS;
use crate::config::storageclass;
use crate::disk::endpoint::{Endpoint, EndpointType};
use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions};
+2 -1
View File
@@ -13,6 +13,7 @@
// limitations under the License.
use super::*;
use crate::config::get_global_storage_class;
struct LatestObjectInfoCandidate {
info: Option<ObjectInfo>,
@@ -605,7 +606,7 @@ impl ECStore {
#[instrument(skip(self))]
pub(super) async fn handle_backend_info(&self) -> rustfs_madmin::BackendInfo {
let (standard_sc_parity, rr_sc_parity) = {
if let Some(sc) = GLOBAL_STORAGE_CLASS.get() {
if let Some(sc) = get_global_storage_class() {
let sc_parity = sc
.get_parity_for_sc(storageclass::CLASS_STANDARD)
.or(Some(self.pools[0].default_parity_count));
+6 -4
View File
@@ -841,11 +841,13 @@ impl ECStore {
}
}
// All sets returned not-found — the listing is simply empty.
// We intentionally do NOT distinguish VolumeNotFound here: during
// listing, a missing volume on a set is equivalent to "no entries"
// and must not surface as an error to the caller. The original
// VolumeNotFound check caused spurious errors when a pool had not
// yet created the bucket prefix on every set.
if is_all_not_found(&errs) {
if is_all_volume_not_found(&errs) {
return Err(StorageError::VolumeNotFound);
}
return Ok(Vec::new());
}
File diff suppressed because it is too large Load Diff
+7
View File
@@ -16,6 +16,7 @@ pub mod account_info;
pub mod audit;
mod audit_runtime_config;
pub mod bucket_meta;
pub mod config_admin;
pub mod event;
pub mod group;
pub mod heal;
@@ -60,6 +61,12 @@ mod tests {
fn test_handler_struct_creation() {
// Test that handler structs can be created
let _account_handler = account_info::AccountInfoHandler {};
let _get_config_kv_handler = config_admin::GetConfigKVHandler {};
let _set_config_kv_handler = config_admin::SetConfigKVHandler {};
let _del_config_kv_handler = config_admin::DelConfigKVHandler {};
let _help_config_kv_handler = config_admin::HelpConfigKVHandler {};
let _get_config_handler = config_admin::GetConfigHandler {};
let _set_config_handler = config_admin::SetConfigHandler {};
let _list_audit_targets = audit::ListAuditTargets {};
let _get_module_switches = module_switch::GetModuleSwitchesHandler {};
let _get_plugin_catalog = plugins_catalog::GetPluginCatalogHandler {};
+3 -2
View File
@@ -27,8 +27,8 @@ mod console_test;
mod route_registration_test;
use handlers::{
audit, bucket_meta, heal, health, kms, module_switch, oidc, plugins_catalog, plugins_instances, pools, profile_admin, quota,
rebalance, replication, site_replication, sts, system, tier, tls_debug, user,
audit, bucket_meta, config_admin, heal, health, kms, module_switch, oidc, plugins_catalog, plugins_instances, pools,
profile_admin, quota, rebalance, replication, site_replication, sts, system, tier, tls_debug, user,
};
use router::{AdminOperation, S3Router};
use s3s::route::S3Route;
@@ -57,6 +57,7 @@ pub fn make_admin_route(console_enabled: bool) -> std::io::Result<impl S3Route>
quota::register_quota_route(&mut r)?;
bucket_meta::register_bucket_meta_route(&mut r)?;
config_admin::register_config_route(&mut r)?;
audit::register_audit_target_route(&mut r)?;
module_switch::register_module_switch_route(&mut r)?;
plugins_catalog::register_plugin_catalog_route(&mut r)?;
+421
View File
@@ -0,0 +1,421 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use rustfs_audit::reload_audit_config;
use rustfs_config::audit::{AUDIT_MQTT_SUB_SYS, AUDIT_REDIS_DEFAULT_CHANNEL, AUDIT_WEBHOOK_SUB_SYS};
use rustfs_config::notify::{NOTIFY_MQTT_SUB_SYS, NOTIFY_REDIS_DEFAULT_CHANNEL, NOTIFY_WEBHOOK_SUB_SYS};
use rustfs_config::oidc::IDENTITY_OPENID_SUB_SYS;
use rustfs_config::{AUDIT_DEFAULT_DIR, EVENT_DEFAULT_DIR};
use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY, EnableState};
use rustfs_ecstore::StorageAPI;
use rustfs_ecstore::config::com::{STORAGE_CLASS_SUB_SYS, read_config_without_migrate};
use rustfs_ecstore::config::storageclass;
use rustfs_ecstore::config::{Config as ServerConfig, KVS, set_global_server_config, set_global_storage_class};
use rustfs_ecstore::new_object_layer_fn;
use rustfs_ecstore::notification_sys::get_global_notification_sys;
use rustfs_iam::oidc::load_oidc_provider_configs_from_server_config;
use rustfs_targets::config::{
validate_amqp_config, validate_kafka_config, validate_mqtt_config, validate_mysql_config, validate_nats_config,
validate_postgres_config, validate_pulsar_config, validate_redis_config, validate_webhook_config,
};
use s3s::{S3Error, S3ErrorCode, S3Result};
use tracing::warn;
use url::Url;
pub fn is_dynamic_config_subsystem(sub_system: &str) -> bool {
matches!(sub_system, STORAGE_CLASS_SUB_SYS | AUDIT_WEBHOOK_SUB_SYS | AUDIT_MQTT_SUB_SYS)
}
fn internal_error(message: impl Into<String>) -> S3Error {
S3Error::with_message(S3ErrorCode::InternalError, message.into())
}
fn invalid_request(message: impl Into<String>) -> S3Error {
S3Error::with_message(S3ErrorCode::InvalidRequest, message.into())
}
async fn apply_storage_class_runtime_config(config: &ServerConfig) -> S3Result<()> {
let Some(store) = new_object_layer_fn() else {
return Err(internal_error("storage layer not initialized"));
};
let kvs = config.get_value(STORAGE_CLASS_SUB_SYS, DEFAULT_DELIMITER).unwrap_or_default();
let set_drive_count = store.set_drive_counts().into_iter().next().unwrap_or(1);
let parsed = storageclass::lookup_config(&kvs, set_drive_count)
.map_err(|err| internal_error(format!("failed to apply storage class config: {err}")))?;
set_global_storage_class(parsed);
Ok(())
}
fn validate_storage_class_kvs(kvs: &KVS, set_drive_counts: &[usize]) -> S3Result<()> {
for count in set_drive_counts {
storageclass::lookup_config(kvs, *count)
.map_err(|err| invalid_request(format!("invalid storage class config: {err}")))?;
}
Ok(())
}
async fn validate_storage_class_config(config: &ServerConfig) -> S3Result<()> {
let Some(store) = new_object_layer_fn() else {
return Err(internal_error("storage layer not initialized"));
};
let kvs = config.get_value(STORAGE_CLASS_SUB_SYS, DEFAULT_DELIMITER).unwrap_or_default();
let set_drive_counts = store.set_drive_counts();
if set_drive_counts.is_empty() {
return validate_storage_class_kvs(&kvs, &[1]);
}
validate_storage_class_kvs(&kvs, &set_drive_counts)
}
fn target_enabled(kvs: &KVS) -> bool {
kvs.lookup(ENABLE_KEY)
.and_then(|value| value.parse::<EnableState>().ok())
.is_some_and(|state| state.is_enabled())
}
fn target_kvs(config: &ServerConfig, sub_system: &str, target: &str) -> KVS {
let mut kvs = config.get_value(sub_system, DEFAULT_DELIMITER).unwrap_or_default();
if target != DEFAULT_DELIMITER
&& let Some(target_kvs) = config.0.get(sub_system).and_then(|targets| targets.get(target))
{
kvs.extend(target_kvs.clone());
}
kvs
}
fn validate_notify_subsystem_config(config: &ServerConfig, sub_system: &str) -> S3Result<()> {
let Some(targets) = config.0.get(sub_system) else {
return Ok(());
};
for target in targets.keys() {
let kvs = target_kvs(config, sub_system, target);
if !target_enabled(&kvs) {
continue;
}
let result = match sub_system {
"notify_webhook" => validate_webhook_config(&kvs, EVENT_DEFAULT_DIR),
"notify_amqp" => validate_amqp_config(&kvs, EVENT_DEFAULT_DIR),
"notify_kafka" => validate_kafka_config(&kvs, EVENT_DEFAULT_DIR),
"notify_mqtt" => validate_mqtt_config(&kvs),
"notify_mysql" => validate_mysql_config(&kvs, EVENT_DEFAULT_DIR),
"notify_nats" => validate_nats_config(&kvs, EVENT_DEFAULT_DIR),
"notify_postgres" => validate_postgres_config(&kvs, EVENT_DEFAULT_DIR),
"notify_pulsar" => validate_pulsar_config(&kvs, EVENT_DEFAULT_DIR),
"notify_redis" => validate_redis_config(&kvs, EVENT_DEFAULT_DIR, NOTIFY_REDIS_DEFAULT_CHANNEL),
_ => return Ok(()),
};
result.map_err(|err| invalid_request(format!("invalid {sub_system} config for target '{target}': {err}")))?;
}
Ok(())
}
fn validate_audit_subsystem_config(config: &ServerConfig, sub_system: &str) -> S3Result<()> {
let Some(targets) = config.0.get(sub_system) else {
return Ok(());
};
for target in targets.keys() {
let kvs = target_kvs(config, sub_system, target);
if !target_enabled(&kvs) {
continue;
}
let result = match sub_system {
"audit_webhook" => validate_webhook_config(&kvs, AUDIT_DEFAULT_DIR),
"audit_amqp" => validate_amqp_config(&kvs, AUDIT_DEFAULT_DIR),
"audit_kafka" => validate_kafka_config(&kvs, AUDIT_DEFAULT_DIR),
"audit_mqtt" => validate_mqtt_config(&kvs),
"audit_mysql" => validate_mysql_config(&kvs, AUDIT_DEFAULT_DIR),
"audit_nats" => validate_nats_config(&kvs, AUDIT_DEFAULT_DIR),
"audit_postgres" => validate_postgres_config(&kvs, AUDIT_DEFAULT_DIR),
"audit_pulsar" => validate_pulsar_config(&kvs, AUDIT_DEFAULT_DIR),
"audit_redis" => validate_redis_config(&kvs, AUDIT_DEFAULT_DIR, AUDIT_REDIS_DEFAULT_CHANNEL),
_ => return Ok(()),
};
result.map_err(|err| invalid_request(format!("invalid {sub_system} config for target '{target}': {err}")))?;
}
Ok(())
}
fn is_valid_provider_id(id: &str) -> bool {
!id.is_empty() && id.chars().all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
}
fn is_valid_scheme(scheme: &str) -> bool {
scheme == "http" || scheme == "https"
}
fn validate_absolute_http_url(value: &str, field_name: &str) -> S3Result<()> {
let parsed = Url::parse(value).map_err(|_| invalid_request(format!("{field_name} must be an absolute http/https URL")))?;
if !is_valid_scheme(parsed.scheme()) || parsed.host_str().is_none() {
return Err(invalid_request(format!("{field_name} must be an absolute http/https URL")));
}
Ok(())
}
fn validate_identity_openid_config(config: &ServerConfig) -> S3Result<()> {
if let Some(targets) = config.0.get(IDENTITY_OPENID_SUB_SYS) {
for target in targets.keys() {
if target == DEFAULT_DELIMITER {
continue;
}
if !is_valid_provider_id(target) {
return Err(invalid_request(format!("invalid provider_id '{target}'")));
}
}
}
for provider in load_oidc_provider_configs_from_server_config(config) {
if !is_valid_provider_id(&provider.id) {
return Err(invalid_request(format!("invalid provider_id '{}'", provider.id)));
}
if provider.config_url.trim().is_empty() {
return Err(invalid_request(format!("identity_openid provider '{}' requires config_url", provider.id)));
}
validate_absolute_http_url(&provider.config_url, "config_url")?;
if provider.client_id.trim().is_empty() {
return Err(invalid_request(format!("identity_openid provider '{}' requires client_id", provider.id)));
}
if !provider.redirect_uri_dynamic {
let Some(redirect_uri) = provider.redirect_uri.as_deref() else {
return Err(invalid_request(format!(
"identity_openid provider '{}' requires redirect_uri when redirect_uri_dynamic is off",
provider.id
)));
};
validate_absolute_http_url(redirect_uri, "redirect_uri")?;
} else if let Some(redirect_uri) = provider.redirect_uri.as_deref() {
validate_absolute_http_url(redirect_uri, "redirect_uri")?;
}
if !provider.scopes.iter().any(|scope| scope == "openid") {
return Err(invalid_request(format!(
"identity_openid provider '{}' scopes must include openid",
provider.id
)));
}
}
Ok(())
}
pub async fn validate_server_config(config: &ServerConfig, sub_system: Option<&str>) -> S3Result<()> {
match sub_system {
Some(STORAGE_CLASS_SUB_SYS) => validate_storage_class_config(config).await,
Some(NOTIFY_WEBHOOK_SUB_SYS) => validate_notify_subsystem_config(config, NOTIFY_WEBHOOK_SUB_SYS),
Some(NOTIFY_MQTT_SUB_SYS) => validate_notify_subsystem_config(config, NOTIFY_MQTT_SUB_SYS),
Some(AUDIT_WEBHOOK_SUB_SYS) => validate_audit_subsystem_config(config, AUDIT_WEBHOOK_SUB_SYS),
Some(AUDIT_MQTT_SUB_SYS) => validate_audit_subsystem_config(config, AUDIT_MQTT_SUB_SYS),
Some(IDENTITY_OPENID_SUB_SYS) => validate_identity_openid_config(config),
Some(_) => Ok(()),
None => {
validate_storage_class_config(config).await?;
validate_notify_subsystem_config(config, NOTIFY_WEBHOOK_SUB_SYS)?;
validate_notify_subsystem_config(config, NOTIFY_MQTT_SUB_SYS)?;
validate_audit_subsystem_config(config, AUDIT_WEBHOOK_SUB_SYS)?;
validate_audit_subsystem_config(config, AUDIT_MQTT_SUB_SYS)?;
validate_identity_openid_config(config)?;
Ok(())
}
}
}
pub async fn apply_dynamic_config_for_subsystem(config: &ServerConfig, sub_system: &str) -> S3Result<bool> {
if !is_dynamic_config_subsystem(sub_system) {
return Ok(false);
}
match sub_system {
STORAGE_CLASS_SUB_SYS => apply_storage_class_runtime_config(config).await?,
AUDIT_WEBHOOK_SUB_SYS | AUDIT_MQTT_SUB_SYS => reload_audit_config(config.clone())
.await
.map_err(|err| internal_error(format!("failed to reload audit config: {err}")))?,
_ => return Ok(false),
}
Ok(true)
}
pub async fn reload_dynamic_config_runtime_state(sub_system: &str) -> S3Result<()> {
if !is_dynamic_config_subsystem(sub_system) {
return Err(internal_error(format!("unsupported dynamic config subsystem: {sub_system}")));
}
let Some(store) = new_object_layer_fn() else {
return Err(internal_error("storage layer not initialized"));
};
let config = read_config_without_migrate(store).await.map_err(|err| {
warn!("peer reload_dynamic_config: failed to load server config for {sub_system}: {err}");
internal_error(format!("failed to load server config: {err}"))
})?;
apply_dynamic_config_for_subsystem(&config, sub_system).await.map_err(|err| {
warn!("peer reload_dynamic_config: failed to apply {sub_system}: {err}");
err
})?;
Ok(())
}
pub async fn reload_runtime_config_snapshot() -> S3Result<()> {
let Some(store) = new_object_layer_fn() else {
return Err(internal_error("storage layer not initialized"));
};
let config = read_config_without_migrate(store).await.map_err(|err| {
warn!("peer reload_runtime_config_snapshot: failed to load server config: {err}");
internal_error(format!("failed to load server config: {err}"))
})?;
// Re-apply dynamic subsystems before publishing the snapshot, so that
// runtime state (e.g. GLOBAL_STORAGE_CLASS) is refreshed on this peer.
for sub_system in [STORAGE_CLASS_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS, AUDIT_MQTT_SUB_SYS] {
if let Err(err) = apply_dynamic_config_for_subsystem(&config, sub_system).await {
warn!("peer reload_runtime_config_snapshot: failed to apply {sub_system}: {err}");
}
}
set_global_server_config(config);
Ok(())
}
pub async fn signal_dynamic_config_reload(sub_system: &str) {
if !is_dynamic_config_subsystem(sub_system) {
return;
}
let Some(notification_sys) = get_global_notification_sys() else {
return;
};
for failure in notification_sys.reload_dynamic_config(sub_system).await {
if let Some(err) = failure.err {
tracing::warn!("peer {} dynamic config reload for {} failed: {}", failure.host, sub_system, err);
}
}
}
pub async fn signal_config_snapshot_reload() {
let Some(notification_sys) = get_global_notification_sys() else {
return;
};
for failure in notification_sys.refresh_config_snapshot().await {
if let Some(err) = failure.err {
tracing::warn!("peer config snapshot refresh failed for {}: {}", failure.host, err);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use rustfs_config::notify::NOTIFY_WEBHOOK_SUB_SYS;
use rustfs_config::oidc::{OIDC_CLIENT_ID, OIDC_CONFIG_URL, OIDC_SCOPES};
use rustfs_config::{MQTT_BROKER, MQTT_QUEUE_DIR, MQTT_TOPIC, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR};
#[test]
fn dynamic_config_subsystems_match_runtime_apply_support() {
assert!(is_dynamic_config_subsystem(AUDIT_WEBHOOK_SUB_SYS));
assert!(is_dynamic_config_subsystem(AUDIT_MQTT_SUB_SYS));
assert!(is_dynamic_config_subsystem(STORAGE_CLASS_SUB_SYS));
assert!(!is_dynamic_config_subsystem("identity_openid"));
assert!(!is_dynamic_config_subsystem("notify_webhook"));
}
#[test]
fn validate_storage_class_kvs_rejects_invalid_parity() {
let mut kvs = KVS::new();
kvs.insert("standard".to_string(), "EC:5".to_string());
let err = validate_storage_class_kvs(&kvs, &[4]).expect_err("invalid parity should fail");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
}
#[test]
fn validate_notify_subsystem_config_rejects_invalid_webhook_endpoint() {
rustfs_ecstore::config::init();
let mut config = ServerConfig::new();
let targets = config.0.get_mut(NOTIFY_WEBHOOK_SUB_SYS).expect("notify webhook defaults");
let kvs = targets.get_mut(DEFAULT_DELIMITER).expect("default target");
kvs.insert(ENABLE_KEY.to_string(), EnableState::On.to_string());
kvs.insert(WEBHOOK_ENDPOINT.to_string(), "not-a-url".to_string());
kvs.insert(WEBHOOK_QUEUE_DIR.to_string(), "/tmp/rustfs-notify".to_string());
let err = validate_notify_subsystem_config(&config, NOTIFY_WEBHOOK_SUB_SYS).expect_err("invalid endpoint should fail");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
}
#[test]
fn validate_audit_subsystem_config_rejects_relative_queue_dir() {
rustfs_ecstore::config::init();
let mut config = ServerConfig::new();
let targets = config.0.get_mut(AUDIT_MQTT_SUB_SYS).expect("audit mqtt defaults");
let kvs = targets.get_mut(DEFAULT_DELIMITER).expect("default target");
kvs.insert(ENABLE_KEY.to_string(), EnableState::On.to_string());
kvs.insert(MQTT_BROKER.to_string(), "mqtt://localhost:1883".to_string());
kvs.insert(MQTT_TOPIC.to_string(), "audit-events".to_string());
kvs.insert(MQTT_QUEUE_DIR.to_string(), "relative/dir".to_string());
let err = validate_audit_subsystem_config(&config, AUDIT_MQTT_SUB_SYS).expect_err("relative queue dir should fail");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
}
#[test]
fn validate_identity_openid_config_rejects_missing_openid_scope() {
rustfs_ecstore::config::init();
let mut config = ServerConfig::new();
let targets = config.0.get_mut(IDENTITY_OPENID_SUB_SYS).expect("openid defaults");
let kvs = targets.get_mut(DEFAULT_DELIMITER).expect("default target");
kvs.insert(
OIDC_CONFIG_URL.to_string(),
"https://issuer.example/.well-known/openid-configuration".to_string(),
);
kvs.insert(OIDC_CLIENT_ID.to_string(), "console".to_string());
kvs.insert(OIDC_SCOPES.to_string(), "profile,email".to_string());
let err = validate_identity_openid_config(&config).expect_err("openid scope should be required");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
}
#[test]
fn validate_identity_openid_config_rejects_invalid_named_provider_id() {
rustfs_ecstore::config::init();
let mut config = ServerConfig::new();
let targets = config.0.get_mut(IDENTITY_OPENID_SUB_SYS).expect("openid defaults");
let default_kvs = targets.get(DEFAULT_DELIMITER).cloned().expect("default target");
let mut named_kvs = default_kvs;
named_kvs.insert(
OIDC_CONFIG_URL.to_string(),
"https://issuer.example/.well-known/openid-configuration".to_string(),
);
named_kvs.insert(OIDC_CLIENT_ID.to_string(), "console".to_string());
targets.insert("bad$id".to_string(), named_kvs);
let err = validate_identity_openid_config(&config).expect_err("provider id should be validated");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
}
}
+1
View File
@@ -12,4 +12,5 @@
// See the License for the specific language governing permissions and
// limitations under the License.
pub mod config;
pub mod site_replication;
+13
View File
@@ -293,8 +293,10 @@ fn is_empty_body_admin_path(method: &Method, uri: &http::Uri) -> bool {
path,
"/minio/admin/v3/set-user-status"
| "/minio/admin/v3/set-group-status"
| "/minio/admin/v3/restore-config-history-kv"
| "/rustfs/admin/v3/set-user-status"
| "/rustfs/admin/v3/set-group-status"
| "/rustfs/admin/v3/restore-config-history-kv"
),
Method::POST => {
matches!(
@@ -1571,6 +1573,17 @@ mod tests {
assert!(!should_force_zero_content_length_for_empty_body_route(&request));
}
#[test]
fn admin_restore_config_history_without_content_length_is_normalized() {
let request = Request::builder()
.method(Method::PUT)
.uri("/minio/admin/v3/restore-config-history-kv?restoreId=test")
.body(())
.expect("request");
assert!(should_force_zero_content_length_for_empty_body_route(&request));
}
#[test]
fn s3_put_object_is_not_normalized() {
let request = Request::builder()
+171 -9
View File
@@ -12,7 +12,10 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::admin::service::site_replication::reload_site_replication_runtime_state;
use crate::admin::service::{
config::{reload_dynamic_config_runtime_state, reload_runtime_config_snapshot},
site_replication::reload_site_replication_runtime_state,
};
use bytes::Bytes;
use futures::Stream;
use futures_util::future::join_all;
@@ -29,7 +32,10 @@ use rustfs_ecstore::{
global::GLOBAL_TierConfigMgr,
metrics_realtime::{CollectMetricsOpts, MetricType, collect_local_metrics},
new_object_layer_fn,
rpc::{LocalPeerS3Client, PeerS3Client},
rpc::{
LocalPeerS3Client, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerS3Client, SERVICE_SIGNAL_REFRESH_CONFIG,
SERVICE_SIGNAL_RELOAD_DYNAMIC,
},
store::{all_local_disk_path, find_local_disk_by_ref},
store_api::{BucketOptions, DeleteBucketOptions, MakeBucketOptions, StorageAPI},
};
@@ -45,7 +51,7 @@ use rustfs_protos::{
proto_gen::node_service::{node_service_server::NodeService as Node, *},
};
use serde::Deserialize;
use std::{io::Cursor, pin::Pin, sync::Arc};
use std::{collections::HashMap, io::Cursor, pin::Pin, sync::Arc};
use tokio::spawn;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
@@ -794,8 +800,49 @@ impl Node for NodeService {
}
async fn signal_service(&self, request: Request<SignalServiceRequest>) -> Result<Response<SignalServiceResponse>, Status> {
let _request = request.into_inner();
Err(unimplemented_rpc("signal_service"))
let request = request.into_inner();
let vars = match request.vars {
Some(vars) => vars.value,
None => HashMap::new(),
};
let raw_signal = vars.get(PEER_RESTSIGNAL).map(String::as_str);
let signal = raw_signal.and_then(|value| value.parse::<u64>().ok());
let sub_system = vars.get(PEER_RESTSUB_SYS).map(String::as_str).unwrap_or_default();
match signal {
Some(SERVICE_SIGNAL_REFRESH_CONFIG) => match reload_runtime_config_snapshot().await {
Ok(()) => Ok(Response::new(SignalServiceResponse {
success: true,
error_info: None,
})),
Err(err) => Ok(Response::new(SignalServiceResponse {
success: false,
error_info: Some(err.to_string()),
})),
},
Some(SERVICE_SIGNAL_RELOAD_DYNAMIC) => match reload_dynamic_config_runtime_state(sub_system).await {
Ok(()) => Ok(Response::new(SignalServiceResponse {
success: true,
error_info: None,
})),
Err(err) => Ok(Response::new(SignalServiceResponse {
success: false,
error_info: Some(err.to_string()),
})),
},
Some(other) => Ok(Response::new(SignalServiceResponse {
success: false,
error_info: Some(format!("unsupported service signal: {other}")),
})),
None if raw_signal.is_some() => Ok(Response::new(SignalServiceResponse {
success: false,
error_info: Some(format!("invalid service signal value: {}", raw_signal.unwrap_or_default())),
})),
None => Ok(Response::new(SignalServiceResponse {
success: false,
error_info: Some("missing service signal".to_string()),
})),
}
}
async fn background_heal_status(
@@ -2450,6 +2497,125 @@ mod tests {
assert!(reload_response.error_info.is_some());
}
#[tokio::test]
async fn test_signal_service_rejects_missing_signal() {
let service = create_test_node_service();
let request = Request::new(SignalServiceRequest {
vars: Some(Mss { value: HashMap::new() }),
});
let response = service.signal_service(request).await;
assert!(response.is_ok());
let signal_response = response.unwrap().into_inner();
assert!(!signal_response.success);
assert_eq!(signal_response.error_info.as_deref(), Some("missing service signal"));
}
#[tokio::test]
async fn test_signal_service_rejects_invalid_signal_value() {
let service = create_test_node_service();
let mut vars = HashMap::new();
vars.insert(PEER_RESTSIGNAL.to_string(), "abc".to_string());
let request = Request::new(SignalServiceRequest {
vars: Some(Mss { value: vars }),
});
let response = service.signal_service(request).await;
assert!(response.is_ok());
let signal_response = response.unwrap().into_inner();
assert!(!signal_response.success);
assert_eq!(signal_response.error_info.as_deref(), Some("invalid service signal value: abc"));
}
#[tokio::test]
async fn test_signal_service_rejects_unsupported_signal() {
let service = create_test_node_service();
let mut vars = HashMap::new();
vars.insert(PEER_RESTSIGNAL.to_string(), "99".to_string());
let request = Request::new(SignalServiceRequest {
vars: Some(Mss { value: vars }),
});
let response = service.signal_service(request).await;
assert!(response.is_ok());
let signal_response = response.unwrap().into_inner();
assert!(!signal_response.success);
assert_eq!(signal_response.error_info.as_deref(), Some("unsupported service signal: 99"));
}
#[tokio::test]
async fn test_signal_service_rejects_non_dynamic_subsystem() {
let service = create_test_node_service();
let mut vars = HashMap::new();
vars.insert(PEER_RESTSIGNAL.to_string(), SERVICE_SIGNAL_RELOAD_DYNAMIC.to_string());
vars.insert(PEER_RESTSUB_SYS.to_string(), "identity_openid".to_string());
let request = Request::new(SignalServiceRequest {
vars: Some(Mss { value: vars }),
});
let response = service.signal_service(request).await;
assert!(response.is_ok());
let signal_response = response.unwrap().into_inner();
assert!(!signal_response.success);
let error_info = signal_response.error_info.expect("expected error info");
assert!(error_info.contains("unsupported dynamic config subsystem: identity_openid"));
}
#[tokio::test]
async fn test_signal_service_refresh_config_requires_object_layer() {
let service = create_test_node_service();
let mut vars = HashMap::new();
vars.insert(PEER_RESTSIGNAL.to_string(), SERVICE_SIGNAL_REFRESH_CONFIG.to_string());
let request = Request::new(SignalServiceRequest {
vars: Some(Mss { value: vars }),
});
let response = service.signal_service(request).await;
assert!(response.is_ok());
let signal_response = response.unwrap().into_inner();
assert!(!signal_response.success);
let error_info = signal_response.error_info.expect("expected error info");
assert!(error_info.contains("storage layer not initialized"));
}
#[tokio::test]
async fn test_signal_service_reload_dynamic_requires_object_layer() {
let service = create_test_node_service();
let mut vars = HashMap::new();
vars.insert(PEER_RESTSIGNAL.to_string(), SERVICE_SIGNAL_RELOAD_DYNAMIC.to_string());
vars.insert(
PEER_RESTSUB_SYS.to_string(),
rustfs_ecstore::config::com::STORAGE_CLASS_SUB_SYS.to_string(),
);
let request = Request::new(SignalServiceRequest {
vars: Some(Mss { value: vars }),
});
let response = service.signal_service(request).await;
assert!(response.is_ok());
let signal_response = response.unwrap().into_inner();
assert!(!signal_response.success);
let error_info = signal_response.error_info.expect("expected error info");
assert!(error_info.contains("storage layer not initialized"));
}
fn assert_unimplemented_status<T>(response: Result<Response<T>, Status>, method: &str) {
let err = match response {
Ok(_) => panic!("unimplemented RPC should return an error status"),
@@ -2493,10 +2659,6 @@ mod tests {
.await,
"get_all_bucket_stats",
);
assert_unimplemented_status(
service.signal_service(Request::new(SignalServiceRequest::default())).await,
"signal_service",
);
assert_unimplemented_status(
service
.background_heal_status(Request::new(BackgroundHealStatusRequest::default()))