From f87b2bee95b92077f9734e2647e68c208771339c Mon Sep 17 00:00:00 2001 From: mujunxiang <1948535941@qq.com> Date: Mon, 2 Dec 2024 21:18:16 +0800 Subject: [PATCH 1/4] scanner status command(1) Signed-off-by: mujunxiang <1948535941@qq.com> --- Cargo.lock | 2 + Cargo.toml | 1 + common/common/src/globals.rs | 5 + .../src/generated/proto_gen/node_service.rs | 4 +- common/protos/src/node.proto | 2 +- ecstore/Cargo.toml | 3 +- ecstore/src/admin_server_info.rs | 30 +- ecstore/src/disk/mod.rs | 12 +- ecstore/src/heal/data_scanner.rs | 25 +- ecstore/src/heal/data_scanner_metric.rs | 40 +- ecstore/src/heal/heal_commands.rs | 2 +- ecstore/src/lib.rs | 1 + ecstore/src/metrics_realtime.rs | 169 +++++ ecstore/src/peer_rest_client.rs | 12 +- ecstore/src/store_api.rs | 6 +- ecstore/src/utils/mod.rs | 1 + ecstore/src/utils/os/linux.rs | 79 +- ecstore/src/utils/os/mod.rs | 27 +- ecstore/src/utils/os/unix.rs | 4 + ecstore/src/utils/os/windows.rs | 4 + ecstore/src/utils/time.rs | 55 ++ madmin/Cargo.toml | 2 + madmin/src/health.rs | 59 +- madmin/src/info_commands.rs | 16 + madmin/src/lib.rs | 1 + madmin/src/metrics.rs | 685 ++++++++++++++++-- rustfs/src/admin/handlers.rs | 237 +++++- rustfs/src/grpc.rs | 17 +- rustfs/src/main.rs | 7 +- 29 files changed, 1363 insertions(+), 145 deletions(-) create mode 100644 ecstore/src/metrics_realtime.rs create mode 100644 ecstore/src/utils/time.rs create mode 100644 madmin/src/info_commands.rs diff --git a/Cargo.lock b/Cargo.lock index 30b83261e..18efd2c8f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1673,6 +1673,8 @@ dependencies = [ name = "madmin" version = "0.0.1" dependencies = [ + "chrono", + "common", "psutil", "serde", ] diff --git a/Cargo.toml b/Cargo.toml index e8f2eea1a..d9fefaae3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,6 +28,7 @@ async-trait = "0.1.83" backon = "1.3.0" bytes = "1.9.0" bytesize = "1.3.0" +chrono = { version = "0.4.38", features = ["serde"] } clap = { version = "4.5.21", features = ["derive"] } ecstore = { path = "./ecstore" } flatbuffers = "24.3.25" diff --git a/common/common/src/globals.rs b/common/common/src/globals.rs index 0bbe07c89..43bd6cddf 100644 --- a/common/common/src/globals.rs +++ b/common/common/src/globals.rs @@ -8,5 +8,10 @@ lazy_static! { pub static ref GLOBAL_Local_Node_Name: RwLock = RwLock::new("".to_string()); pub static ref GLOBAL_Rustfs_Host: RwLock = RwLock::new("".to_string()); pub static ref GLOBAL_Rustfs_Port: RwLock = RwLock::new("9000".to_string()); + pub static ref GLOBAL_Rustfs_Addr: RwLock = RwLock::new("".to_string()); pub static ref GLOBAL_Conn_Map: RwLock> = RwLock::new(HashMap::new()); } + +pub async fn set_global_addr(addr: &str) { + *GLOBAL_Rustfs_Addr.write().await = addr.to_string(); +} diff --git a/common/protos/src/generated/proto_gen/node_service.rs b/common/protos/src/generated/proto_gen/node_service.rs index 42353e86c..6c4aef0e1 100644 --- a/common/protos/src/generated/proto_gen/node_service.rs +++ b/common/protos/src/generated/proto_gen/node_service.rs @@ -735,8 +735,8 @@ pub struct GetMemInfoResponse { } #[derive(Clone, PartialEq, ::prost::Message)] pub struct GetMetricsRequest { - #[prost(uint64, tag = "1")] - pub metric_type: u64, + #[prost(bytes = "vec", tag = "1")] + pub metric_type: ::prost::alloc::vec::Vec, #[prost(bytes = "vec", tag = "2")] pub opts: ::prost::alloc::vec::Vec, } diff --git a/common/protos/src/node.proto b/common/protos/src/node.proto index e3c0ad667..924f4695e 100644 --- a/common/protos/src/node.proto +++ b/common/protos/src/node.proto @@ -504,7 +504,7 @@ message GetMemInfoResponse { } message GetMetricsRequest { - uint64 metric_type = 1; + bytes metric_type = 1; bytes opts = 2; } diff --git a/ecstore/Cargo.toml b/ecstore/Cargo.toml index 224fef880..4fce9a183 100644 --- a/ecstore/Cargo.toml +++ b/ecstore/Cargo.toml @@ -14,7 +14,7 @@ backon.workspace = true blake2 = "0.10.6" bytes.workspace = true common.workspace = true -chrono = { version = "0.4.38", features = ["serde"] } +chrono.workspace = true reader.workspace = true glob = "0.3.1" thiserror.workspace = true @@ -62,6 +62,7 @@ s3s-policy.workspace = true rand.workspace = true pin-project-lite.workspace = true md-5.workspace = true +madmin.workspace = true workers.workspace = true madmin.workspace = true diff --git a/ecstore/src/admin_server_info.rs b/ecstore/src/admin_server_info.rs index d0224bd47..bee9d9964 100644 --- a/ecstore/src/admin_server_info.rs +++ b/ecstore/src/admin_server_info.rs @@ -37,21 +37,21 @@ pub struct MemStats { #[derive(Debug, Default, Serialize, Deserialize)] pub struct ServerProperties { - state: String, - endpoint: String, - scheme: String, - uptime: u64, - version: String, - commit_id: String, - network: HashMap, - disks: Vec, - pool_number: i32, - pool_numbers: Vec, - mem_stats: MemStats, - max_procs: u64, - num_cpu: u64, - runtime_version: String, - rustfs_env_vars: HashMap, + pub state: String, + pub endpoint: String, + pub scheme: String, + pub uptime: u64, + pub version: String, + pub commit_id: String, + pub network: HashMap, + pub disks: Vec, + pub pool_number: i32, + pub pool_numbers: Vec, + pub mem_stats: MemStats, + pub max_procs: u64, + pub num_cpu: u64, + pub runtime_version: String, + pub rustfs_env_vars: HashMap, } async fn is_server_resolvable(endpoint: &Endpoint) -> Result<()> { diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index f3533916a..079d299b0 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -27,6 +27,7 @@ use crate::{ use endpoint::Endpoint; use futures::StreamExt; use local::LocalDisk; +use madmin::info_commands::DiskMetrics; use protos::proto_gen::node_service::{ node_service_client::NodeServiceClient, ReadAtRequest, ReadAtResponse, WriteRequest, WriteResponse, }; @@ -35,7 +36,6 @@ use serde::{Deserialize, Serialize}; use std::{ any::Any, cmp::Ordering, - collections::HashMap, fmt::Debug, io::{Cursor, SeekFrom}, path::PathBuf, @@ -521,16 +521,6 @@ pub struct DiskInfo { pub error: String, } -#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] -pub struct DiskMetrics { - api_calls: HashMap, - total_waiting: u32, - total_errors_availability: u64, - total_errors_timeout: u64, - total_writes: u64, - total_deletes: u64, -} - #[derive(Clone, Debug, Default)] pub struct Info { pub total: u64, diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index 61efee453..c428d1993 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/ecstore/src/heal/data_scanner.rs @@ -13,6 +13,7 @@ use std::{ }; use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt}; +use chrono::{DateTime, TimeZone, Utc}; use lazy_static::lazy_static; use rand::Rng; use rmp_serde::{Deserializer, Serializer}; @@ -138,7 +139,7 @@ async fn run_data_scanner() { loop { let stop_fn = ScannerMetrics::log(ScannerMetric::ScanCycle); cycle_info.current = cycle_info.next; - cycle_info.started = SystemTime::now(); + cycle_info.started = Utc::now(); { globalScannerMetrics.write().await.set_cycle(Some(cycle_info.clone())).await; } @@ -166,7 +167,7 @@ async fn run_data_scanner() { Ok(_) => { cycle_info.next += 1; cycle_info.current = 0; - cycle_info.cycle_completed.push(SystemTime::now()); + cycle_info.cycle_completed.push(Utc::now()); if cycle_info.cycle_completed.len() > DATA_USAGE_UPDATE_DIR_CYCLES as usize { cycle_info.cycle_completed = cycle_info.cycle_completed [cycle_info.cycle_completed.len() - DATA_USAGE_UPDATE_DIR_CYCLES as usize..] @@ -252,8 +253,8 @@ async fn get_cycle_scan_mode(current_cycle: u64, bitrot_start_cycle: u64, bitrot pub struct CurrentScannerCycle { pub current: u64, pub next: u64, - pub started: SystemTime, - pub cycle_completed: Vec, + pub started: DateTime, + pub cycle_completed: Vec>, } impl Default for CurrentScannerCycle { @@ -261,7 +262,7 @@ impl Default for CurrentScannerCycle { Self { current: Default::default(), next: Default::default(), - started: SystemTime::now(), + started: Utc::now(), cycle_completed: Default::default(), } } @@ -285,7 +286,7 @@ impl CurrentScannerCycle { // write "started" rmp::encode::write_str(&mut wr, "started")?; - rmp::encode::write_uint(&mut wr, system_time_to_timestamp(&self.started))?; + rmp::encode::write_sint(&mut wr, system_time_to_timestamp(&self.started))?; // write "cycle_completed" rmp::encode::write_str(&mut wr, "cycle_completed")?; @@ -328,14 +329,14 @@ impl CurrentScannerCycle { // self.next = u; // } "started" => { - let u: u64 = rmp::decode::read_int(&mut cur)?; + let u: i64 = rmp::decode::read_int(&mut cur)?; let started = timestamp_to_system_time(u); self.started = started; } "cycleCompleted" => { let mut buf = Vec::new(); let _ = cur.read_to_end(&mut buf)?; - let u: Vec = + let u: Vec> = Deserialize::deserialize(&mut Deserializer::new(&buf[..])).expect("Deserialization failed"); self.cycle_completed = u; } @@ -348,13 +349,13 @@ impl CurrentScannerCycle { } // 将 SystemTime 转换为时间戳 -fn system_time_to_timestamp(time: &SystemTime) -> u64 { - time.duration_since(UNIX_EPOCH).expect("Time went backwards").as_secs() +fn system_time_to_timestamp(time: &DateTime) -> i64 { + time.timestamp_micros() } // 将时间戳转换为 SystemTime -fn timestamp_to_system_time(timestamp: u64) -> SystemTime { - UNIX_EPOCH + std::time::Duration::new(timestamp, 0) +fn timestamp_to_system_time(timestamp: i64) -> DateTime { + DateTime::from_timestamp_micros(timestamp).unwrap_or_default() } #[derive(Clone, Debug, Default)] diff --git a/ecstore/src/heal/data_scanner_metric.rs b/ecstore/src/heal/data_scanner_metric.rs index d45b372df..628f92774 100644 --- a/ecstore/src/heal/data_scanner_metric.rs +++ b/ecstore/src/heal/data_scanner_metric.rs @@ -1,5 +1,8 @@ +use chrono::{DateTime, Utc}; +use common::globals::GLOBAL_Local_Node_Name; use common::last_minute::{AccElem, LastMinuteLatency}; use lazy_static::lazy_static; +use madmin::metrics::ScannerMetrics as M_ScannerMetrics; use std::future::Future; use std::pin::Pin; use std::sync::atomic::AtomicU64; @@ -125,7 +128,7 @@ pub type TimeSizeFn = Arc Pin + Send> pub type TimeFn = Arc Pin + Send>> + Send + Sync + 'static>; pub struct ScannerMetrics { - operations: Vec, + operations: Vec, latency: Vec, cycle_info: RwLock>, current_paths: HashMap, @@ -140,7 +143,7 @@ impl Default for ScannerMetrics { impl ScannerMetrics { pub fn new() -> Self { Self { - operations: (0..ScannerMetric::Last as usize).map(|_| AtomicU32::new(0)).collect(), + operations: (0..ScannerMetric::Last as usize).map(|_| AtomicU64::new(0)).collect(), latency: vec![LockedLastMinuteLatency::default(); ScannerMetric::LastRealtime as usize], cycle_info: RwLock::new(None), current_paths: HashMap::new(), @@ -195,6 +198,39 @@ impl ScannerMetrics { }) }) } + + pub async fn get_cycle(&self) -> Option { + let r = self.cycle_info.read().await; + if let Some(c) = r.as_ref() { + return Some(c.clone()); + } + None + } + + pub async fn get_current_paths(&self) -> Vec { + let mut res = Vec::new(); + let prefix = format!("{}/", GLOBAL_Local_Node_Name.read().await); + self.current_paths.iter().for_each(|(k, v)| { + res.push(format!("{}/{}/{}", prefix, k, v)); + }); + res + } + + pub async fn report(&self) -> M_ScannerMetrics { + let mut m = M_ScannerMetrics::default(); + if let Some(cycle) = self.get_cycle().await { + m.current_cycle = cycle.current; + m.cycles_completed_at = cycle.cycle_completed; + m.current_started = cycle.started; + } + m.collected_at = Utc::now(); + m.active_paths = self.get_current_paths().await; + for (i, v) in self.operations.iter().enumerate() { + m.life_time_ops.insert(i.to_string(), v.load(Ordering::SeqCst)); + } + + m + } } pub type CloseDiskFn = Arc Pin + Send>> + Send + Sync + 'static>; diff --git a/ecstore/src/heal/heal_commands.rs b/ecstore/src/heal/heal_commands.rs index f6eaac6ef..2943f1e5b 100644 --- a/ecstore/src/heal/heal_commands.rs +++ b/ecstore/src/heal/heal_commands.rs @@ -131,7 +131,7 @@ impl Default for HealStartSuccess { pub type HealStopSuccess = HealStartSuccess; -#[derive(Debug, Default)] +#[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct HealingDisk { pub id: String, pub heal_id: String, diff --git a/ecstore/src/lib.rs b/ecstore/src/lib.rs index a0a1ed34a..244a7d8d1 100644 --- a/ecstore/src/lib.rs +++ b/ecstore/src/lib.rs @@ -11,6 +11,7 @@ pub mod error; mod file_meta; pub mod global; pub mod heal; +pub mod metrics_realtime; pub mod notification_sys; pub mod peer; mod peer_rest_client; diff --git a/ecstore/src/metrics_realtime.rs b/ecstore/src/metrics_realtime.rs new file mode 100644 index 000000000..a219723b6 --- /dev/null +++ b/ecstore/src/metrics_realtime.rs @@ -0,0 +1,169 @@ +use std::collections::{HashMap, HashSet}; + +use chrono::Utc; +use common::globals::{GLOBAL_Local_Node_Name, GLOBAL_Rustfs_Addr}; +use madmin::metrics::{DiskIOStats, DiskMetric, RealtimeMetrics}; +use serde::{Deserialize, Serialize}; + +use crate::{ + admin_server_info::get_local_server_property, + heal::{ + data_scanner_metric::globalScannerMetrics, + heal_commands::{DRIVE_STATE_OK, DRIVE_STATE_UNFORMATTED}, + }, + new_object_layer_fn, + store_api::StorageAPI, + utils::os::get_drive_stats, +}; + +#[derive(Debug, Default, Serialize, Deserialize)] +pub struct CollectMetricsOpts { + pub hosts: HashSet, + pub disks: HashSet, + pub job_id: String, + pub dep_id: String, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +pub struct MetricType(u32); + +impl MetricType { + // 定义一些常量 + pub const NONE: MetricType = MetricType(0); + pub const SCANNER: MetricType = MetricType(1 << 0); + pub const DISK: MetricType = MetricType(1 << 1); + pub const OS: MetricType = MetricType(1 << 2); + pub const BATCH_JOBS: MetricType = MetricType(1 << 3); + pub const SITE_RESYNC: MetricType = MetricType(1 << 4); + pub const NET: MetricType = MetricType(1 << 5); + pub const MEM: MetricType = MetricType(1 << 6); + pub const CPU: MetricType = MetricType(1 << 7); + pub const RPC: MetricType = MetricType(1 << 8); + + // MetricsAll must be last. + pub const ALL: MetricType = MetricType((1 << 9) - 1); + + pub fn new(t: u32) -> Self { + Self(t) + } +} + +impl MetricType { + fn contains(&self, x: &MetricType) -> bool { + (self.0 & x.0) == x.0 + } +} + +pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts) -> RealtimeMetrics { + let mut real_time_metrics = RealtimeMetrics::default(); + if types.0 == MetricType::NONE.0 { + return real_time_metrics; + } + + let mut by_host_name = GLOBAL_Rustfs_Addr.read().await.clone(); + if !opts.hosts.is_empty() { + let server = get_local_server_property().await; + if opts.hosts.contains(&server.endpoint) { + by_host_name = server.endpoint; + } else { + return real_time_metrics; + } + } + let local_node_name = GLOBAL_Local_Node_Name.read().await.clone(); + if by_host_name.starts_with(":") && !local_node_name.starts_with(":") { + by_host_name = local_node_name; + } + + if types.contains(&MetricType::DISK) { + let mut aggr = DiskMetric { + collected_at: Utc::now(), + ..Default::default() + }; + for (name, disk) in collect_local_disks_metrics(&opts.disks).await.into_iter() { + real_time_metrics.by_disk.insert(name, disk.clone()); + aggr.merge(&disk); + } + real_time_metrics.aggregated.disk = Some(aggr); + } + + if types.contains(&MetricType::SCANNER) { + let metrics = globalScannerMetrics.read().await.report().await; + real_time_metrics.aggregated.scanner = Some(metrics); + } + RealtimeMetrics::default() +} + +async fn collect_local_disks_metrics(disks: &HashSet) -> HashMap { + let store = match new_object_layer_fn() { + Some(store) => store, + None => return HashMap::new(), + }; + + let mut metrics = HashMap::new(); + let storage_info = store.local_storage_info().await; + for d in storage_info.disks.iter() { + if !disks.is_empty() { + if !disks.contains(&d.endpoint) { + continue; + } + } + + if d.state != *DRIVE_STATE_OK && d.state != *DRIVE_STATE_UNFORMATTED { + metrics.insert( + d.endpoint.clone(), + DiskMetric { + n_disks: 1, + offline: 1, + ..Default::default() + }, + ); + continue; + } + + let mut dm = DiskMetric { + n_disks: 1, + ..Default::default() + }; + if d.healing { + dm.healing += 1; + } + + if let Some(m) = &d.metrics { + for (k, v) in m.api_calls.iter() { + if *v != 0 { + dm.life_time_ops.insert(k.clone(), *v); + } + } + for (k, v) in m.last_minute.iter() { + if v.count != 0 { + dm.last_minute.operations.insert(k.clone(), v.clone()); + } + } + } + + if let Ok(st) = get_drive_stats(d.major, d.minor) { + dm.io_stats = DiskIOStats { + read_ios: st.read_ios, + read_merges: st.read_merges, + read_sectors: st.read_sectors, + read_ticks: st.read_ticks, + write_ios: st.write_ios, + write_merges: st.write_merges, + write_sectors: st.write_sectors, + write_ticks: st.write_ticks, + current_ios: st.current_ios, + total_ticks: st.total_ticks, + req_ticks: st.req_ticks, + discard_ios: st.discard_ios, + discard_merges: st.discard_merges, + discard_sectors: st.discard_sectors, + discard_ticks: st.discard_ticks, + flush_ios: st.flush_ios, + flush_ticks: st.flush_ticks, + }; + } + metrics.insert(d.endpoint.clone(), dm); + } + + metrics +} diff --git a/ecstore/src/peer_rest_client.rs b/ecstore/src/peer_rest_client.rs index 469562e2f..4bd115912 100644 --- a/ecstore/src/peer_rest_client.rs +++ b/ecstore/src/peer_rest_client.rs @@ -7,7 +7,7 @@ use crate::{ use common::error::{Error, Result}; use madmin::{ health::{Cpus, MemInfo, OsInfo, Partitions, ProcInfo, SysConfig, SysErrors, SysService}, - metrics::{CollectMetricsOpts, MetricType, RealtimeMetrics}, + metrics::RealtimeMetrics, net::NetInfo, }; use protos::{ @@ -266,11 +266,13 @@ impl PeerRestClient { let mut client = node_service_time_out_client(&self.addr) .await .map_err(|err| Error::msg(err.to_string()))?; - let mut buf = Vec::new(); - opts.serialize(&mut Serializer::new(&mut buf))?; + let mut buf_t = Vec::new(); + t.serialize(&mut Serializer::new(&mut buf_t))?; + let mut buf_o = Vec::new(); + opts.serialize(&mut Serializer::new(&mut buf_o))?; let request = Request::new(GetMetricsRequest { - metric_type: t, - opts: buf, + metric_type: buf_t, + opts: buf_o, }); let response = client.get_metrics(request).await?.into_inner(); diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 2f9d57935..acd8d1dc1 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -1,3 +1,4 @@ +use crate::heal::heal_commands::HealingDisk; use crate::heal::heal_ops::HealSequence; use crate::{ disk::DiskStore, @@ -8,6 +9,7 @@ use crate::{ }; use futures::StreamExt; use http::HeaderMap; +use madmin::info_commands::DiskMetrics; use rmp_serde::Serializer; use s3s::{dto::StreamingBlob, Body}; use serde::{Deserialize, Serialize}; @@ -830,8 +832,8 @@ pub struct StorageDisk { pub read_latency: f64, pub write_latency: f64, pub utilization: f64, - // pub metrics: Option, - // pub heal_info: Option, + pub metrics: Option, + pub heal_info: Option, pub used_inodes: u64, pub free_inodes: u64, pub local: bool, diff --git a/ecstore/src/utils/mod.rs b/ecstore/src/utils/mod.rs index 65aac72b2..467d8cdfa 100644 --- a/ecstore/src/utils/mod.rs +++ b/ecstore/src/utils/mod.rs @@ -6,5 +6,6 @@ pub mod hash; pub mod net; pub mod os; pub mod path; +pub mod time; pub mod wildcard; pub mod xml; diff --git a/ecstore/src/utils/os/linux.rs b/ecstore/src/utils/os/linux.rs index ab9e79299..1ee5c3f5e 100644 --- a/ecstore/src/utils/os/linux.rs +++ b/ecstore/src/utils/os/linux.rs @@ -1,9 +1,15 @@ use nix::sys::stat::{self, stat}; use nix::sys::statfs::{self, statfs, FsType}; -use std::io::{Error, ErrorKind}; +use std::fs::File; +use std::io::{self, BufRead, Error, ErrorKind}; use std::path::Path; -use crate::{disk::Info, error::Result}; +use crate::{ + disk::Info, + error::{Error as e_Error, Result}, +}; + +use super::IOStats; /// returns total and free bytes available in a directory, e.g. `/`. pub fn get_info(p: impl AsRef) -> std::io::Result { @@ -110,3 +116,72 @@ pub fn same_disk(disk1: &str, disk2: &str) -> Result { Ok(stat1.st_dev == stat2.st_dev) } + +pub fn get_drive_stats(major: u32, minor: u32) -> Result { + read_drive_stats(&format!("/sys/dev/block/{}:{}/stat", major, minor)) +} + +fn read_drive_stats(stats_file: &str) -> Result { + let stats = read_stat(stats_file)?; + if stats.len() < 11 { + return Err(e_Error::from_string(format!("found invalid format while reading {}", stats_file))); + } + let mut io_stats = IOStats { + read_ios: stats[0], + read_merges: stats[1], + read_sectors: stats[2], + read_ticks: stats[3], + write_ios: stats[4], + write_merges: stats[5], + write_sectors: stats[6], + write_ticks: stats[7], + current_ios: stats[8], + total_ticks: stats[9], + req_ticks: stats[10], + ..Default::default() + }; + + if stats.len() > 14 { + io_stats.discard_ios = stats[11]; + io_stats.discard_merges = stats[12]; + io_stats.discard_sectors = stats[13]; + io_stats.discard_ticks = stats[14]; + } + Ok(io_stats) +} + +fn read_stat(file_name: &str) -> Result> { + // 打开文件 + let path = Path::new(file_name); + let file = File::open(&path)?; + + // 创建一个 BufReader + let reader = io::BufReader::new(file); + + // 读取第一行 + let mut stats = Vec::new(); + for line in reader.lines() { + let line = line?; + // 分割行并解析为 u64 + for token in line.trim().split_whitespace() { + let ui64: u64 = token.parse()?; + stats.push(ui64); + } + break; // 只读取第一行 + } + + Ok(stats) +} + +#[cfg(test)] +mod test { + use super::get_drive_stats; + + #[test] + fn test_stats() { + let major = 7; + let minor = 11; + let s = get_drive_stats(major, minor).unwrap(); + println!("{:?}", s); + } +} diff --git a/ecstore/src/utils/os/mod.rs b/ecstore/src/utils/os/mod.rs index 83c3d734b..73fda2f99 100644 --- a/ecstore/src/utils/os/mod.rs +++ b/ecstore/src/utils/os/mod.rs @@ -6,10 +6,31 @@ mod unix; mod windows; #[cfg(target_os = "linux")] -pub use linux::{get_info, same_disk}; +pub use linux::{get_drive_stats, get_info, same_disk}; // pub use linux::same_disk; #[cfg(all(unix, not(target_os = "linux")))] -pub use unix::{get_info, same_disk}; +pub use unix::{get_drive_stats, get_info, same_disk}; #[cfg(target_os = "windows")] -pub use windows::{get_info, same_disk}; +pub use windows::{get_drive_stats, get_info, same_disk}; + +#[derive(Debug, Default)] +pub struct IOStats { + pub read_ios: u64, + pub read_merges: u64, + pub read_sectors: u64, + pub read_ticks: u64, + pub write_ios: u64, + pub write_merges: u64, + pub write_sectors: u64, + pub write_ticks: u64, + pub current_ios: u64, + pub total_ticks: u64, + pub req_ticks: u64, + pub discard_ios: u64, + pub discard_merges: u64, + pub discard_sectors: u64, + pub discard_ticks: u64, + pub flush_ios: u64, + pub flush_ticks: u64, +} diff --git a/ecstore/src/utils/os/unix.rs b/ecstore/src/utils/os/unix.rs index 544f1c859..559a2a943 100644 --- a/ecstore/src/utils/os/unix.rs +++ b/ecstore/src/utils/os/unix.rs @@ -75,3 +75,7 @@ pub fn same_disk(disk1: &str, disk2: &str) -> Result { Ok(stat1.st_dev == stat2.st_dev) } + +pub fn get_drive_stats(major: u32, minor: u32) -> Result { + IOStats::default() +} diff --git a/ecstore/src/utils/os/windows.rs b/ecstore/src/utils/os/windows.rs index d5742a689..126e77912 100644 --- a/ecstore/src/utils/os/windows.rs +++ b/ecstore/src/utils/os/windows.rs @@ -134,3 +134,7 @@ fn get_fs_type(p: &[WCHAR]) -> Result { pub fn same_disk(disk1: &str, disk2: &str) -> Result { Ok(false) } + +pub fn get_drive_stats(major: u32, minor: u32) -> Result { + IOStats::default() +} diff --git a/ecstore/src/utils/time.rs b/ecstore/src/utils/time.rs new file mode 100644 index 000000000..791db2d44 --- /dev/null +++ b/ecstore/src/utils/time.rs @@ -0,0 +1,55 @@ +use std::time::Duration; + +use tracing::info; + +pub fn parse_duration(s: &str) -> Option { + if s.ends_with("ms") { + if let Ok(s) = s.trim_end_matches("ms").parse::() { + return Some(Duration::from_millis(s)); + } + } else if s.ends_with("s") { + if let Ok(s) = s.trim_end_matches('s').parse::() { + return Some(Duration::from_secs(s)); + } + } else if s.ends_with("m") { + if let Ok(s) = s.trim_end_matches('m').parse::() { + return Some(Duration::from_secs(s * 60)); + } + } else if s.ends_with("h") { + if let Ok(s) = s.trim_end_matches('h').parse::() { + return Some(Duration::from_secs(s * 60 * 60)); + } + } + info!("can not parse duration, s: {}", s); + None +} + +#[cfg(test)] +mod test { + use std::time::Duration; + + use super::parse_duration; + + #[test] + fn test_parse_dur() { + let s = String::from("3s"); + let dur = parse_duration(&s); + println!("{:?}", dur); + assert_eq!(Some(Duration::from_secs(3)), dur); + + let s = String::from("3ms"); + let dur = parse_duration(&s); + println!("{:?}", dur); + assert_eq!(Some(Duration::from_millis(3)), dur); + + let s = String::from("3m"); + let dur = parse_duration(&s); + println!("{:?}", dur); + assert_eq!(Some(Duration::from_secs(3 * 60)), dur); + + let s = String::from("3h"); + let dur = parse_duration(&s); + println!("{:?}", dur); + assert_eq!(Some(Duration::from_secs(3 * 60 * 60)), dur); + } +} diff --git a/madmin/Cargo.toml b/madmin/Cargo.toml index f312d26a6..77235f0f2 100644 --- a/madmin/Cargo.toml +++ b/madmin/Cargo.toml @@ -7,5 +7,7 @@ rust-version.workspace = true version.workspace = true [dependencies] +chrono.workspace = true +common.workspace = true psutil = "3.3.0" serde.workspace = true diff --git a/madmin/src/health.rs b/madmin/src/health.rs index 25d3fec9a..892073096 100644 --- a/madmin/src/health.rs +++ b/madmin/src/health.rs @@ -2,25 +2,26 @@ use std::collections::HashMap; use serde::{Deserialize, Serialize}; -#[derive(Debug, Default, Serialize, Deserialize)] +#[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct NodeCommon { pub addr: String, - pub error: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, } #[derive(Debug, Default, Serialize, Deserialize)] pub struct Cpu { - vendor_id: String, - family: String, - model: String, - stepping: i32, - physical_id: String, - model_name: String, - mhz: f64, - cache_size: i32, - flags: Vec, - microcode: String, - cores: u64, + pub vendor_id: String, + pub family: String, + pub model: String, + pub stepping: i32, + pub physical_id: String, + pub model_name: String, + pub mhz: f64, + pub cache_size: i32, + pub flags: Vec, + pub microcode: String, + pub cores: u64, } #[derive(Debug, Default, Serialize, Deserialize)] @@ -157,19 +158,29 @@ pub fn get_sys_errors(_add: &str) -> SysErrors { SysErrors::default() } -#[derive(Debug, Default, Serialize, Deserialize)] +#[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct MemInfo { node_common: NodeCommon, - total: u64, - used: u64, - free: u64, - available: u64, - shared: u64, - cache: u64, - buffers: u64, - swap_space_total: u64, - swap_space_free: u64, - limit: u64, + #[serde(skip_serializing_if = "Option::is_none")] + total: Option, + #[serde(skip_serializing_if = "Option::is_none")] + used: Option, + #[serde(skip_serializing_if = "Option::is_none")] + free: Option, + #[serde(skip_serializing_if = "Option::is_none")] + available: Option, + #[serde(skip_serializing_if = "Option::is_none")] + shared: Option, + #[serde(skip_serializing_if = "Option::is_none")] + cache: Option, + #[serde(skip_serializing_if = "Option::is_none")] + buffers: Option, + #[serde(rename = "swap_space_total", skip_serializing_if = "Option::is_none")] + swap_space_total: Option, + #[serde(rename = "swap_space_free", skip_serializing_if = "Option::is_none")] + swap_space_free: Option, + #[serde(skip_serializing_if = "Option::is_none")] + limit: Option, } pub fn get_mem_info(_addr: &str) -> MemInfo { diff --git a/madmin/src/info_commands.rs b/madmin/src/info_commands.rs new file mode 100644 index 000000000..17f75e185 --- /dev/null +++ b/madmin/src/info_commands.rs @@ -0,0 +1,16 @@ +use std::collections::HashMap; + +use serde::{Deserialize, Serialize}; + +use crate::metrics::TimedAction; + +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct DiskMetrics { + pub last_minute: HashMap, + pub api_calls: HashMap, + pub total_waiting: u32, + pub total_errors_availability: u64, + pub total_errors_timeout: u64, + pub total_writes: u64, + pub total_deletes: u64, +} diff --git a/madmin/src/lib.rs b/madmin/src/lib.rs index ddd593455..908eb69da 100644 --- a/madmin/src/lib.rs +++ b/madmin/src/lib.rs @@ -1,3 +1,4 @@ pub mod health; +pub mod info_commands; pub mod metrics; pub mod net; diff --git a/madmin/src/metrics.rs b/madmin/src/metrics.rs index 15abc0bfc..fee5b365c 100644 --- a/madmin/src/metrics.rs +++ b/madmin/src/metrics.rs @@ -1,69 +1,656 @@ -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; +use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; -#[derive(Debug, Default, Serialize, Deserialize)] +use crate::health::MemInfo; + +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] pub struct TimedAction { - count: u64, - acc_time: u64, - bytes: u64, + #[serde(rename = "count")] + pub count: u64, + #[serde(rename = "acc_time_ns")] + pub acc_time: u64, + #[serde(rename = "bytes")] + pub bytes: u64, } -#[derive(Debug, Default, Serialize, Deserialize)] +impl TimedAction { + pub fn merge(&mut self, other: &TimedAction) { + self.count += other.count; + self.acc_time += other.acc_time; + self.bytes += other.bytes; + } +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct DiskIOStats { - read_ios: u64, - read_merges: u64, - read_sectors: u64, - read_ticks: u64, - write_ios: u64, - write_merges: u64, - write_sectors: u64, - write_ticks: u64, - current_ios: u64, - total_ticks: u64, - req_ticks: u64, - discard_ios: u64, - discard_merges: u64, - discard_sectors: u64, - discard_ticks: u64, - flush_ios: u64, - flush_ticks: u64, + #[serde(rename = "read_ios")] + pub read_ios: u64, + #[serde(rename = "read_merges")] + pub read_merges: u64, + #[serde(rename = "read_sectors")] + pub read_sectors: u64, + #[serde(rename = "read_ticks")] + pub read_ticks: u64, + #[serde(rename = "write_ios")] + pub write_ios: u64, + #[serde(rename = "write_merges")] + pub write_merges: u64, + #[serde(rename = "write_sectors")] + pub write_sectors: u64, + #[serde(rename = "write_ticks")] + pub write_ticks: u64, + #[serde(rename = "current_ios")] + pub current_ios: u64, + #[serde(rename = "total_ticks")] + pub total_ticks: u64, + #[serde(rename = "req_ticks")] + pub req_ticks: u64, + #[serde(rename = "discard_ios")] + pub discard_ios: u64, + #[serde(rename = "discard_merges")] + pub discard_merges: u64, + #[serde(rename = "discard_secotrs")] + pub discard_sectors: u64, + #[serde(rename = "discard_ticks")] + pub discard_ticks: u64, + #[serde(rename = "flush_ios")] + pub flush_ios: u64, + #[serde(rename = "flush_ticks")] + pub flush_ticks: u64, } -#[derive(Debug, Default, Serialize, Deserialize)] +#[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct DiskMetric { - collected_at: u64, - n_disks: usize, - offline: usize, - healing: usize, - life_time_ops: HashMap, - last_minute: HashMap, - io_stats: DiskIOStats, + #[serde(rename = "collected")] + pub collected_at: DateTime, + #[serde(rename = "n_disks")] + pub n_disks: usize, + #[serde(rename = "offline")] + pub offline: usize, + #[serde(rename = "healing")] + pub healing: usize, + #[serde(rename = "life_time_ops")] + pub life_time_ops: HashMap, + #[serde(rename = "last_minute")] + pub last_minute: Operations, + #[serde(rename = "iostats")] + pub io_stats: DiskIOStats, } -#[derive(Debug, Default, Serialize, Deserialize)] -pub struct Metrics {} +impl DiskMetric { + pub fn merge(&mut self, other: &DiskMetric) { + if self.collected_at < other.collected_at { + self.collected_at = other.collected_at; + } + self.n_disks += other.n_disks; + self.offline += other.offline; + self.healing += other.healing; + + for (k, v) in other.life_time_ops.iter() { + *self.life_time_ops.entry(k.clone()).or_insert(0) += v; + } + + for (k, v) in other.last_minute.operations.iter() { + self.last_minute.operations.entry(k.clone()).or_default().merge(v); + } + } +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct LastMinute { + #[serde(rename = "actions")] + pub actions: HashMap, + #[serde(rename = "ilm")] + pub ilm: HashMap, +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct ScannerMetrics { + #[serde(rename = "collected")] + pub collected_at: DateTime, + #[serde(rename = "current_cycle")] + pub current_cycle: u64, + #[serde(rename = "current_started")] + pub current_started: DateTime, + #[serde(rename = "cycle_complete_times")] + pub cycles_completed_at: Vec>, + #[serde(rename = "ongoing_buckets")] + pub ongoing_buckets: usize, + #[serde(rename = "life_time_ops")] + pub life_time_ops: HashMap, + #[serde(rename = "ilm_ops")] + pub life_time_ilm: HashMap, + #[serde(rename = "last_minute")] + pub last_minute: LastMinute, + #[serde(rename = "active")] + pub active_paths: Vec, +} + +impl ScannerMetrics { + pub fn merge(&mut self, other: &Self) { + if self.collected_at < other.collected_at { + self.collected_at = other.collected_at; + } + + if self.ongoing_buckets < other.ongoing_buckets { + self.ongoing_buckets = other.ongoing_buckets; + } + + if self.current_cycle < other.current_cycle { + self.current_cycle = other.current_cycle; + self.cycles_completed_at = other.cycles_completed_at.clone(); + self.current_started = other.current_started; + } + + if other.cycles_completed_at.len() > self.cycles_completed_at.len() { + self.cycles_completed_at = other.cycles_completed_at.clone(); + } + + if !other.life_time_ops.is_empty() && self.life_time_ops.is_empty() { + self.life_time_ops = other.life_time_ops.clone(); + } + + for (k, v) in other.life_time_ops.iter() { + *self.life_time_ops.entry(k.clone()).or_default() += v; + } + + for (k, v) in other.last_minute.actions.iter() { + self.last_minute.actions.entry(k.clone()).or_default().merge(v); + } + + for (k, v) in other.life_time_ilm.iter() { + *self.life_time_ilm.entry(k.clone()).or_default() += v; + } + + for (k, v) in other.last_minute.ilm.iter() { + self.last_minute.ilm.entry(k.clone()).or_default().merge(v); + } + + self.active_paths.extend(other.active_paths.clone().into_iter()); + + self.active_paths.sort(); + } +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct Metrics { + #[serde(rename = "scanner", skip_serializing_if = "Option::is_none")] + pub scanner: Option, + #[serde(rename = "disk", skip_serializing_if = "Option::is_none")] + pub disk: Option, + #[serde(rename = "os", skip_serializing_if = "Option::is_none")] + pub os: Option, + #[serde(rename = "batchJobs", skip_serializing_if = "Option::is_none")] + pub batch_jobs: Option, + #[serde(rename = "siteResync", skip_serializing_if = "Option::is_none")] + pub site_resync: Option, + #[serde(rename = "net", skip_serializing_if = "Option::is_none")] + pub net: Option, + #[serde(rename = "mem", skip_serializing_if = "Option::is_none")] + pub mem: Option, + #[serde(rename = "cpu", skip_serializing_if = "Option::is_none")] + pub cpu: Option, + #[serde(rename = "rpc", skip_serializing_if = "Option::is_none")] + pub rpc: Option, +} + +impl Metrics { + pub fn merge(&mut self, other: &Self) { + if let Some(scanner) = other.scanner.as_ref() { + match self.scanner { + Some(ref mut s_scanner) => s_scanner.merge(scanner), + None => self.scanner = Some(scanner.clone()), + } + } + + if let Some(disk) = other.disk.as_ref() { + match self.disk { + Some(ref mut s_disk) => s_disk.merge(disk), + None => self.disk = Some(disk.clone()), + } + } + + if let Some(os) = other.os.as_ref() { + match self.os { + Some(ref mut s_os) => s_os.merge(os), + None => self.os = Some(os.clone()), + } + } + + if let Some(batch_jobs) = other.batch_jobs.as_ref() { + match self.batch_jobs { + Some(ref mut s_batch_jobs) => s_batch_jobs.merge(batch_jobs), + None => self.batch_jobs = Some(batch_jobs.clone()), + } + } + + if let Some(site_resync) = other.site_resync.as_ref() { + match self.site_resync { + Some(ref mut s_site_resync) => s_site_resync.merge(site_resync), + None => self.site_resync = Some(site_resync.clone()), + } + } + + if let Some(net) = other.net.as_ref() { + match self.net { + Some(ref mut s_net) => s_net.merge(net), + None => self.net = Some(net.clone()), + } + } + + if let Some(rpc) = other.rpc.as_ref() { + match self.rpc { + Some(ref mut s_rpc) => s_rpc.merge(rpc), + None => self.rpc = Some(rpc.clone()), + } + } + } +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct RPCMetrics { + #[serde(rename = "collectedAt")] + pub collected_at: DateTime, + + pub connected: i32, + + #[serde(rename = "reconnectCount")] + pub reconnect_count: i32, + + pub disconnected: i32, + + #[serde(rename = "outgoingStreams")] + pub outgoing_streams: i32, + + #[serde(rename = "incomingStreams")] + pub incoming_streams: i32, + + #[serde(rename = "outgoingBytes")] + pub outgoing_bytes: i64, + + #[serde(rename = "incomingBytes")] + pub incoming_bytes: i64, + + #[serde(rename = "outgoingMessages")] + pub outgoing_messages: i64, + + #[serde(rename = "incomingMessages")] + pub incoming_messages: i64, + + pub out_queue: i32, + + #[serde(rename = "lastPongTime")] + pub last_pong_time: DateTime, + + #[serde(rename = "lastPingMS")] + pub last_ping_ms: f64, + + #[serde(rename = "maxPingDurMS")] + pub max_ping_dur_ms: f64, // Maximum across all merged entries. + + #[serde(rename = "lastConnectTime")] + pub last_connect_time: DateTime, + + #[serde(rename = "byDestination", skip_serializing_if = "Option::is_none")] + pub by_destination: Option>, + + #[serde(rename = "byCaller", skip_serializing_if = "Option::is_none")] + pub by_caller: Option>, +} + +impl RPCMetrics { + pub fn merge(&mut self, other: &Self) { + if self.collected_at < other.collected_at { + self.collected_at = other.collected_at; + } + + if self.last_connect_time < other.last_connect_time { + self.last_connect_time = other.last_connect_time; + } + + self.connected += other.connected; + self.disconnected += other.disconnected; + self.reconnect_count += other.reconnect_count; + self.outgoing_streams += other.outgoing_streams; + self.incoming_streams += other.incoming_streams; + self.outgoing_bytes += other.outgoing_bytes; + self.incoming_bytes += other.incoming_bytes; + self.outgoing_messages += other.outgoing_messages; + self.incoming_messages += other.incoming_messages; + self.out_queue += other.out_queue; + + if self.last_pong_time < other.last_pong_time { + self.last_pong_time = other.last_pong_time; + self.last_ping_ms = other.last_ping_ms; + } + + if self.max_ping_dur_ms < other.max_ping_dur_ms { + self.max_ping_dur_ms = other.max_ping_dur_ms; + } + + if let Some(by_destination) = other.by_destination.as_ref() { + match self.by_destination.as_mut() { + Some(s_by_de) => { + for (key, value) in by_destination { + s_by_de + .entry(key.to_string()) + .and_modify(|v| v.merge(value)) + .or_insert(value.clone()); + } + } + None => self.by_destination = Some(by_destination.clone()), + } + } + + if let Some(by_caller) = other.by_caller.as_ref() { + match self.by_caller.as_mut() { + Some(s_by_caller) => { + for (key, value) in by_caller { + s_by_caller + .entry(key.to_string()) + .and_modify(|v| v.merge(value)) + .or_insert(value.clone()); + } + } + None => self.by_caller = Some(by_caller.clone()), + } + } + } +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct CPUMetrics {} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct NetMetrics { + #[serde(rename = "collected")] + pub collected_at: DateTime, + #[serde(rename = "interfaceName")] + pub interface_name: String, + #[serde(rename = "netstats")] + pub net_stats: NetDevLine, +} + +impl NetMetrics { + pub fn merge(&mut self, other: &Self) { + if self.collected_at < other.collected_at { + self.collected_at = other.collected_at; + } + + self.net_stats.rx_bytes += other.net_stats.rx_bytes; + self.net_stats.rx_packets += other.net_stats.rx_packets; + self.net_stats.rx_errors += other.net_stats.rx_errors; + self.net_stats.rx_dropped += other.net_stats.rx_dropped; + self.net_stats.rx_fifo += other.net_stats.rx_fifo; + self.net_stats.rx_frame += other.net_stats.rx_frame; + self.net_stats.rx_compressed += other.net_stats.rx_compressed; + self.net_stats.rx_multicast += other.net_stats.rx_multicast; + self.net_stats.tx_bytes += other.net_stats.tx_bytes; + self.net_stats.tx_packets += other.net_stats.tx_packets; + self.net_stats.tx_errors += other.net_stats.tx_errors; + self.net_stats.tx_dropped += other.net_stats.tx_dropped; + self.net_stats.tx_fifo += other.net_stats.tx_fifo; + self.net_stats.tx_collisions += other.net_stats.tx_collisions; + self.net_stats.tx_carrier += other.net_stats.tx_carrier; + self.net_stats.tx_compressed += other.net_stats.tx_compressed; + } +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct NetDevLine { + #[serde(rename = "name")] + pub name: String, // The name of the interface. + + #[serde(rename = "rx_bytes")] + pub rx_bytes: u64, // Cumulative count of bytes received. + + #[serde(rename = "rx_packets")] + pub rx_packets: u64, // Cumulative count of packets received. + + #[serde(rename = "rx_errors")] + pub rx_errors: u64, // Cumulative count of receive errors encountered. + + #[serde(rename = "rx_dropped")] + pub rx_dropped: u64, // Cumulative count of packets dropped while receiving. + + #[serde(rename = "rx_fifo")] + pub rx_fifo: u64, // Cumulative count of FIFO buffer errors. + + #[serde(rename = "rx_frame")] + pub rx_frame: u64, // Cumulative count of packet framing errors. + + #[serde(rename = "rx_compressed")] + pub rx_compressed: u64, // Cumulative count of compressed packets received by the device driver. + + #[serde(rename = "rx_multicast")] + pub rx_multicast: u64, // Cumulative count of multicast frames received by the device driver. + + #[serde(rename = "tx_bytes")] + pub tx_bytes: u64, // Cumulative count of bytes transmitted. + + #[serde(rename = "tx_packets")] + pub tx_packets: u64, // Cumulative count of packets transmitted. + + #[serde(rename = "tx_errors")] + pub tx_errors: u64, // Cumulative count of transmit errors encountered. + + #[serde(rename = "tx_dropped")] + pub tx_dropped: u64, // Cumulative count of packets dropped while transmitting. + + #[serde(rename = "tx_fifo")] + pub tx_fifo: u64, // Cumulative count of FIFO buffer errors. + + #[serde(rename = "tx_collisions")] + pub tx_collisions: u64, // Cumulative count of collisions detected on the interface. + + #[serde(rename = "tx_carrier")] + pub tx_carrier: u64, // Cumulative count of carrier losses detected by the device driver. + + #[serde(rename = "tx_compressed")] + pub tx_compressed: u64, // Cumulative count of compressed packets transmitted by the device driver. +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct MemMetrics { + #[serde(rename = "collected")] + pub collected_at: DateTime, + #[serde(rename = "memInfo")] + pub info: MemInfo, +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct SiteResyncMetrics { + #[serde(rename = "collected")] + pub collected_at: DateTime, + #[serde(rename = "resyncStatus", skip_serializing_if = "Option::is_none")] + pub resync_status: Option, + #[serde(rename = "startTime")] + pub start_time: DateTime, + #[serde(rename = "lastUpdate")] + pub last_update: DateTime, + #[serde(rename = "numBuckets")] + pub num_buckets: i64, + #[serde(rename = "resyncID")] + pub resync_id: String, + #[serde(rename = "deplID")] + pub depl_id: String, + #[serde(rename = "completedReplicationSize")] + pub replicated_size: i64, + #[serde(rename = "replicationCount")] + pub replicated_count: i64, + #[serde(rename = "failedReplicationSize")] + pub failed_size: i64, + #[serde(rename = "failedReplicationCount")] + pub failed_count: i64, + #[serde(rename = "failedBuckets")] + pub failed_buckets: Vec, + #[serde(rename = "bucket", skip_serializing_if = "Option::is_none")] + pub bucket: Option, + #[serde(rename = "object", skip_serializing_if = "Option::is_none")] + pub object: Option, +} + +impl SiteResyncMetrics { + pub fn merge(&mut self, other: &Self) { + if self.collected_at < other.collected_at { + *self = other.clone(); + } + } +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct BatchJobMetrics { + #[serde(rename = "collected")] + pub collected_at: DateTime, + #[serde(rename = "Jobs")] + pub jobs: HashMap, +} + +impl BatchJobMetrics { + pub fn merge(&mut self, other: &BatchJobMetrics) { + if other.jobs.is_empty() { + return; + } + + if self.collected_at < other.collected_at { + self.collected_at = other.collected_at; + } + + for (k, v) in other.jobs.clone().into_iter() { + self.jobs.insert(k, v); + } + } +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct JobMetric { + #[serde(rename = "jobID")] + pub job_id: String, + #[serde(rename = "jobType")] + pub job_type: String, + #[serde(rename = "startTime")] + pub start_time: DateTime, + #[serde(rename = "lastUpdate")] + pub last_update: DateTime, + #[serde(rename = "retryAttempts")] + pub retry_attempts: i32, + pub complete: bool, + pub failed: bool, + // Specific job type data + #[serde(skip_serializing_if = "Option::is_none")] + pub replicate: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub key_rotate: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub expired: Option, +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct ReplicateInfo { + #[serde(rename = "lastBucket")] + pub bucket: String, + #[serde(rename = "lastObject")] + pub object: String, + #[serde(rename = "objects")] + pub objects: i64, + #[serde(rename = "objectsFailed")] + pub objects_failed: i64, + #[serde(rename = "bytesTransferred")] + pub bytes_transferred: i64, + #[serde(rename = "bytesFailed")] + pub bytes_failed: i64, +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct ExpirationInfo { + #[serde(rename = "lastBucket")] + pub bucket: String, + #[serde(rename = "lastObject")] + pub object: String, + #[serde(rename = "objects")] + pub objects: i64, + #[serde(rename = "objectsFailed")] + pub objects_failed: i64, +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct KeyRotationInfo { + #[serde(rename = "lastBucket")] + pub bucket: String, + #[serde(rename = "lastObject")] + pub object: String, + #[serde(rename = "objects")] + pub objects: i64, + #[serde(rename = "objectsFailed")] + pub objects_failed: i64, +} #[derive(Debug, Default, Serialize, Deserialize)] pub struct RealtimeMetrics { - errors: Vec, - hosts: Vec, - aggregated: Metrics, - by_host: HashMap, - by_disk: HashMap, - finally: bool, + #[serde(rename = "errors")] + pub errors: Vec, + #[serde(rename = "hosts")] + pub hosts: Vec, + #[serde(rename = "aggregated")] + pub aggregated: Metrics, + #[serde(rename = "by_host")] + pub by_host: HashMap, + #[serde(rename = "by_disk")] + pub by_disk: HashMap, + #[serde(rename = "final")] + pub finally: bool, } -#[derive(Debug, Default, Serialize, Deserialize)] -pub struct CollectMetricsOpts { - hosts: HashSet, - disks: HashSet, - job_id: String, - dep_id: String, +impl RealtimeMetrics { + pub fn merge(&mut self, other: Self) { + if !other.errors.is_empty() { + self.errors.extend(other.errors.into_iter()); + } + + for (k, v) in other.by_host.into_iter() { + *self.by_host.entry(k).or_default() = v; + } + + self.hosts.extend(other.hosts.into_iter()); + self.aggregated.merge(&other.aggregated); + self.hosts.sort(); + + for (k, v) in other.by_disk.into_iter() { + self.by_disk.entry(k.to_string()).and_modify(|h| *h = v.clone()).or_insert(v); + } + } } -pub type MetricType = u64; - -pub fn collect_local_metrics(_types: MetricType, _opts: &CollectMetricsOpts) -> RealtimeMetrics { - RealtimeMetrics::default() +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct OsMetrics { + #[serde(rename = "collected")] + pub collected_at: DateTime, + #[serde(rename = "life_time_ops")] + pub life_time_ops: HashMap, + #[serde(rename = "last_minute")] + pub last_minute: Operations, +} + +impl OsMetrics { + pub fn merge(&mut self, other: &Self) { + if self.collected_at < other.collected_at { + self.collected_at = other.collected_at; + } + + for (k, v) in other.life_time_ops.iter() { + *self.life_time_ops.entry(k.clone()).or_default() += v; + } + + for (k, v) in other.last_minute.operations.iter() { + self.last_minute.operations.entry(k.clone()).or_default().merge(v); + } + } +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct Operations { + #[serde(rename = "operations")] + pub operations: HashMap, } diff --git a/rustfs/src/admin/handlers.rs b/rustfs/src/admin/handlers.rs index b355ac8e7..9db661d50 100644 --- a/rustfs/src/admin/handlers.rs +++ b/rustfs/src/admin/handlers.rs @@ -9,30 +9,41 @@ use ecstore::error::Error as ec_Error; use ecstore::global::GLOBAL_ALlHealState; use ecstore::heal::heal_commands::HealOpts; use ecstore::heal::heal_ops::new_heal_sequence; +use ecstore::metrics_realtime::{collect_local_metrics, CollectMetricsOpts, MetricType}; use ecstore::peer::is_reserved_or_invalid_bucket; use ecstore::store::is_valid_object_prefix; use ecstore::store_api::StorageAPI; use ecstore::utils::path::path_join; +use ecstore::utils::time::parse_duration; use ecstore::utils::xml; use ecstore::GLOBAL_Endpoints; use ecstore::{new_object_layer_fn, store_api::BackendInfo}; +use futures::Stream; use http::Uri; use hyper::StatusCode; +use madmin::metrics::RealtimeMetrics; use matchit::Params; -use s3s::S3ErrorCode; +use s3s::stream::{ByteStream, DynByteStream}; use s3s::{ dto::{AssumeRoleOutput, Credentials, Timestamp}, s3_error, Body, S3Error, S3Request, S3Response, S3Result, }; +use s3s::{S3ErrorCode, StdError}; use serde::{Deserialize, Serialize}; use serde_urlencoded::from_bytes; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::path::PathBuf; +use std::pin::Pin; use std::sync::Arc; +use std::task::{Context, Poll}; +use std::time::Duration as std_Duration; +use std::u64; use time::{Duration, OffsetDateTime}; use tokio::spawn; -use tokio::sync::mpsc; -use tracing::{info, warn}; +use tokio::sync::mpsc::error::TryRecvError; +use tokio::sync::mpsc::{self, Receiver}; +use tokio::time::interval; +use tracing::{error, info, warn}; pub mod service_account; @@ -228,14 +239,220 @@ impl Operation for DataUsageInfoHandler { } } +#[derive(Debug, Serialize, Deserialize)] +struct MetricsParams { + disks: String, + hosts: String, + #[serde(rename = "interval")] + tick: String, + n: u64, + types: u32, + #[serde(rename = "by-disk")] + by_disk: String, + #[serde(rename = "by-host")] + by_host: String, + #[serde(rename = "by-jobID")] + by_job_id: String, + #[serde(rename = "by-depID")] + by_dep_id: String, +} + +impl Default for MetricsParams { + fn default() -> Self { + Self { + disks: Default::default(), + hosts: Default::default(), + tick: Default::default(), + n: u64::MAX, + types: Default::default(), + by_disk: Default::default(), + by_host: Default::default(), + by_job_id: Default::default(), + by_dep_id: Default::default(), + } + } +} + +fn extract_metrics_init_params(uri: &Uri) -> MetricsParams { + let mut mp = MetricsParams::default(); + if let Some(query) = uri.query() { + let params: Vec<&str> = query.split('&').collect(); + for param in params { + let mut parts = param.split('='); + if let Some(key) = parts.next() { + if key == "disks" { + if let Some(value) = parts.next() { + mp.disks = value.to_string(); + } + } + if key == "hosts" { + if let Some(value) = parts.next() { + mp.hosts = value.to_string(); + } + } + if key == "interval" { + if let Some(value) = parts.next() { + mp.tick = value.to_string(); + } + } + if key == "n" { + if let Some(value) = parts.next() { + mp.n = value.parse::().unwrap_or(u64::MAX); + } + } + if key == "types" { + if let Some(value) = parts.next() { + mp.types = value.parse::().unwrap_or_default(); + } + } + if key == "by-disk" { + if let Some(value) = parts.next() { + mp.by_disk = value.to_string(); + } + } + if key == "by-host" { + if let Some(value) = parts.next() { + mp.by_host = value.to_string(); + } + } + if key == "by-jobID" { + if let Some(value) = parts.next() { + mp.by_job_id = value.to_string(); + } + } + if key == "by-depID" { + if let Some(value) = parts.next() { + mp.by_dep_id = value.to_string(); + } + } + } + } + } + mp +} + +struct MetricsStream { + inner: Receiver>, +} + +impl Stream for MetricsStream { + type Item = Result; + + fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + let this = Pin::into_inner(self); + match this.inner.try_recv() { + Ok(re) => Poll::Ready(Some(re)), + Err(err) => { + if err == TryRecvError::Empty { + return Poll::Pending; + } + Poll::Ready(None) + } + } + } +} + +impl ByteStream for MetricsStream {} + pub struct MetricsHandler {} #[async_trait::async_trait] impl Operation for MetricsHandler { - async fn call(&self, _req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle MetricsHandler"); + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + info!("handle MetricsHandler, req: {:?}, params: {:?}", req, params); + let Some(cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")) }; + info!("cred: {:?}", cred); - return Err(s3_error!(NotImplemented)); + let mp = extract_metrics_init_params(&req.uri); + info!("mp: {:?}", mp); + + let tick = match parse_duration(&mp.tick) { + Some(i) => i, + None => std_Duration::from_secs(1), + }; + + let mut n = mp.n; + if n == 0 { + n = u64::MAX; + } + + let types = if mp.types != 0 { + MetricType::new(mp.types) + } else { + MetricType::ALL + }; + + let disks = mp.disks.split(",").map(String::from).collect::>(); + let by_disk = mp.by_disk == "true"; + let mut disk_map = HashSet::new(); + if !disks.is_empty() && !disks[0].is_empty() { + for d in disks.iter() { + if !d.is_empty() { + disk_map.insert(d.to_string()); + } + } + } + + let job_id = mp.by_job_id; + let hosts = mp.hosts.split(",").map(String::from).collect::>(); + let by_host = mp.by_host == "true"; + let mut host_map = HashSet::new(); + if !hosts.is_empty() && !hosts[0].is_empty() { + for d in hosts.iter() { + if !d.is_empty() { + host_map.insert(d.to_string()); + } + } + } + + let d_id = mp.by_dep_id; + let mut interval = interval(tick); + + let opts = CollectMetricsOpts { + hosts: host_map, + disks: disk_map, + job_id, + dep_id: d_id, + }; + let (tx, rx) = mpsc::channel(10); + let in_stream: DynByteStream = Box::pin(MetricsStream { inner: rx }); + let body = Body::from(in_stream); + tokio::spawn(async move { + while n > 0 { + let mut m = RealtimeMetrics::default(); + let m_local = collect_local_metrics(types, &opts).await; + m.merge(m_local); + + if !by_host { + m.by_host = HashMap::new(); + } + if !by_disk { + m.by_disk = HashMap::new(); + } + + m.finally = n <= 1; + + // todo write resp + match serde_json::to_vec(&m) { + Ok(re) => { + let _ = tx.send(Ok(Bytes::from(re))).await; + } + Err(e) => { + error!("MetricsHandler: json encode failed, err: {:?}", e); + return; + } + } + + n -= 1; + if n == 0 { + break; + } + + interval.tick().await; + } + }); + + Ok(S3Response::new((StatusCode::OK, body))) } } @@ -330,10 +547,10 @@ impl Operation for HealHandler { _err_body: String, } - let heap_path = path_join(&[PathBuf::from(hip.bucket.clone()), PathBuf::from(hip.obj_prefix.clone())]); + let heal_path = path_join(&[PathBuf::from(hip.bucket.clone()), PathBuf::from(hip.obj_prefix.clone())]); if !hip.client_token.is_empty() && !hip.force_start && !hip.force_stop { match GLOBAL_ALlHealState - .pop_heal_status_json(heap_path.to_str().unwrap_or_default(), &hip.client_token) + .pop_heal_status_json(heal_path.to_str().unwrap_or_default(), &hip.client_token) .await { Ok(b) => { @@ -351,7 +568,7 @@ impl Operation for HealHandler { let tx_clone = tx.clone(); spawn(async move { match GLOBAL_ALlHealState - .stop_heal_sequence(heap_path.to_str().unwrap_or_default()) + .stop_heal_sequence(heal_path.to_str().unwrap_or_default()) .await { Ok(b) => { diff --git a/rustfs/src/grpc.rs b/rustfs/src/grpc.rs index 8ef9d42ba..9e9180af0 100644 --- a/rustfs/src/grpc.rs +++ b/rustfs/src/grpc.rs @@ -14,9 +14,12 @@ use ecstore::{ }, erasure::Writer, heal::{ + data_usage_cache::DataUsageCache, - heal_commands::{get_local_background_heal_status, HealOpts}, + + heal_commands::{get_local_background_heal_status, {get_local_background_heal_status, HealOpts}, }, + metrics_realtime::{collect_local_metrics, CollectMetricsOpts, MetricType}, new_object_layer_fn, peer::{LocalPeerS3Client, PeerS3Client}, store::{all_local_disk_path, find_local_disk}, @@ -33,6 +36,7 @@ use madmin::{ }, metrics::{collect_local_metrics, CollectMetricsOpts}, }; +use madmin::net::get_net_info; use protos::{ models::{PingBody, PingBodyBuilder}, proto_gen::node_service::{node_service_server::NodeService as Node, *}, @@ -1709,9 +1713,14 @@ impl Node for NodeService { async fn get_metrics(&self, request: Request) -> Result, Status> { let request = request.into_inner(); - let mut buf = Deserializer::new(Cursor::new(request.opts)); - let opts: CollectMetricsOpts = Deserialize::deserialize(&mut buf).unwrap(); - let info = collect_local_metrics(request.metric_type, &opts); + let mut buf_t = Deserializer::new(Cursor::new(request.metric_type)); + let t: MetricType = Deserialize::deserialize(&mut buf_t).unwrap(); + + let mut buf_o = Deserializer::new(Cursor::new(request.opts)); + let opts: CollectMetricsOpts = Deserialize::deserialize(&mut buf_o).unwrap(); + + let info = collect_local_metrics(t, &opts).await; + let mut buf = Vec::new(); if let Err(err) = info.serialize(&mut Serializer::new(&mut buf)) { return Ok(tonic::Response::new(GetMetricsResponse { diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index 0c2f6bc17..be239520d 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -4,7 +4,10 @@ mod grpc; mod service; mod storage; use clap::Parser; -use common::error::{Error, Result}; +use common::{ + error::{Error, Result}, + globals::set_global_addr, +}; use ecstore::global::set_global_rustfs_port; use ecstore::heal::background_heal_ops::init_auto_heal; use ecstore::utils::net::{self, get_available_port}; @@ -103,6 +106,8 @@ async fn run(opt: config::Opt) -> Result<()> { ); } + set_global_addr(&opt.address).await; + set_global_endpoints(endpoint_pools.as_ref().clone()); update_erasure_type(setup_type).await; From f7076393928425657b0ff9ed2b3fa1bab5c181b4 Mon Sep 17 00:00:00 2001 From: mujunxiang <1948535941@qq.com> Date: Tue, 3 Dec 2024 09:46:09 +0800 Subject: [PATCH 2/4] scanner status command(2) Signed-off-by: mujunxiang <1948535941@qq.com> --- rustfs/src/admin/handlers.rs | 24 +++++++++++------------- 1 file changed, 11 insertions(+), 13 deletions(-) diff --git a/rustfs/src/admin/handlers.rs b/rustfs/src/admin/handlers.rs index 9db661d50..519e5efc4 100644 --- a/rustfs/src/admin/handlers.rs +++ b/rustfs/src/admin/handlers.rs @@ -18,7 +18,7 @@ use ecstore::utils::time::parse_duration; use ecstore::utils::xml; use ecstore::GLOBAL_Endpoints; use ecstore::{new_object_layer_fn, store_api::BackendInfo}; -use futures::Stream; +use futures::{Stream, StreamExt}; use http::Uri; use hyper::StatusCode; use madmin::metrics::RealtimeMetrics; @@ -43,6 +43,7 @@ use tokio::spawn; use tokio::sync::mpsc::error::TryRecvError; use tokio::sync::mpsc::{self, Receiver}; use tokio::time::interval; +use tokio_stream::wrappers::ReceiverStream; use tracing::{error, info, warn}; pub mod service_account; @@ -332,23 +333,16 @@ fn extract_metrics_init_params(uri: &Uri) -> MetricsParams { } struct MetricsStream { - inner: Receiver>, + inner: ReceiverStream>, } impl Stream for MetricsStream { type Item = Result; - fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + info!("MetricsStream poll_next"); let this = Pin::into_inner(self); - match this.inner.try_recv() { - Ok(re) => Poll::Ready(Some(re)), - Err(err) => { - if err == TryRecvError::Empty { - return Poll::Pending; - } - Poll::Ready(None) - } - } + this.inner.poll_next_unpin(cx) } } @@ -415,10 +409,13 @@ impl Operation for MetricsHandler { dep_id: d_id, }; let (tx, rx) = mpsc::channel(10); - let in_stream: DynByteStream = Box::pin(MetricsStream { inner: rx }); + let in_stream: DynByteStream = Box::pin(MetricsStream { + inner: ReceiverStream::new(rx), + }); let body = Body::from(in_stream); tokio::spawn(async move { while n > 0 { + info!("loop, n: {n}"); let mut m = RealtimeMetrics::default(); let m_local = collect_local_metrics(types, &opts).await; m.merge(m_local); @@ -435,6 +432,7 @@ impl Operation for MetricsHandler { // todo write resp match serde_json::to_vec(&m) { Ok(re) => { + info!("got metrics, send it to client, m: {m:?}, re: {re:?}"); let _ = tx.send(Ok(Bytes::from(re))).await; } Err(e) => { From bedc81e973cc12ad343d8a9e165340ac3eee3b2a Mon Sep 17 00:00:00 2001 From: mujunxiang <1948535941@qq.com> Date: Tue, 3 Dec 2024 10:14:46 +0800 Subject: [PATCH 3/4] scanner status command(3) Signed-off-by: mujunxiang <1948535941@qq.com> --- ecstore/src/heal/data_scanner.rs | 4 ++-- ecstore/src/heal/data_scanner_metric.rs | 7 ++----- rustfs/src/admin/handlers.rs | 10 ++++++---- 3 files changed, 10 insertions(+), 11 deletions(-) diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index c428d1993..8562d32df 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/ecstore/src/heal/data_scanner.rs @@ -9,11 +9,11 @@ use std::{ atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}, Arc, }, - time::{Duration, SystemTime, UNIX_EPOCH}, + time::{Duration, SystemTime}, }; use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt}; -use chrono::{DateTime, TimeZone, Utc}; +use chrono::{DateTime, Utc}; use lazy_static::lazy_static; use rand::Rng; use rmp_serde::{Deserializer, Serializer}; diff --git a/ecstore/src/heal/data_scanner_metric.rs b/ecstore/src/heal/data_scanner_metric.rs index 628f92774..b0f364bdd 100644 --- a/ecstore/src/heal/data_scanner_metric.rs +++ b/ecstore/src/heal/data_scanner_metric.rs @@ -1,4 +1,4 @@ -use chrono::{DateTime, Utc}; +use chrono::Utc; use common::globals::GLOBAL_Local_Node_Name; use common::last_minute::{AccElem, LastMinuteLatency}; use lazy_static::lazy_static; @@ -10,10 +10,7 @@ use std::sync::Once; use std::time::{Duration, UNIX_EPOCH}; use std::{ collections::HashMap, - sync::{ - atomic::{AtomicU32, Ordering}, - Arc, - }, + sync::{atomic::Ordering, Arc}, time::SystemTime, }; use tokio::sync::RwLock; diff --git a/rustfs/src/admin/handlers.rs b/rustfs/src/admin/handlers.rs index 519e5efc4..1419beaed 100644 --- a/rustfs/src/admin/handlers.rs +++ b/rustfs/src/admin/handlers.rs @@ -39,10 +39,9 @@ use std::task::{Context, Poll}; use std::time::Duration as std_Duration; use std::u64; use time::{Duration, OffsetDateTime}; -use tokio::spawn; -use tokio::sync::mpsc::error::TryRecvError; -use tokio::sync::mpsc::{self, Receiver}; +use tokio::sync::mpsc::{self}; use tokio::time::interval; +use tokio::{select, spawn}; use tokio_stream::wrappers::ReceiverStream; use tracing::{error, info, warn}; @@ -446,7 +445,10 @@ impl Operation for MetricsHandler { break; } - interval.tick().await; + select! { + _ = tx.closed() => { return; } + _ = interval.tick() => {} + } } }); From fdf5555f11c405393a01284dacc6d08871631989 Mon Sep 17 00:00:00 2001 From: mujunxiang <1948535941@qq.com> Date: Tue, 3 Dec 2024 10:36:03 +0800 Subject: [PATCH 4/4] rebase main Signed-off-by: mujunxiang <1948535941@qq.com> --- ecstore/Cargo.toml | 1 - ecstore/src/peer_rest_client.rs | 8 ++++++-- rustfs/src/grpc.rs | 12 +++--------- rustfs/src/main.rs | 1 - 4 files changed, 9 insertions(+), 13 deletions(-) diff --git a/ecstore/Cargo.toml b/ecstore/Cargo.toml index 4fce9a183..316355bef 100644 --- a/ecstore/Cargo.toml +++ b/ecstore/Cargo.toml @@ -64,7 +64,6 @@ pin-project-lite.workspace = true md-5.workspace = true madmin.workspace = true workers.workspace = true -madmin.workspace = true [target.'cfg(not(windows))'.dependencies] diff --git a/ecstore/src/peer_rest_client.rs b/ecstore/src/peer_rest_client.rs index 4bd115912..d60b537d0 100644 --- a/ecstore/src/peer_rest_client.rs +++ b/ecstore/src/peer_rest_client.rs @@ -1,8 +1,12 @@ use std::{collections::HashMap, io::Cursor, time::SystemTime}; use crate::{ - admin_server_info::ServerProperties, endpoints::EndpointServerPools, global::is_dist_erasure, - heal::heal_commands::BgHealState, store_api::StorageInfo, + admin_server_info::ServerProperties, + endpoints::EndpointServerPools, + global::is_dist_erasure, + heal::heal_commands::BgHealState, + metrics_realtime::{CollectMetricsOpts, MetricType}, + store_api::StorageInfo, }; use common::error::{Error, Result}; use madmin::{ diff --git a/rustfs/src/grpc.rs b/rustfs/src/grpc.rs index 9e9180af0..25c3391d9 100644 --- a/rustfs/src/grpc.rs +++ b/rustfs/src/grpc.rs @@ -14,10 +14,8 @@ use ecstore::{ }, erasure::Writer, heal::{ - data_usage_cache::DataUsageCache, - - heal_commands::{get_local_background_heal_status, {get_local_background_heal_status, HealOpts}, + heal_commands::{get_local_background_heal_status, HealOpts}, }, metrics_realtime::{collect_local_metrics, CollectMetricsOpts, MetricType}, new_object_layer_fn, @@ -29,12 +27,8 @@ use futures::{Stream, StreamExt}; use lock::{lock_args::LockArgs, Locker, GLOBAL_LOCAL_SERVER}; use common::globals::GLOBAL_Local_Node_Name; -use madmin::net::get_net_info; -use madmin::{ - health::{ - get_cpus, get_mem_info, get_os_info, get_partitions, get_proc_info, get_sys_config, get_sys_errors, get_sys_services, - }, - metrics::{collect_local_metrics, CollectMetricsOpts}, +use madmin::health::{ + get_cpus, get_mem_info, get_os_info, get_partitions, get_proc_info, get_sys_config, get_sys_errors, get_sys_services, }; use madmin::net::get_net_info; use protos::{ diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index be239520d..4303f2fe0 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -14,7 +14,6 @@ use ecstore::utils::net::{self, get_available_port}; use ecstore::{ endpoints::EndpointServerPools, heal::data_scanner::init_data_scanner, - notification_sys::new_global_notification_sys, set_global_endpoints, store::{init_local_disks, ECStore}, update_erasure_type,