From 4a173bbd98db0486037f96a969ba485ef02c5e9b Mon Sep 17 00:00:00 2001 From: Nugine Date: Tue, 3 Dec 2024 12:01:53 +0800 Subject: [PATCH 1/3] ci: allow clippy warnings --- .github/workflows/rust.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index b2f2eb774..2e6570341 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -62,7 +62,8 @@ jobs: run: cargo fmt --all --check - name: cargo clippy - run: cargo clippy -- -D warnings + run: cargo clippy + # run: cargo clippy -- -D warnings - name: cargo test run: cargo test --all --exclude e2e_test From 7148c5d5cd38d5a1540fff3af50abfd170f4898c Mon Sep 17 00:00:00 2001 From: mujunxiang <1948535941@qq.com> Date: Tue, 3 Dec 2024 15:31:43 +0800 Subject: [PATCH 2/3] sanner(1) Signed-off-by: mujunxiang <1948535941@qq.com> --- ecstore/src/disk/local.rs | 1 + ecstore/src/disk/mod.rs | 1 + ecstore/src/heal/data_scanner.rs | 18 +++++++-- ecstore/src/heal/data_scanner_metric.rs | 2 + ecstore/src/heal/data_usage_cache.rs | 10 ++--- ecstore/src/metrics_realtime.rs | 50 ++++++++++++++++++++++++- ecstore/src/set_disk.rs | 2 + rustfs/src/admin/handlers.rs | 2 +- 8 files changed, 76 insertions(+), 10 deletions(-) diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index c85db05fe..6117df74b 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -2092,6 +2092,7 @@ impl DiskAPI for LocalDisk { ) .await?; data_usage_info.info.last_update = Some(SystemTime::now()); + info!("ns_scanner completed: {data_usage_info:?}"); Ok(data_usage_info) } diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 079d299b0..ab57ce45a 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -351,6 +351,7 @@ impl DiskAPI for Disk { updates: Sender, scan_mode: HealScanMode, ) -> Result { + info!("ns_scanner"); match self { Disk::Local(local_disk) => local_disk.ns_scanner(cache, updates, scan_mode).await, Disk::Remote(remote_disk) => remote_disk.ns_scanner(cache, updates, scan_mode).await, diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index 8562d32df..16ff8cc05 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/ecstore/src/heal/data_scanner.rs @@ -86,23 +86,28 @@ lazy_static! { } pub async fn init_data_scanner() { - let mut r = rand::thread_rng(); - let random = r.gen_range(0.0..1.0); tokio::spawn(async move { loop { run_data_scanner().await; + let random = { + let mut r = rand::thread_rng(); + r.gen_range(0.0..1.0) + }; let duration = Duration::from_secs_f64(random * (SCANNER_CYCLE.load(std::sync::atomic::Ordering::SeqCst) as f64)); let sleep_duration = if duration < Duration::new(1, 0) { Duration::new(1, 0) } else { duration }; + + info!("data scanner will sleeping {sleep_duration:?}"); sleep(sleep_duration).await; } }); } async fn run_data_scanner() { + info!("run_data_scanner"); let Some(store) = new_object_layer_fn() else { error!("errServerNotInitialized"); return; @@ -163,8 +168,10 @@ async fn run_data_scanner() { }); let mut res = HashMap::new(); res.insert("cycle".to_string(), cycle_info.current.to_string()); + info!("start ns_scanner"); match store.clone().ns_scanner(tx, cycle_info.current as usize, scan_mode).await { Ok(_) => { + info!("ns_scanner completed"); cycle_info.next += 1; cycle_info.current = 0; cycle_info.cycle_completed.push(Utc::now()); @@ -176,9 +183,14 @@ async fn run_data_scanner() { globalScannerMetrics.write().await.set_cycle(Some(cycle_info.clone())).await; let mut tmp = Vec::new(); tmp.write_u64::(cycle_info.next).unwrap(); - let _ = save_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH, &tmp).await; + if let Ok(data) = cycle_info.marshal_msg(&tmp) { + let _ = save_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH, &data).await; + } else { + let _ = save_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH, &tmp).await; + } } Err(err) => { + info!("ns_scanner failed: {:?}", err); res.insert("error".to_string(), err.to_string()); } } diff --git a/ecstore/src/heal/data_scanner_metric.rs b/ecstore/src/heal/data_scanner_metric.rs index b0f364bdd..77b97c1d5 100644 --- a/ecstore/src/heal/data_scanner_metric.rs +++ b/ecstore/src/heal/data_scanner_metric.rs @@ -14,6 +14,7 @@ use std::{ time::SystemTime, }; use tokio::sync::RwLock; +use tracing::info; use super::data_scanner::{CurrentScannerCycle, UpdateCurrentPathFn}; @@ -148,6 +149,7 @@ impl ScannerMetrics { } pub async fn set_cycle(&mut self, c: Option) { + info!("ScannerMetrics set_cycle {c:?}"); *self.cycle_info.write().await = c; } diff --git a/ecstore/src/heal/data_usage_cache.rs b/ecstore/src/heal/data_usage_cache.rs index 09be08b99..fbc6003b6 100644 --- a/ecstore/src/heal/data_usage_cache.rs +++ b/ecstore/src/heal/data_usage_cache.rs @@ -131,7 +131,7 @@ const OBJECTS_VERSION_COUNT_INTERVALS: [ObjectHistogramInterval; DATA_USAGE_VERS ]; // sizeHistogram is a size histogram. -#[derive(Clone, Serialize, Deserialize)] +#[derive(Clone, Debug, Serialize, Deserialize)] pub struct SizeHistogram(Vec); impl Default for SizeHistogram { @@ -168,7 +168,7 @@ impl SizeHistogram { } // versionsHistogram is a histogram of number of versions in an object. -#[derive(Clone, Serialize, Deserialize)] +#[derive(Clone, Debug, Serialize, Deserialize)] pub struct VersionsHistogram(Vec); impl Default for VersionsHistogram { @@ -238,7 +238,7 @@ impl ReplicationAllStats { } } -#[derive(Clone, Default, Serialize, Deserialize)] +#[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct DataUsageEntry { pub children: DataUsageHashMap, // These fields do no include any children. @@ -340,7 +340,7 @@ pub struct DataUsageEntryInfo { pub entry: DataUsageEntry, } -#[derive(Clone, Default, Serialize, Deserialize)] +#[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct DataUsageCacheInfo { pub name: String, pub next_cycle: u32, @@ -367,7 +367,7 @@ pub struct DataUsageCacheInfo { // } // } -#[derive(Clone, Default, Serialize, Deserialize)] +#[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct DataUsageCache { pub info: DataUsageCacheInfo, pub cache: HashMap, diff --git a/ecstore/src/metrics_realtime.rs b/ecstore/src/metrics_realtime.rs index a219723b6..97b396970 100644 --- a/ecstore/src/metrics_realtime.rs +++ b/ecstore/src/metrics_realtime.rs @@ -4,6 +4,7 @@ use chrono::Utc; use common::globals::{GLOBAL_Local_Node_Name, GLOBAL_Rustfs_Addr}; use madmin::metrics::{DiskIOStats, DiskMetric, RealtimeMetrics}; use serde::{Deserialize, Serialize}; +use tracing::info; use crate::{ admin_server_info::get_local_server_property, @@ -55,8 +56,10 @@ impl MetricType { } pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts) -> RealtimeMetrics { + info!("collect_local_metrics"); let mut real_time_metrics = RealtimeMetrics::default(); if types.0 == MetricType::NONE.0 { + info!("types is None, return"); return real_time_metrics; } @@ -75,11 +78,13 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts) } if types.contains(&MetricType::DISK) { + info!("start get disk metrics"); let mut aggr = DiskMetric { collected_at: Utc::now(), ..Default::default() }; for (name, disk) in collect_local_disks_metrics(&opts.disks).await.into_iter() { + info!("got disk metric, name: {name}, metric: {disk:?}"); real_time_metrics.by_disk.insert(name, disk.clone()); aggr.merge(&disk); } @@ -87,10 +92,31 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts) } if types.contains(&MetricType::SCANNER) { + info!("start get scanner metrics"); let metrics = globalScannerMetrics.read().await.report().await; real_time_metrics.aggregated.scanner = Some(metrics); } - RealtimeMetrics::default() + + if types.contains(&MetricType::OS) {} + + if types.contains(&MetricType::BATCH_JOBS) {} + + if types.contains(&MetricType::SITE_RESYNC) {} + + if types.contains(&MetricType::NET) {} + + if types.contains(&MetricType::MEM) {} + + if types.contains(&MetricType::CPU) {} + + if types.contains(&MetricType::RPC) {} + + real_time_metrics + .by_host + .insert(by_host_name.clone(), real_time_metrics.aggregated.clone()); + real_time_metrics.hosts.push(by_host_name); + + real_time_metrics } async fn collect_local_disks_metrics(disks: &HashSet) -> HashMap { @@ -167,3 +193,25 @@ async fn collect_local_disks_metrics(disks: &HashSet) -> HashMap, heal_scan_mode: HealScanMode, ) -> Result<()> { + info!("ns_scanner"); if buckets.is_empty() { return Ok(()); } @@ -2846,6 +2847,7 @@ impl SetDisks { } let _ = join_all(futures).await; let _ = task.await; + info!("ns_scanner completed"); Ok(()) } diff --git a/rustfs/src/admin/handlers.rs b/rustfs/src/admin/handlers.rs index 1419beaed..fcec96c97 100644 --- a/rustfs/src/admin/handlers.rs +++ b/rustfs/src/admin/handlers.rs @@ -431,7 +431,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:?}"); + info!("got metrics, send it to client, m: {m:?}"); let _ = tx.send(Ok(Bytes::from(re))).await; } Err(e) => { From 14627a5c2bff9615ed76a9da58937b00ad7e073e Mon Sep 17 00:00:00 2001 From: mujunxiang <1948535941@qq.com> Date: Tue, 3 Dec 2024 17:05:23 +0800 Subject: [PATCH 3/3] scanner(2) Signed-off-by: mujunxiang <1948535941@qq.com> --- common/lock/src/lrwmutex.rs | 5 +- ecstore/src/heal/data_scanner.rs | 80 +++++++++++++------------ ecstore/src/heal/data_scanner_metric.rs | 1 + ecstore/src/set_disk.rs | 10 +++- 4 files changed, 54 insertions(+), 42 deletions(-) diff --git a/common/lock/src/lrwmutex.rs b/common/lock/src/lrwmutex.rs index fa65f1a7c..9bc3415ef 100644 --- a/common/lock/src/lrwmutex.rs +++ b/common/lock/src/lrwmutex.rs @@ -2,6 +2,7 @@ use std::time::{Duration, Instant}; use rand::Rng; use tokio::{sync::RwLock, time::sleep}; +use tracing::info; #[derive(Debug, Default)] pub struct LRWMutex { @@ -87,14 +88,14 @@ impl LRWMutex { pub async fn un_lock(&self) { let is_write = true; if !self.unlock(is_write).await { - panic!("Trying to un_lock() while no Lock() is active") + info!("Trying to un_lock() while no Lock() is active") } } pub async fn un_r_lock(&self) { let is_write = false; if !self.unlock(is_write).await { - panic!("Trying to un_r_lock() while no Lock() is active") + info!("Trying to un_r_lock() while no Lock() is active") } } diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index 16ff8cc05..721a09f84 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/ecstore/src/heal/data_scanner.rs @@ -12,7 +12,6 @@ use std::{ time::{Duration, SystemTime}, }; -use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt}; use chrono::{DateTime, Utc}; use lazy_static::lazy_static; use rand::Rng; @@ -113,33 +112,15 @@ async fn run_data_scanner() { return; }; - let mut cycle_info = CurrentScannerCycle::default(); - - let mut buf = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH) + let buf = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH) .await .map_or(Vec::new(), |buf| buf); - match buf.len().cmp(&8) { - std::cmp::Ordering::Less => {} - std::cmp::Ordering::Equal => { - cycle_info.next = match Cursor::new(buf).read_u64::() { - Ok(buf) => buf, - Err(_) => { - error!("can not decode DATA_USAGE_BLOOM_NAME_PATH"); - return; - } - }; - } - std::cmp::Ordering::Greater => { - cycle_info.next = match Cursor::new(buf[..8].to_vec()).read_u64::() { - Ok(buf) => buf, - Err(_) => { - error!("can not decode DATA_USAGE_BLOOM_NAME_PATH"); - return; - } - }; - let _ = cycle_info.unmarshal_msg(&buf.split_off(8)); - } - } + + let mut buf_t = Deserializer::new(Cursor::new(buf)); + let mut cycle_info: CurrentScannerCycle = match Deserialize::deserialize(&mut buf_t) { + Ok(info) => info, + Err(_) => CurrentScannerCycle::default(), + }; loop { let stop_fn = ScannerMetrics::log(ScannerMetric::ScanCycle); @@ -176,18 +157,12 @@ async fn run_data_scanner() { cycle_info.current = 0; 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..] - .to_vec(); + let _ = cycle_info.cycle_completed.remove(0); } globalScannerMetrics.write().await.set_cycle(Some(cycle_info.clone())).await; - let mut tmp = Vec::new(); - tmp.write_u64::(cycle_info.next).unwrap(); - if let Ok(data) = cycle_info.marshal_msg(&tmp) { - let _ = save_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH, &data).await; - } else { - let _ = save_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH, &tmp).await; - } + let mut wr = Vec::new(); + cycle_info.serialize(&mut Serializer::new(&mut wr)).unwrap(); + let _ = save_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH, &wr).await; } Err(err) => { info!("ns_scanner failed: {:?}", err); @@ -261,7 +236,7 @@ async fn get_cycle_scan_mode(current_cycle: u64, bitrot_start_cycle: u64, bitrot HEAL_NORMAL_SCAN } -#[derive(Clone, Debug)] +#[derive(Clone, Debug, Serialize, Deserialize)] pub struct CurrentScannerCycle { pub current: u64, pub next: u64, @@ -1081,3 +1056,34 @@ pub async fn scan_data_folder( } // pub fn eval_action_from_lifecycle(lc: &BucketLifecycleConfiguration, lr: &ObjectLockConfiguration, rcfg: &ReplicationConfiguration， obj: &ObjectInfo) + +#[cfg(test)] +mod tests { + use std::io::Cursor; + + use chrono::Utc; + use rmp_serde::{Deserializer, Serializer}; + use serde::{Deserialize, Serialize}; + + use super::CurrentScannerCycle; + + #[test] + fn test_current_cycle() { + let cycle_info = CurrentScannerCycle { + current: 0, + next: 1, + started: Utc::now(), + cycle_completed: vec![Utc::now(), Utc::now()], + }; + + println!("{cycle_info:?}"); + + let mut wr = Vec::new(); + cycle_info.serialize(&mut Serializer::new(&mut wr)).unwrap(); + + let mut buf_t = Deserializer::new(Cursor::new(wr)); + let c: CurrentScannerCycle = Deserialize::deserialize(&mut buf_t).unwrap(); + + println!("{c:?}"); + } +} diff --git a/ecstore/src/heal/data_scanner_metric.rs b/ecstore/src/heal/data_scanner_metric.rs index 77b97c1d5..d96ad89c6 100644 --- a/ecstore/src/heal/data_scanner_metric.rs +++ b/ecstore/src/heal/data_scanner_metric.rs @@ -218,6 +218,7 @@ impl ScannerMetrics { pub async fn report(&self) -> M_ScannerMetrics { let mut m = M_ScannerMetrics::default(); if let Some(cycle) = self.get_cycle().await { + info!("cycle: {cycle:?}"); m.current_cycle = cycle.current; m.cycles_completed_at = cycle.cycle_completed; m.current_started = cycle.started; diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index eb784bfb6..e43fc478a 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -2769,9 +2769,9 @@ impl SetDisks { let buckets_results_tx_clone = buckets_results_tx.clone(); futures.push(async move { loop { - match bucket_rx_clone.write().await.recv().await { - None => return, - Some(bucket_info) => { + match bucket_rx_clone.write().await.try_recv() { + Err(_) => return, + Ok(bucket_info) => { let cache_name = Path::new(&bucket_info.name).join(DATA_USAGE_CACHE_NAME); let mut cache = match DataUsageCache::load(self, &cache_name.to_string_lossy()).await { Ok(cache) => cache, @@ -2842,10 +2842,14 @@ impl SetDisks { let _ = cache.save(&cache_name.to_string_lossy()).await; } } + info!("continue scanner"); } }); } + info!("ns_scanner start"); let _ = join_all(futures).await; + drop(buckets_results_tx); + info!("1"); let _ = task.await; info!("ns_scanner completed"); Ok(())