gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-10 07:06:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(obs): Improve tracing instrumentation (#2086)

Browse Source 
Co-authored-by: loverustfs <hello@rustfs.com>
This commit is contained in:
evan slack
2026-03-07 07:03:20 -05:00
committed by GitHub
parent 8c4735ff88
commit d52a10c5fb
18 changed files with 110 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/scanner.rs
+75 -68
View File
@@ -32,7 +32,7 @@ use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
use tokio::time::Duration;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
use tracing::{debug, error, info, instrument, warn};
fn data_scanner_start_delay() -> Duration {
let secs = rustfs_utils::get_env_u64(ENV_DATA_SCANNER_START_DELAY_SECS, DEFAULT_DATA_SCANNER_START_DELAY_SECS);
@@ -100,6 +100,7 @@ pub async fn read_background_heal_info(storeapi: Arc<ECStore>) -> BackgroundHeal
}
/// Save background healing information to storage
#[instrument(skip(storeapi))]
pub async fn save_background_heal_info(storeapi: Arc<ECStore>, info: BackgroundHealInfo) {
// Skip for ErasureSD setup
if is_erasure_sd().await {
@@ -128,6 +129,77 @@ fn get_lock_acquire_timeout() -> Duration {
Duration::from_secs(rustfs_utils::get_env_u64("RUSTFS_LOCK_ACQUIRE_TIMEOUT", 5))
}
#[instrument(skip_all)]
async fn run_scan_cycle(ctx: &CancellationToken, storeapi: &Arc<ECStore>, cycle_info: &mut CurrentCycle) {
cycle_info.current = cycle_info.next;
cycle_info.started = Utc::now();
global_metrics().set_cycle(Some(cycle_info.clone())).await;
let background_heal_info = read_background_heal_info(storeapi.clone()).await;
let scan_mode = get_cycle_scan_mode(
cycle_info.current,
background_heal_info.bitrot_start_cycle,
background_heal_info.bitrot_start_time,
);
if background_heal_info.current_scan_mode != scan_mode {
let mut new_heal_info = background_heal_info.clone();
new_heal_info.current_scan_mode = scan_mode;
if scan_mode == HealScanMode::Deep {
new_heal_info.bitrot_start_cycle = cycle_info.current;
new_heal_info.bitrot_start_time = Some(Utc::now());
}
save_background_heal_info(storeapi.clone(), new_heal_info).await;
}
let (sender, receiver) = mpsc::channel::<DataUsageInfo>(1);
let storeapi_clone = storeapi.clone();
let ctx_clone = ctx.clone();
tokio::spawn(async move {
store_data_usage_in_backend(ctx_clone, storeapi_clone, receiver).await;
});
let done_cycle = Metrics::time(Metric::ScanCycle);
let cycle_start = std::time::Instant::now();
if let Err(e) = storeapi
.clone()
.nsscanner(ctx.clone(), sender, cycle_info.current, scan_mode)
.await
{
error!("Failed to scan namespace: {e}");
emit_scan_cycle_complete(false, cycle_start.elapsed());
} else {
done_cycle();
emit_scan_cycle_complete(true, cycle_start.elapsed());
info!("Namespace scanned successfully");
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.split_off(data_usage_update_dir_cycles() as usize);
}
global_metrics().set_cycle(Some(cycle_info.clone())).await;
let cycle_info_buf = cycle_info.marshal().unwrap_or_default();
let mut buf = Vec::with_capacity(cycle_info_buf.len() + 8);
buf.extend_from_slice(&cycle_info.next.to_le_bytes());
buf.extend_from_slice(&cycle_info_buf);
if let Err(e) = save_config(storeapi.clone(), &DATA_USAGE_BLOOM_NAME_PATH, buf).await {
error!("Failed to save data usage bloom name to {}: {}", &*DATA_USAGE_BLOOM_NAME_PATH, e);
} else {
info!("Data usage bloom name saved successfully");
}
}
}
pub async fn run_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) -> Result<(), ScannerError> {
// Acquire leader lock (write lock) to ensure only one scanner runs
let _guard = match storeapi.new_ns_lock(RUSTFS_META_BUCKET, "leader.lock").await {
@@ -167,73 +239,7 @@ pub async fn run_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) ->
break;
}
_ = ticker.tick() => {
cycle_info.current = cycle_info.next;
cycle_info.started = Utc::now();
global_metrics().set_cycle(Some(cycle_info.clone())).await;
let background_heal_info = read_background_heal_info(storeapi.clone()).await;
let scan_mode = get_cycle_scan_mode(cycle_info.current, background_heal_info.bitrot_start_cycle, background_heal_info.bitrot_start_time);
if background_heal_info.current_scan_mode != scan_mode {
let mut new_heal_info = background_heal_info.clone();
new_heal_info.current_scan_mode = scan_mode;
if scan_mode == HealScanMode::Deep {
new_heal_info.bitrot_start_cycle = cycle_info.current;
new_heal_info.bitrot_start_time = Some(Utc::now());
}
save_background_heal_info(storeapi.clone(), new_heal_info).await;
}
let (sender, receiver) = mpsc::channel::<DataUsageInfo>(1);
let storeapi_clone = storeapi.clone();
let ctx_clone = ctx.clone();
tokio::spawn(async move {
store_data_usage_in_backend(ctx_clone, storeapi_clone, receiver).await;
});
let done_cycle = Metrics::time(Metric::ScanCycle);
let cycle_start = std::time::Instant::now();
if let Err(e) = storeapi.clone().nsscanner(ctx.clone(), sender, cycle_info.current, scan_mode).await {
error!("Failed to scan namespace: {e}");
emit_scan_cycle_complete(false, cycle_start.elapsed());
} else {
done_cycle();
emit_scan_cycle_complete(true, cycle_start.elapsed());
info!("Namespace scanned successfully");
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.split_off(data_usage_update_dir_cycles() as usize);
}
global_metrics().set_cycle(Some(cycle_info.clone())).await;
let cycle_info_buf = cycle_info.marshal().unwrap_or_default();
let mut buf = Vec::with_capacity(cycle_info_buf.len() + 8);
buf.extend_from_slice(&cycle_info.next.to_le_bytes());
buf.extend_from_slice(&cycle_info_buf);
if let Err(e) = save_config(storeapi.clone(), &DATA_USAGE_BLOOM_NAME_PATH, buf).await {
error!("Failed to save data usage bloom name to {}: {}", &*DATA_USAGE_BLOOM_NAME_PATH, e);
} else {
info!("Data usage bloom name saved successfully");
}
}
run_scan_cycle(&ctx, &storeapi, &mut cycle_info).await;
ticker.reset();
}
}
@@ -247,6 +253,7 @@ pub async fn run_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) ->
}
/// Store data usage info in backend. Will store all objects sent on the receiver until closed.
#[instrument(skip(ctx, storeapi))]
pub async fn store_data_usage_in_backend(
ctx: CancellationToken,
storeapi: Arc<ECStore>,
crates/scanner/src/scanner_io.rs
+4
View File
@@ -86,6 +86,7 @@ pub trait ScannerIODisk: Send + Sync + Debug + 'static {
#[async_trait::async_trait]
impl ScannerIO for ECStore {
#[tracing::instrument(skip(self, updates))]
async fn nsscanner(
&self,
ctx: CancellationToken,
@@ -222,6 +223,7 @@ impl ScannerIO for ECStore {
#[async_trait::async_trait]
impl ScannerIOCache for SetDisks {
#[tracing::instrument(skip(self, updates))]
async fn nsscanner_cache(
self: Arc<Self>,
ctx: CancellationToken,
@@ -562,6 +564,8 @@ impl ScannerIODisk for Disk {
Ok(size_summary)
}
#[tracing::instrument(skip(self, updates, cache))]
async fn nsscanner_disk(
&self,
ctx: CancellationToken,