gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-21 03:46:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(storage): narrow metadata bounds (#3343)

Browse Source
This commit is contained in:
安正超
2026-06-11 12:04:02 +08:00
committed by GitHub
parent 6b86e3875c
commit 686068d8bd
5 changed files with 108 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/migration.rs
+15 -10
View File
@@ -17,7 +17,9 @@
use crate::bucket::metadata::BUCKET_METADATA_FILE;
use crate::bucket::replication::{decode_resync_file, encode_resync_file};
use crate::disk::{BUCKET_META_PREFIX, MIGRATING_META_BUCKET, RUSTFS_META_BUCKET};
use crate::store_api::{BucketOptions, ObjectOptions, PutObjReader, StorageAPI};
use crate::store_api::{
BucketOperations, BucketOptions, ListOperations, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader,
};
use http::HeaderMap;
use rustfs_policy::auth::UserIdentity;
use rustfs_policy::policy::PolicyDoc;
@@ -176,7 +178,10 @@ fn normalize_bucket_meta_blob(path: &str, data: &[u8]) -> std::result::Result<Op
/// Uses list_bucket (from disk volumes) to get bucket names, since list_objects_v2 on the legacy
/// meta bucket may not work (legacy format differs from object layer expectations).
/// Skips buckets that already exist in RustFS (idempotent).
pub async fn try_migrate_bucket_metadata<S: StorageAPI>(store: Arc<S>) {
pub async fn try_migrate_bucket_metadata<S>(store: Arc<S>)
where
S: BucketOperations + ObjectIO + ObjectOperations,
{
let buckets_list = match store
.list_bucket(&BucketOptions {
no_metadata: true,
@@ -218,13 +223,10 @@ pub async fn try_migrate_bucket_metadata<S: StorageAPI>(store: Arc<S>) {
}
}
async fn migrate_one_if_missing<S: StorageAPI>(
store: Arc<S>,
opts: &ObjectOptions,
headers: &HeaderMap,
path: &str,
label: &str,
) {
async fn migrate_one_if_missing<S>(store: Arc<S>, opts: &ObjectOptions, headers: &HeaderMap, path: &str, label: &str)
where
S: ObjectIO + ObjectOperations,
{
if store
.get_object_info(RUSTFS_META_BUCKET, path, &ObjectOptions::default())
.await
@@ -275,7 +277,10 @@ async fn migrate_one_if_missing<S: StorageAPI>(
/// Lists all objects under the IAM prefix in the source, copies each to the target if not present.
/// Skips objects that already exist in RustFS (idempotent).
/// If list_objects_v2 on the legacy bucket fails (e.g. format differs), migration is skipped.
pub async fn try_migrate_iam_config<S: StorageAPI>(store: Arc<S>) {
pub async fn try_migrate_iam_config<S>(store: Arc<S>)
where
S: ListOperations + ObjectIO + ObjectOperations,
{
let opts = ObjectOptions {
max_parity: true,
no_lock: true,
crates/ecstore/src/config/com.rs
+19 -16
View File
@@ -16,7 +16,7 @@ use crate::config::{Config, KVS, audit, notify, oidc, set_global_storage_class,
use crate::disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET};
use crate::error::{Error, Result};
use crate::global::is_first_cluster_node_local;
use crate::store_api::{ObjectInfo, ObjectOptions, PutObjReader, StorageAPI};
use crate::store_api::{ObjectIO, ObjectInfo, ObjectOperations, ObjectOptions, PutObjReader};
use http::HeaderMap;
use rustfs_config::audit::{
AUDIT_AMQP_KEYS, AUDIT_AMQP_SUB_SYS, AUDIT_KAFKA_KEYS, AUDIT_KAFKA_SUB_SYS, AUDIT_MQTT_KEYS, AUDIT_MQTT_SUB_SYS,
@@ -181,12 +181,12 @@ fn audit_target_descriptors() -> [TargetConfigDescriptor; 9] {
}
#[instrument(skip(api))]
pub async fn read_config<S: StorageAPI>(api: Arc<S>, file: &str) -> Result<Vec<u8>> {
pub async fn read_config<S: ObjectIO>(api: Arc<S>, file: &str) -> Result<Vec<u8>> {
let (data, _obj) = read_config_with_metadata(api, file, &ObjectOptions::default()).await?;
Ok(data)
}
pub async fn read_config_no_lock<S: StorageAPI>(api: Arc<S>, file: &str) -> Result<Vec<u8>> {
pub async fn read_config_no_lock<S: ObjectIO>(api: Arc<S>, file: &str) -> Result<Vec<u8>> {
let (data, _obj) = read_config_with_metadata(
api,
file,
@@ -199,7 +199,7 @@ pub async fn read_config_no_lock<S: StorageAPI>(api: Arc<S>, file: &str) -> Resu
Ok(data)
}
pub async fn read_config_with_metadata<S: StorageAPI>(
pub async fn read_config_with_metadata<S: ObjectIO>(
api: Arc<S>,
file: &str,
opts: &ObjectOptions,
@@ -227,7 +227,7 @@ pub async fn read_config_with_metadata<S: StorageAPI>(
}
#[instrument(skip(api, data))]
pub async fn save_config<S: StorageAPI>(api: Arc<S>, file: &str, data: Vec<u8>) -> Result<()> {
pub async fn save_config<S: ObjectIO>(api: Arc<S>, file: &str, data: Vec<u8>) -> Result<()> {
save_config_with_opts(
api,
file,
@@ -241,7 +241,7 @@ pub async fn save_config<S: StorageAPI>(api: Arc<S>, file: &str, data: Vec<u8>)
}
#[instrument(skip(api))]
pub async fn delete_config<S: StorageAPI>(api: Arc<S>, file: &str) -> Result<()> {
pub async fn delete_config<S: ObjectOperations>(api: Arc<S>, file: &str) -> Result<()> {
match api
.delete_object(
RUSTFS_META_BUCKET,
@@ -265,7 +265,7 @@ pub async fn delete_config<S: StorageAPI>(api: Arc<S>, file: &str) -> Result<()>
}
}
pub async fn save_config_with_opts<S: StorageAPI>(api: Arc<S>, file: &str, data: Vec<u8>, opts: &ObjectOptions) -> Result<()> {
pub async fn save_config_with_opts<S: ObjectIO>(api: Arc<S>, file: &str, data: Vec<u8>, opts: &ObjectOptions) -> Result<()> {
let mut put_data = PutObjReader::from_vec(data);
if let Err(err) = api.put_object(RUSTFS_META_BUCKET, file, &mut put_data, opts).await {
error!("save_config_with_opts: err: {:?}, file: {}", err, file);
@@ -280,7 +280,7 @@ fn new_server_config() -> Config {
async fn new_and_save_server_config<S>(api: Arc<S>) -> Result<Config>
where
S: StorageAPI + StorageAdminApi,
S: ObjectIO + StorageAdminApi,
{
let mut cfg = new_server_config();
lookup_configs(&mut cfg, api.clone()).await;
@@ -982,7 +982,10 @@ fn is_object_not_found(err: &Error) -> bool {
*err == Error::FileNotFound || matches!(err, Error::ObjectNotFound(_, _) | Error::BucketNotFound(_))
}
pub async fn try_migrate_server_config<S: StorageAPI>(api: Arc<S>) {
pub async fn try_migrate_server_config<S>(api: Arc<S>)
where
S: ObjectIO + ObjectOperations,
{
let config_file = get_config_file();
match api
.get_object_info(
@@ -1065,7 +1068,7 @@ 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>(api: Arc<S>, context: &str) -> Result<Config>
where
S: StorageAPI + StorageAdminApi,
S: ObjectIO + StorageAdminApi,
{
warn!("Configuration not found ({}): Start initializing new configuration", context);
let cfg = if is_first_cluster_node_local().await {
@@ -1087,7 +1090,7 @@ fn handle_config_read_error(err: Error, file_path: &str) -> Result<Config> {
pub async fn read_config_without_migrate<S>(api: Arc<S>) -> Result<Config>
where
S: StorageAPI + StorageAdminApi,
S: ObjectIO + StorageAdminApi,
{
let config_file = get_config_file();
@@ -1101,7 +1104,7 @@ where
async fn read_server_config<S>(api: Arc<S>, data: &[u8]) -> Result<Config>
where
S: StorageAPI + StorageAdminApi,
S: ObjectIO + StorageAdminApi,
{
// If the provided data is empty, try to read from the file again
if data.is_empty() {
@@ -1125,7 +1128,7 @@ where
Ok(cfg.merge())
}
pub async fn save_server_config<S: StorageAPI>(api: Arc<S>, cfg: &Config) -> Result<()> {
pub async fn save_server_config<S: ObjectIO>(api: Arc<S>, cfg: &Config) -> Result<()> {
let config_file = get_config_file();
let existing = match read_config(api.clone(), &config_file).await {
Ok(v) => Some(v),
@@ -1156,7 +1159,7 @@ pub async fn save_server_config<S: StorageAPI>(api: Arc<S>, cfg: &Config) -> Res
pub async fn lookup_configs<S>(cfg: &mut Config, api: Arc<S>)
where
S: StorageAPI + StorageAdminApi,
S: StorageAdminApi,
{
// TODO: from etcd
if let Err(err) = apply_dynamic_config(cfg, api).await {
@@ -1166,7 +1169,7 @@ where
async fn apply_dynamic_config<S>(cfg: &mut Config, api: Arc<S>) -> Result<()>
where
S: StorageAPI + StorageAdminApi,
S: StorageAdminApi,
{
for key in SUB_SYSTEMS_DYNAMIC.iter() {
apply_dynamic_config_for_sub_sys(cfg, api.clone(), key).await?;
@@ -1177,7 +1180,7 @@ where
async fn apply_dynamic_config_for_sub_sys<S>(cfg: &mut Config, api: Arc<S>, subsys: &str) -> Result<()>
where
S: StorageAPI + StorageAdminApi,
S: StorageAdminApi,
{
let set_drive_counts = StorageAdminApi::set_drive_counts(api.as_ref());
if subsys == STORAGE_CLASS_SUB_SYS {
crates/ecstore/src/rebalance.rs
+4 -4
View File
@@ -566,11 +566,11 @@ impl RebalanceMeta {
pub fn new() -> Self {
Self::default()
}
pub async fn load<S: StorageAPI>(&mut self, store: Arc<S>) -> Result<()> {
pub async fn load<S: ObjectIO>(&mut self, store: Arc<S>) -> Result<()> {
self.load_with_opts(store, ObjectOptions::default()).await
}
pub async fn load_with_opts<S: StorageAPI>(&mut self, store: Arc<S>, opts: ObjectOptions) -> Result<()> {
pub async fn load_with_opts<S: ObjectIO>(&mut self, store: Arc<S>, opts: ObjectOptions) -> Result<()> {
let (data, _) = read_config_with_metadata(store, REBAL_META_NAME, &opts).await?;
if data.is_empty() {
info!("rebalanceMeta load_with_opts: no data");
@@ -599,11 +599,11 @@ impl RebalanceMeta {
Ok(())
}
pub async fn save<S: StorageAPI>(&self, store: Arc<S>) -> Result<()> {
pub async fn save<S: ObjectIO>(&self, store: Arc<S>) -> Result<()> {
self.save_with_opts(store, ObjectOptions::default()).await
}
pub async fn save_with_opts<S: StorageAPI>(&self, store: Arc<S>, opts: ObjectOptions) -> Result<()> {
pub async fn save_with_opts<S: ObjectIO>(&self, store: Arc<S>, opts: ObjectOptions) -> Result<()> {
if self.pool_stats.is_empty() {
info!("rebalanceMeta save_with_opts: no pool stats");
return Ok(());
crates/ecstore/src/tier/tier.rs
+11 -9
View File
@@ -46,12 +46,11 @@ use crate::tier::{
warm_backend::{check_warm_backend, new_warm_backend},
};
use crate::{
StorageAPI,
config::com::{CONFIG_PREFIX, read_config},
disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET},
global::is_first_cluster_node_local,
store::ECStore,
store_api::{ObjectIO as _, ObjectOptions, PutObjReader},
store_api::{ObjectIO, ObjectOperations, ObjectOptions, PutObjReader},
};
use rustfs_rio::HashReader;
use rustfs_utils::path::{SLASH_SEPARATOR, path_join};
@@ -1033,14 +1032,14 @@ impl TierConfigMgr {
self.save_tiering_config(api).await
}
pub async fn save_tiering_config<S: StorageAPI>(&self, api: Arc<S>) -> std::result::Result<(), std::io::Error> {
pub async fn save_tiering_config<S: ObjectIO>(&self, api: Arc<S>) -> std::result::Result<(), std::io::Error> {
let data = encode_external_tiering_config_blob(self)?;
let config_file = tier_config_path(TIER_CONFIG_FILE);
self.save_config(api, &config_file, data).await
}
pub async fn save_config<S: StorageAPI>(
pub async fn save_config<S: ObjectIO>(
&self,
api: Arc<S>,
file: &str,
@@ -1058,7 +1057,7 @@ impl TierConfigMgr {
.await
}
pub async fn save_config_with_opts<S: StorageAPI>(
pub async fn save_config_with_opts<S: ObjectIO>(
&self,
api: Arc<S>,
file: &str,
@@ -1100,7 +1099,7 @@ impl TierConfigMgr {
}
}
async fn new_and_save_tiering_config<S: StorageAPI>(api: Arc<S>) -> Result<TierConfigMgr> {
async fn new_and_save_tiering_config<S: ObjectIO>(api: Arc<S>) -> Result<TierConfigMgr> {
let mut cfg = TierConfigMgr {
driver_cache: HashMap::new(),
tiers: HashMap::new(),
@@ -1159,7 +1158,7 @@ async fn load_tier_config(api: Arc<ECStore>) -> std::result::Result<TierConfigMg
}
}
async fn read_tier_config_from_bucket<S: StorageAPI>(
async fn read_tier_config_from_bucket<S: ObjectIO>(
api: Arc<S>,
bucket: &str,
path: &str,
@@ -1177,7 +1176,7 @@ async fn read_tier_config_from_bucket<S: StorageAPI>(
Ok(Some(data))
}
async fn write_tier_config_to_rustfs<S: StorageAPI>(api: Arc<S>, path: &str, data: Bytes) -> io::Result<()> {
async fn write_tier_config_to_rustfs<S: ObjectIO>(api: Arc<S>, path: &str, data: Bytes) -> io::Result<()> {
let mut put_data = PutObjReader::from_vec(data.to_vec());
api.put_object(
RUSTFS_META_BUCKET,
@@ -1193,7 +1192,10 @@ async fn write_tier_config_to_rustfs<S: StorageAPI>(api: Arc<S>, path: &str, dat
Ok(())
}
pub async fn try_migrate_tiering_config<S: StorageAPI>(api: Arc<S>) {
pub async fn try_migrate_tiering_config<S>(api: Arc<S>)
where
S: ObjectIO + ObjectOperations,
{
let target_path = tier_config_path(TIER_CONFIG_FILE);
if api
.get_object_info(