refactor(storage): use admin API for observability reads (#3335)

This commit is contained in:
安正超
2026-06-11 08:13:15 +08:00
committed by GitHub
parent 8ae0cad667
commit 94c53af264
9 changed files with 74 additions and 50 deletions
+3 -3
View File
@@ -20,7 +20,6 @@ use crate::{
global::{GLOBAL_BOOT_TIME, GLOBAL_Endpoints, get_global_deployment_id},
new_object_layer_fn,
notification_sys::get_global_notification_sys,
store_api::StorageAPI,
};
use crate::data_usage::load_data_usage_cache;
@@ -32,6 +31,7 @@ use rustfs_protos::{
models::{PingBody, PingBodyBuilder},
proto_gen::node_service::{PingRequest, PingResponse},
};
use rustfs_storage_api::StorageAdminApi;
use std::{
collections::{HashMap, HashSet},
time::{Duration, SystemTime},
@@ -186,7 +186,7 @@ pub async fn get_local_server_property() -> ServerProperties {
// sensitive.insert(rustfs_config::ENV_RUSTFS_ACCESS_KEY.to_string());
// sensitive.insert(rustfs_config::ENV_RUSTFS_SECRET_KEY.to_string());
if let Some(store) = new_object_layer_fn() {
let storage_info = store.local_storage_info().await;
let storage_info = StorageAdminApi::local_storage_info(store.as_ref()).await;
props.state = ITEM_ONLINE.to_string();
props.disks = storage_info.disks;
} else {
@@ -252,7 +252,7 @@ pub async fn get_server_info(get_pools: bool) -> InfoMessage {
warn!("load_data_usage_from_backend end {:?}", after3 - after2);
let backend_info = store.clone().backend_info().await;
let backend_info = StorageAdminApi::backend_info(store.as_ref()).await;
let after4 = OffsetDateTime::now_utc();
+3 -2
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::{admin_server_info::get_local_server_property, new_object_layer_fn, store_api::StorageAPI};
use crate::{admin_server_info::get_local_server_property, new_object_layer_fn};
use chrono::Utc;
use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR, heal_channel::DriveState, metrics::global_metrics};
use rustfs_io_metrics::internode_metrics::global_internode_metrics;
@@ -22,6 +22,7 @@ use rustfs_madmin::metrics::{
ScannerPacingPressureSnapshot as MadminScannerPacingPressureSnapshot,
ScannerSourceCycleSnapshot as MadminScannerSourceCycleSnapshot, TimedAction as MadminTimedAction,
};
use rustfs_storage_api::StorageAdminApi;
use rustfs_utils::os::get_drive_stats;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
@@ -269,7 +270,7 @@ async fn collect_local_disks_metrics(disks: &HashSet<String>) -> HashMap<String,
};
let mut metrics = HashMap::new();
let storage_info = store.local_storage_info().await;
let storage_info = StorageAdminApi::local_storage_info(store.as_ref()).await;
for d in storage_info.disks.iter() {
if !disks.is_empty() && !disks.contains(&d.endpoint) {
continue;
+7 -4
View File
@@ -12,7 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::StorageAPI;
use crate::admin_server_info::get_commit_id;
use crate::error::{Error, Result};
use crate::global::{GLOBAL_BOOT_TIME, get_global_endpoints};
@@ -26,6 +25,7 @@ use rustfs_madmin::health::{Cpus, MemInfo, OsInfo, Partitions, ProcInfo, SysConf
use rustfs_madmin::metrics::RealtimeMetrics;
use rustfs_madmin::net::NetInfo;
use rustfs_madmin::{ItemState, ServerProperties, StorageInfo};
use rustfs_storage_api::StorageAdminApi;
use std::collections::hash_map::DefaultHasher;
use std::future::Future;
use std::hash::{Hash, Hasher};
@@ -273,7 +273,10 @@ impl NotificationSys {
join_all(futures).await
}
pub async fn storage_info<S: StorageAPI>(&self, api: &S) -> rustfs_madmin::StorageInfo {
pub async fn storage_info<S>(&self, api: &S) -> rustfs_madmin::StorageInfo
where
S: StorageAdminApi<BackendInfo = rustfs_madmin::BackendInfo, StorageInfo = rustfs_madmin::StorageInfo>,
{
let mut futures = Vec::with_capacity(self.peer_clients.len());
let endpoints = get_global_endpoints();
let peer_timeout = Duration::from_secs(5);
@@ -307,14 +310,14 @@ impl NotificationSys {
let mut replies = join_all(futures).await;
replies.push(Some(api.local_storage_info().await));
replies.push(Some(StorageAdminApi::local_storage_info(api).await));
let mut disks = Vec::new();
for info in replies.into_iter().flatten() {
disks.extend(info.disks);
}
let backend = api.backend_info().await;
let backend = StorageAdminApi::backend_info(api).await;
rustfs_madmin::StorageInfo { disks, backend }
}
+1
View File
@@ -41,6 +41,7 @@ rustfs-ecstore = { workspace = true }
rustfs-iam = { workspace = true }
rustfs-io-metrics = { workspace = true }
rustfs-notify = { workspace = true }
rustfs-storage-api = { workspace = true }
rustfs-utils = { workspace = true, features = ["ip"] }
chrono = { workspace = true }
flate2 = { workspace = true }
+7 -6
View File
@@ -34,12 +34,13 @@ use rustfs_ecstore::bucket::metadata_sys::get_quota_config;
use rustfs_ecstore::bucket::replication::GLOBAL_REPLICATION_STATS;
use rustfs_ecstore::data_usage::load_data_usage_from_backend;
use rustfs_ecstore::global::get_global_bucket_monitor;
use rustfs_ecstore::new_object_layer_fn;
use rustfs_ecstore::pools::{get_total_usable_capacity, get_total_usable_capacity_free};
use rustfs_ecstore::store_api::{BucketOperations, BucketOptions};
use rustfs_ecstore::{StorageAPI, new_object_layer_fn};
use rustfs_iam::{get_global_iam_sys, oidc::oidc_plugin_authn_metrics_snapshot};
use rustfs_io_metrics::internode_metrics::global_internode_metrics;
use rustfs_io_metrics::{ProcessStatusSnapshot, snapshot_process_resource_and_system};
use rustfs_storage_api::StorageAdminApi;
use std::collections::HashMap;
use std::time::Duration;
use sysinfo::{Networks, System};
@@ -151,7 +152,7 @@ pub async fn collect_cluster_and_health_stats() -> (ClusterStats, ClusterHealthS
return (ClusterStats::default(), ClusterHealthStats::default());
};
let storage_info = store.storage_info().await;
let storage_info = StorageAdminApi::storage_info(store.as_ref()).await;
let raw_capacity: u64 = storage_info.disks.iter().map(|d| d.total_space).sum();
let used: u64 = storage_info.disks.iter().map(|d| d.used_space).sum();
let usable_capacity = get_total_usable_capacity(&storage_info.disks, &storage_info) as u64;
@@ -525,7 +526,7 @@ pub async fn collect_disk_and_system_drive_stats() -> (Vec<DiskStats>, Vec<Drive
return (Vec::new(), Vec::new(), DriveCountStats::default());
};
let storage_info = store.storage_info().await;
let storage_info = StorageAdminApi::storage_info(store.as_ref()).await;
let disk_stats = storage_info
.disks
.iter()
@@ -710,7 +711,7 @@ pub fn collect_internode_network_stats() -> Option<NetworkStats> {
/// Collect cluster config metrics from backend parity configuration.
pub async fn collect_cluster_config_stats() -> Option<ClusterConfigStats> {
let store = new_object_layer_fn()?;
let backend = store.backend_info().await;
let backend = StorageAdminApi::backend_info(store.as_ref()).await;
Some(ClusterConfigStats {
rrs_parity: backend.rr_sc_parity.unwrap_or_default() as u32,
@@ -724,8 +725,8 @@ pub async fn collect_erasure_set_stats() -> Vec<ErasureSetStats> {
return Vec::new();
};
let storage_info = store.storage_info().await;
let backend = store.backend_info().await;
let storage_info = StorageAdminApi::storage_info(store.as_ref()).await;
let backend = StorageAdminApi::backend_info(store.as_ref()).await;
let mut grouped: HashMap<(usize, usize), ErasureSetStats> = HashMap::new();
for disk in &storage_info.disks {