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

Merge pull request #145 from rustfs/scanner

Browse Source 
Scanner
This commit is contained in:
junxiangMu
2024-12-03 17:07:43 +08:00
committed by GitHub
parent 4a173bbd98 14627a5c2b
commit c6c48cb803
9 changed files with 125 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/lock/src/lrwmutex.rs
+3 -2
View File
@@ -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")
}
}
ecstore/src/disk/local.rs
+1
View File
@@ -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)
}
ecstore/src/disk/mod.rs
+1
View File
@@ -351,6 +351,7 @@ impl DiskAPI for Disk {
updates: Sender<DataUsageEntry>,
scan_mode: HealScanMode,
) -> Result<DataUsageCache> {
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,
ecstore/src/heal/data_scanner.rs
+53 -35
View File
@@ -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;
@@ -86,55 +85,42 @@ 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;
};
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::<LittleEndian>() {
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::<LittleEndian>() {
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);
@@ -163,22 +149,23 @@ 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());
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::<LittleEndian>(cycle_info.next).unwrap();
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);
res.insert("error".to_string(), err.to_string());
}
}
@@ -249,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,
@@ -1069,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:?}");
}
}
ecstore/src/heal/data_scanner_metric.rs
+3
View File
@@ -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<CurrentScannerCycle>) {
info!("ScannerMetrics set_cycle {c:?}");
*self.cycle_info.write().await = c;
}
@@ -216,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;
ecstore/src/heal/data_usage_cache.rs
+5 -5
View File
@@ -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<u64>);
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<u64>);
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<String, DataUsageEntry>,
ecstore/src/metrics_realtime.rs
+49 -1
View File
@@ -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<String>) -> HashMap<String, DiskMetric> {
@@ -167,3 +193,25 @@ async fn collect_local_disks_metrics(disks: &HashSet<String>) -> HashMap<String,
metrics
}
#[cfg(test)]
mod test {
use super::MetricType;
#[test]
fn tes_types() {
let t = MetricType::ALL;
assert!(t.contains(&MetricType::NONE));
assert!(t.contains(&MetricType::DISK));
assert!(t.contains(&MetricType::OS));
assert!(t.contains(&MetricType::BATCH_JOBS));
assert!(t.contains(&MetricType::SITE_RESYNC));
assert!(t.contains(&MetricType::NET));
assert!(t.contains(&MetricType::MEM));
assert!(t.contains(&MetricType::CPU));
assert!(t.contains(&MetricType::RPC));
let disk = MetricType::new(1 << 1);
assert!(disk.contains(&MetricType::DISK));
}
}
ecstore/src/set_disk.rs
+9 -3
View File
@@ -2668,6 +2668,7 @@ impl SetDisks {
updates: Sender<DataUsageCache>,
heal_scan_mode: HealScanMode,
) -> Result<()> {
info!("ns_scanner");
if buckets.is_empty() {
return Ok(());
}
@@ -2768,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,
@@ -2841,11 +2842,16 @@ 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(())
}
rustfs/src/admin/handlers.rs
+1 -1
View File
@@ -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) => {