gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-24 13:16:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(ecstore): route admin read internals (#3336)

Browse Source
This commit is contained in:
安正超
2026-06-11 08:43:42 +08:00
committed by GitHub
parent 94c53af264
commit f325b9f714
5 changed files with 128 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/pools.rs
+4 -3
View File
@@ -40,7 +40,7 @@ use crate::notification_sys::get_global_notification_sys;
use crate::set_disk::SetDisks;
use crate::store_api::{
BucketOperations, BucketOptions, GetObjectReader, HealOperations, MakeBucketOptions, ObjectIO, ObjectOperations,
ObjectOptions, StorageAPI,
ObjectOptions,
};
use crate::{global::GLOBAL_LifecycleSys, sets::Sets, store::ECStore};
use byteorder::{ByteOrder, LittleEndian, WriteBytesExt};
@@ -53,6 +53,7 @@ use rustfs_common::defer;
use rustfs_common::heal_channel::HealOpts;
use rustfs_concurrency::workers::Workers;
use rustfs_filemeta::{FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
use rustfs_storage_api::StorageAdminApi;
use rustfs_utils::path::{encode_dir_object, path_join, path_to_bucket_object, path_to_bucket_object_with_base_path};
use s3s::dto::{BucketLifecycleConfiguration, DefaultRetention, ReplicationConfiguration};
use serde::{Deserialize, Serialize};
@@ -1233,8 +1234,8 @@ impl ECStore {
async fn get_decommission_pool_space_info(&self, idx: usize) -> Result<PoolSpaceInfo> {
if let Some(sets) = self.pools.get(idx) {
let mut info = sets.storage_info().await;
info.backend = self.backend_info().await;
let mut info = sets.storage_info_snapshot().await;
info.backend = StorageAdminApi::backend_info(self).await;
let total = get_total_usable_capacity(&info.disks, &info);
let free = get_total_usable_capacity_free(&info.disks, &info);
crates/ecstore/src/set_disk.rs
+31 -17
View File
@@ -1557,6 +1557,34 @@ impl SetDisks {
}
}
impl SetDisks {
pub(crate) async fn storage_info_snapshot(&self) -> rustfs_madmin::StorageInfo {
let disks = self.get_disks_internal().await;
get_storage_info(&disks, &self.set_endpoints).await
}
pub(crate) async fn local_storage_info_snapshot(&self) -> rustfs_madmin::StorageInfo {
let disks = self.get_disks_internal().await;
let mut local_disks: Vec<Option<DiskStore>> = Vec::new();
let mut local_endpoints = Vec::new();
for (i, ep) in self.set_endpoints.iter().enumerate() {
if ep.is_local {
local_disks.push(disks[i].clone());
local_endpoints.push(ep.clone());
}
}
get_storage_info(&local_disks, &local_endpoints).await
}
pub(crate) async fn disk_inventory(&self) -> Vec<Option<DiskStore>> {
self.get_disks_internal().await
}
}
#[async_trait::async_trait]
impl StorageAPI for SetDisks {
#[tracing::instrument(skip(self))]
@@ -1592,30 +1620,16 @@ impl StorageAPI for SetDisks {
}
#[tracing::instrument(skip(self))]
async fn storage_info(&self) -> rustfs_madmin::StorageInfo {
let disks = self.get_disks_internal().await;
get_storage_info(&disks, &self.set_endpoints).await
self.storage_info_snapshot().await
}
#[tracing::instrument(skip(self))]
async fn local_storage_info(&self) -> rustfs_madmin::StorageInfo {
let disks = self.get_disks_internal().await;
let mut local_disks: Vec<Option<DiskStore>> = Vec::new();
let mut local_endpoints = Vec::new();
for (i, ep) in self.set_endpoints.iter().enumerate() {
if ep.is_local {
local_disks.push(disks[i].clone());
local_endpoints.push(ep.clone());
}
}
get_storage_info(&local_disks, &local_endpoints).await
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.get_disks_internal().await)
Ok(self.disk_inventory().await)
}
#[tracing::instrument(skip(self))]
crates/ecstore/src/sets.rs
+42 -35
View File
@@ -287,6 +287,46 @@ impl Sets {
self.get_disks(self.get_hashed_set_index(key))
}
pub(crate) async fn storage_info_snapshot(&self) -> rustfs_madmin::StorageInfo {
let mut futures = Vec::with_capacity(self.disk_set.len());
for set in self.disk_set.iter() {
futures.push(set.storage_info_snapshot())
}
let results = join_all(futures).await;
let mut disks = Vec::new();
for res in results.into_iter() {
disks.extend_from_slice(&res.disks);
}
rustfs_madmin::StorageInfo {
disks,
..Default::default()
}
}
pub(crate) async fn local_storage_info_snapshot(&self) -> rustfs_madmin::StorageInfo {
let mut futures = Vec::with_capacity(self.disk_set.len());
for set in self.disk_set.iter() {
futures.push(set.local_storage_info_snapshot())
}
let results = join_all(futures).await;
let mut disks = Vec::new();
for res in results.into_iter() {
disks.extend_from_slice(&res.disks);
}
rustfs_madmin::StorageInfo {
disks,
..Default::default()
}
}
fn get_hashed_set_index(&self, input: &str) -> usize {
match self.distribution_algo {
DistributionAlgoVersion::V1 => crc_hash(input, self.disk_set.len()),
@@ -867,44 +907,11 @@ impl StorageAPI for Sets {
}
#[tracing::instrument(skip(self))]
async fn storage_info(&self) -> rustfs_madmin::StorageInfo {
let mut futures = Vec::with_capacity(self.disk_set.len());
for set in self.disk_set.iter() {
futures.push(set.storage_info())
}
let results = join_all(futures).await;
let mut disks = Vec::new();
for res in results.into_iter() {
disks.extend_from_slice(&res.disks);
}
rustfs_madmin::StorageInfo {
disks,
..Default::default()
}
self.storage_info_snapshot().await
}
#[tracing::instrument(skip(self))]
async fn local_storage_info(&self) -> rustfs_madmin::StorageInfo {
let mut futures = Vec::with_capacity(self.disk_set.len());
for set in self.disk_set.iter() {
futures.push(set.local_storage_info())
}
let results = join_all(futures).await;
let mut disks = Vec::new();
for res in results.into_iter() {
disks.extend_from_slice(&res.disks);
}
rustfs_madmin::StorageInfo {
disks,
..Default::default()
}
self.local_storage_info_snapshot().await
}
#[tracing::instrument(skip(self))]
crates/ecstore/src/store/rebalance.rs
+5 -4
View File
@@ -14,6 +14,7 @@
use super::*;
use crate::config::get_global_storage_class;
use rustfs_storage_api::StorageAdminApi;
struct LatestObjectInfoCandidate {
info: Option<ObjectInfo>,
@@ -700,7 +701,7 @@ impl ECStore {
let mut drives_per_set = Vec::new();
let mut total_sets = Vec::new();
for (idx, set_count) in self.set_drive_counts().iter().enumerate() {
for (idx, set_count) in StorageAdminApi::set_drive_counts(self).iter().enumerate() {
if let Some(sc_parity) = standard_sc_parity {
standard_sc_data.push(set_count - sc_parity);
}
@@ -755,7 +756,7 @@ impl ECStore {
let mut futures = Vec::with_capacity(self.pools.len());
for pool in self.pools.iter() {
futures.push(pool.local_storage_info())
futures.push(pool.local_storage_info_snapshot())
}
let results = join_all(futures).await;
@@ -776,14 +777,14 @@ impl ECStore {
warn!("Local storage info deduplication: {} -> {}", original_count, disks.len());
}
let backend = self.backend_info().await;
let backend = StorageAdminApi::backend_info(self).await;
rustfs_madmin::StorageInfo { backend, disks }
}
#[instrument(skip(self))]
pub(super) async fn handle_get_disks(&self, pool_idx: usize, set_idx: usize) -> Result<Vec<Option<DiskStore>>> {
if pool_idx < self.pools.len() && set_idx < self.pools[pool_idx].disk_set.len() {
self.pools[pool_idx].disk_set[set_idx].get_disks(0, 0).await
Ok(self.pools[pool_idx].disk_set[set_idx].disk_inventory().await)
} else {
Err(rebalance_disk_set_lookup_error(pool_idx, set_idx, self.pools.len()))
}