mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-21 11:56:38 +00:00
refactor(storage): remove old admin surfaces (#3340)
This commit is contained in:
@@ -31,6 +31,7 @@ use rustfs_config::notify::{
|
||||
};
|
||||
use rustfs_config::oidc::{IDENTITY_OPENID_KEYS, IDENTITY_OPENID_SUB_SYS, OIDC_REDIRECT_URI_DYNAMIC};
|
||||
use rustfs_config::{COMMENT_KEY, DEFAULT_DELIMITER, ENABLE_KEY, EnableState, RUSTFS_REGION};
|
||||
use rustfs_storage_api::StorageAdminApi;
|
||||
use rustfs_utils::path::SLASH_SEPARATOR;
|
||||
use serde_json::{Map, Value};
|
||||
use std::collections::{HashMap, HashSet};
|
||||
@@ -277,7 +278,10 @@ fn new_server_config() -> Config {
|
||||
Config::new()
|
||||
}
|
||||
|
||||
async fn new_and_save_server_config<S: StorageAPI>(api: Arc<S>) -> Result<Config> {
|
||||
async fn new_and_save_server_config<S>(api: Arc<S>) -> Result<Config>
|
||||
where
|
||||
S: StorageAPI + StorageAdminApi,
|
||||
{
|
||||
let mut cfg = new_server_config();
|
||||
lookup_configs(&mut cfg, api.clone()).await;
|
||||
save_server_config(api, &cfg).await?;
|
||||
@@ -1059,7 +1063,10 @@ pub async fn try_migrate_server_config<S: StorageAPI>(api: Arc<S>) {
|
||||
}
|
||||
|
||||
/// Handle the situation where the configuration file does not exist, create and save a new configuration
|
||||
async fn handle_missing_config<S: StorageAPI>(api: Arc<S>, context: &str) -> Result<Config> {
|
||||
async fn handle_missing_config<S>(api: Arc<S>, context: &str) -> Result<Config>
|
||||
where
|
||||
S: StorageAPI + StorageAdminApi,
|
||||
{
|
||||
warn!("Configuration not found ({}): Start initializing new configuration", context);
|
||||
let cfg = if is_first_cluster_node_local().await {
|
||||
new_and_save_server_config(api.clone()).await?
|
||||
@@ -1078,7 +1085,10 @@ fn handle_config_read_error(err: Error, file_path: &str) -> Result<Config> {
|
||||
Err(err)
|
||||
}
|
||||
|
||||
pub async fn read_config_without_migrate<S: StorageAPI>(api: Arc<S>) -> Result<Config> {
|
||||
pub async fn read_config_without_migrate<S>(api: Arc<S>) -> Result<Config>
|
||||
where
|
||||
S: StorageAPI + StorageAdminApi,
|
||||
{
|
||||
let config_file = get_config_file();
|
||||
|
||||
// Try to read the configuration file
|
||||
@@ -1089,7 +1099,10 @@ pub async fn read_config_without_migrate<S: StorageAPI>(api: Arc<S>) -> Result<C
|
||||
}
|
||||
}
|
||||
|
||||
async fn read_server_config<S: StorageAPI>(api: Arc<S>, data: &[u8]) -> Result<Config> {
|
||||
async fn read_server_config<S>(api: Arc<S>, data: &[u8]) -> Result<Config>
|
||||
where
|
||||
S: StorageAPI + StorageAdminApi,
|
||||
{
|
||||
// If the provided data is empty, try to read from the file again
|
||||
if data.is_empty() {
|
||||
let config_file = get_config_file();
|
||||
@@ -1141,14 +1154,20 @@ pub async fn save_server_config<S: StorageAPI>(api: Arc<S>, cfg: &Config) -> Res
|
||||
save_config(api, &config_file, data).await
|
||||
}
|
||||
|
||||
pub async fn lookup_configs<S: StorageAPI>(cfg: &mut Config, api: Arc<S>) {
|
||||
pub async fn lookup_configs<S>(cfg: &mut Config, api: Arc<S>)
|
||||
where
|
||||
S: StorageAPI + StorageAdminApi,
|
||||
{
|
||||
// TODO: from etcd
|
||||
if let Err(err) = apply_dynamic_config(cfg, api).await {
|
||||
error!("apply_dynamic_config err {:?}", &err);
|
||||
}
|
||||
}
|
||||
|
||||
async fn apply_dynamic_config<S: StorageAPI>(cfg: &mut Config, api: Arc<S>) -> Result<()> {
|
||||
async fn apply_dynamic_config<S>(cfg: &mut Config, api: Arc<S>) -> Result<()>
|
||||
where
|
||||
S: StorageAPI + StorageAdminApi,
|
||||
{
|
||||
for key in SUB_SYSTEMS_DYNAMIC.iter() {
|
||||
apply_dynamic_config_for_sub_sys(cfg, api.clone(), key).await?;
|
||||
}
|
||||
@@ -1156,8 +1175,11 @@ async fn apply_dynamic_config<S: StorageAPI>(cfg: &mut Config, api: Arc<S>) -> R
|
||||
Ok(())
|
||||
}
|
||||
|
||||
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();
|
||||
async fn apply_dynamic_config_for_sub_sys<S>(cfg: &mut Config, api: Arc<S>, subsys: &str) -> Result<()>
|
||||
where
|
||||
S: StorageAPI + StorageAdminApi,
|
||||
{
|
||||
let set_drive_counts = StorageAdminApi::set_drive_counts(api.as_ref());
|
||||
if subsys == STORAGE_CLASS_SUB_SYS {
|
||||
let kvs = cfg.get_value(STORAGE_CLASS_SUB_SYS, DEFAULT_DELIMITER).unwrap_or_default();
|
||||
|
||||
@@ -1210,6 +1232,7 @@ mod tests {
|
||||
use rustfs_lock::client::LockClient;
|
||||
use rustfs_lock::client::local::LocalClient;
|
||||
use rustfs_lock::{LockError, LockInfo, LockResponse, LockStats};
|
||||
use rustfs_storage_api::StorageAdminApi;
|
||||
use serde_json::Value;
|
||||
use serial_test::serial;
|
||||
use std::collections::HashMap;
|
||||
@@ -1702,6 +1725,14 @@ mod tests {
|
||||
async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result<rustfs_lock::NamespaceLockWrapper> {
|
||||
self.set_disks.new_ns_lock(bucket, object).await
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl StorageAdminApi for LockingConfigStorage {
|
||||
type BackendInfo = rustfs_madmin::BackendInfo;
|
||||
type StorageInfo = rustfs_madmin::StorageInfo;
|
||||
type Disk = crate::disk::DiskStore;
|
||||
type Error = Error;
|
||||
|
||||
async fn backend_info(&self) -> rustfs_madmin::BackendInfo {
|
||||
panic!("unused in test")
|
||||
@@ -1715,12 +1746,15 @@ mod tests {
|
||||
panic!("unused in test")
|
||||
}
|
||||
|
||||
async fn get_disks(&self, _pool_idx: usize, _set_idx: usize) -> Result<Vec<Option<crate::disk::DiskStore>>> {
|
||||
async fn disk_set_inventory(
|
||||
&self,
|
||||
_selector: rustfs_storage_api::DiskSetSelector,
|
||||
) -> Result<Vec<Option<crate::disk::DiskStore>>> {
|
||||
panic!("unused in test")
|
||||
}
|
||||
|
||||
fn set_drive_counts(&self) -> Vec<usize> {
|
||||
panic!("unused in test")
|
||||
vec![self.set_disks.set_endpoints.len()]
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1613,29 +1613,6 @@ impl StorageAPI for SetDisks {
|
||||
|
||||
Ok(NamespaceLockWrapper::new(set_lock, resource, self.locker_owner.clone()))
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn backend_info(&self) -> rustfs_madmin::BackendInfo {
|
||||
unimplemented!()
|
||||
}
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn storage_info(&self) -> rustfs_madmin::StorageInfo {
|
||||
self.storage_info_snapshot().await
|
||||
}
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn local_storage_info(&self) -> rustfs_madmin::StorageInfo {
|
||||
self.local_storage_info_snapshot().await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn get_disks(&self, _pool_idx: usize, _set_idx: usize) -> Result<Vec<Option<DiskStore>>> {
|
||||
Ok(self.disk_inventory().await)
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
fn set_drive_counts(&self) -> Vec<usize> {
|
||||
unimplemented!()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
@@ -1837,7 +1814,7 @@ impl ObjectOperations for SetDisks {
|
||||
}
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn delete_object_version(&self, bucket: &str, object: &str, fi: &FileInfo, force_del_marker: bool) -> Result<()> {
|
||||
let disks = self.get_disks(0, 0).await?;
|
||||
let disks = self.disk_inventory().await;
|
||||
let write_quorum = disks.len() / 2 + 1;
|
||||
|
||||
let mut futures = Vec::with_capacity(disks.len());
|
||||
@@ -2213,9 +2190,8 @@ impl ObjectOperations for SetDisks {
|
||||
.await
|
||||
.map_err(|e| to_object_err(e, vec![bucket, object]))?;
|
||||
|
||||
if let Ok(disks) = self.get_disks(0, 0).await {
|
||||
record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
|
||||
}
|
||||
let disks = self.disk_inventory().await;
|
||||
record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
|
||||
|
||||
let mut oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended);
|
||||
oi.replication_decision = goi.replication_decision;
|
||||
@@ -2243,9 +2219,8 @@ impl ObjectOperations for SetDisks {
|
||||
.await
|
||||
.map_err(|e| to_object_err(e, vec![bucket, object]))?;
|
||||
|
||||
if let Ok(disks) = self.get_disks(0, 0).await {
|
||||
record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
|
||||
}
|
||||
let disks = self.disk_inventory().await;
|
||||
record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
|
||||
|
||||
let mut obj_info = ObjectInfo::from_file_info(&dfi, bucket, object, opts.versioned || opts.version_suspended);
|
||||
obj_info.size = goi.size;
|
||||
@@ -2519,7 +2494,7 @@ impl ObjectOperations for SetDisks {
|
||||
let event_name = EventName::LifecycleTransition.as_str();
|
||||
let mut should_notify_transition = true;
|
||||
|
||||
let disks = self.get_disks(0, 0).await?;
|
||||
let disks = self.disk_inventory().await;
|
||||
|
||||
if let Err(err) = self.delete_object_version(bucket, object, &fi, false).await {
|
||||
should_notify_transition = false;
|
||||
|
||||
@@ -901,28 +901,6 @@ impl StorageAPI for Sets {
|
||||
async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result<NamespaceLockWrapper> {
|
||||
self.disk_set[0].new_ns_lock(bucket, object).await
|
||||
}
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn backend_info(&self) -> rustfs_madmin::BackendInfo {
|
||||
unimplemented!()
|
||||
}
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn storage_info(&self) -> rustfs_madmin::StorageInfo {
|
||||
self.storage_info_snapshot().await
|
||||
}
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn local_storage_info(&self) -> rustfs_madmin::StorageInfo {
|
||||
self.local_storage_info_snapshot().await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn get_disks(&self, _pool_idx: usize, _set_idx: usize) -> Result<Vec<Option<DiskStore>>> {
|
||||
unimplemented!()
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
fn set_drive_counts(&self) -> Vec<usize> {
|
||||
unimplemented!()
|
||||
}
|
||||
}
|
||||
|
||||
async fn _close_storage_disks(disks: &[Option<DiskStore>]) {
|
||||
|
||||
@@ -706,28 +706,6 @@ impl StorageAPI for ECStore {
|
||||
async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result<NamespaceLockWrapper> {
|
||||
self.handle_new_ns_lock(bucket, object).await
|
||||
}
|
||||
#[instrument(skip(self))]
|
||||
async fn backend_info(&self) -> rustfs_madmin::BackendInfo {
|
||||
self.handle_backend_info().await
|
||||
}
|
||||
#[instrument(skip(self))]
|
||||
async fn storage_info(&self) -> rustfs_madmin::StorageInfo {
|
||||
self.handle_storage_info().await
|
||||
}
|
||||
#[instrument(skip(self))]
|
||||
async fn local_storage_info(&self) -> rustfs_madmin::StorageInfo {
|
||||
self.handle_local_storage_info().await
|
||||
}
|
||||
|
||||
#[instrument(skip(self))]
|
||||
async fn get_disks(&self, pool_idx: usize, set_idx: usize) -> Result<Vec<Option<DiskStore>>> {
|
||||
self.handle_get_disks(pool_idx, set_idx).await
|
||||
}
|
||||
|
||||
#[instrument(skip(self))]
|
||||
fn set_drive_counts(&self) -> Vec<usize> {
|
||||
self.handle_set_drive_counts()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
|
||||
@@ -315,10 +315,8 @@ impl ECStore {
|
||||
return (idx, 0, Vec::new());
|
||||
}
|
||||
|
||||
let disk_infos = match pool.get_disks_by_key(object).get_disks(0, 0).await {
|
||||
Ok(disks) => get_disk_infos(&disks).await,
|
||||
Err(_) => Vec::new(),
|
||||
};
|
||||
let disks = pool.get_disks_by_key(object).disk_inventory().await;
|
||||
let disk_infos = get_disk_infos(&disks).await;
|
||||
|
||||
(idx, pool.set_count, disk_infos)
|
||||
}))
|
||||
|
||||
@@ -15,7 +15,6 @@
|
||||
use crate::bucket::metadata_sys::get_versioning_config;
|
||||
use crate::bucket::versioning::VersioningApi as _;
|
||||
use crate::config::storageclass;
|
||||
use crate::disk::DiskStore;
|
||||
use crate::error::{Error, Result};
|
||||
use crate::rio::{HashReader, LimitReader};
|
||||
use crate::store_utils::clean_metadata;
|
||||
|
||||
@@ -182,11 +182,4 @@ pub trait StorageAPI:
|
||||
ObjectIO + BucketOperations + ObjectOperations + ListOperations + MultipartOperations + HealOperations + Debug
|
||||
{
|
||||
async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result<NamespaceLockWrapper>;
|
||||
|
||||
async fn backend_info(&self) -> rustfs_madmin::BackendInfo;
|
||||
async fn storage_info(&self) -> rustfs_madmin::StorageInfo;
|
||||
async fn local_storage_info(&self) -> rustfs_madmin::StorageInfo;
|
||||
|
||||
async fn get_disks(&self, pool_idx: usize, set_idx: usize) -> Result<Vec<Option<DiskStore>>>;
|
||||
fn set_drive_counts(&self) -> Vec<usize>;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user