From 6cb47049e8373cee6672ff9fd42f2df41e0e600f Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 8 Jul 2026 23:51:34 +0800 Subject: [PATCH] fix(obs): isolate process sampler windows (#4492) * fix(obs): isolate process sampler windows Refs rustfs/backlog#1004 Refs rustfs/backlog#986 - add a reusable ProcessSampler so callers can own independent sysinfo refresh windows - wire separate sampler instances for obs metrics scheduling and memory observability - keep compatibility helpers while avoiding cross-task CPU and disk delta interference Co-Authored-By: heihutu * fix(obs): import process sampler bundle helper Refs rustfs/backlog#1004 Refs rustfs/backlog#986 - import collect_process_metric_bundle_with in the metrics scheduler - drop the stale collect_process_metric_bundle import after switching scheduler sampling to independent process samplers Co-Authored-By: heihutu * fix(obs): move process sampler into blocking task Refs rustfs/backlog#1004 Refs rustfs/backlog#986 - move the memory observability process sampler into the spawn_blocking closure - satisfy the closure static lifetime required by tokio while keeping the isolated sampler design intact Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- crates/io-metrics/src/lib.rs | 5 +- crates/io-metrics/src/sampler/mod.rs | 4 +- crates/io-metrics/src/sampler/process.rs | 63 ++++++++++++++++++----- crates/lock/src/fast_lock/manager.rs | 13 ++++- crates/obs/src/metrics/scheduler.rs | 6 ++- crates/obs/src/metrics/stats_collector.rs | 18 ++++++- rustfs/src/memory_observability.rs | 16 +++--- 7 files changed, 98 insertions(+), 27 deletions(-) diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index a568e2b61..85ed48e9b 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -164,8 +164,9 @@ pub use process_lock_metrics::{ }; pub use s3_api_metrics::{init_s3_metrics, record_s3_op}; pub use sampler::{ - ProcessResourceSnapshot, ProcessStatusSnapshot, ProcessSystemSnapshot, snapshot_process_platform, snapshot_process_resource, - snapshot_process_resource_and_system, snapshot_process_system, + ProcessResourceSnapshot, ProcessSampler, ProcessStatusSnapshot, ProcessSystemSnapshot, snapshot_process_platform, + snapshot_process_resource, snapshot_process_resource_and_system, snapshot_process_resource_and_system_with, + snapshot_process_system, }; pub use system_path_metrics::record_system_path_failure; diff --git a/crates/io-metrics/src/sampler/mod.rs b/crates/io-metrics/src/sampler/mod.rs index f793eb1fd..90ff3e362 100644 --- a/crates/io-metrics/src/sampler/mod.rs +++ b/crates/io-metrics/src/sampler/mod.rs @@ -16,7 +16,7 @@ pub mod process; pub mod system; pub use process::{ - ProcessResourceSnapshot, ProcessStatusSnapshot, ProcessSystemSnapshot, snapshot_process_resource, - snapshot_process_resource_and_system, snapshot_process_system, + ProcessResourceSnapshot, ProcessSampler, ProcessStatusSnapshot, ProcessSystemSnapshot, snapshot_process_resource, + snapshot_process_resource_and_system, snapshot_process_resource_and_system_with, snapshot_process_system, }; pub use system::snapshot_process_platform; diff --git a/crates/io-metrics/src/sampler/process.rs b/crates/io-metrics/src/sampler/process.rs index 2d1b1de6e..d1fde2127 100644 --- a/crates/io-metrics/src/sampler/process.rs +++ b/crates/io-metrics/src/sampler/process.rs @@ -15,21 +15,43 @@ use std::sync::{Mutex, OnceLock}; use sysinfo::{Pid, ProcessRefreshKind, ProcessStatus, ProcessesToUpdate, System}; -static PROCESS_SYSTEM: OnceLock> = OnceLock::new(); +static PROCESS_SAMPLER: OnceLock> = OnceLock::new(); #[inline] fn current_pid() -> Pid { Pid::from_u32(std::process::id()) } -#[inline] -fn process_system() -> &'static Mutex { - PROCESS_SYSTEM.get_or_init(|| { +#[derive(Debug)] +pub struct ProcessSampler { + pid: Pid, + sys: System, +} + +impl Default for ProcessSampler { + fn default() -> Self { + Self::new() + } +} + +impl ProcessSampler { + #[inline] + pub fn new() -> Self { let pid = current_pid(); - let mut system = System::new(); - system.refresh_processes_specifics(ProcessesToUpdate::Some(&[pid]), true, ProcessRefreshKind::everything()); - Mutex::new(system) - }) + let mut sys = System::new(); + sys.refresh_processes_specifics(ProcessesToUpdate::Some(&[pid]), true, ProcessRefreshKind::everything()); + Self { pid, sys } + } + + #[inline] + pub fn snapshot_resource_and_system(&mut self) -> (ProcessResourceSnapshot, ProcessSystemSnapshot) { + snapshot_process_resource_and_system_with(self) + } +} + +#[inline] +fn process_sampler() -> &'static Mutex { + PROCESS_SAMPLER.get_or_init(|| Mutex::new(ProcessSampler::new())) } #[derive(Debug, Clone, Copy, Default, PartialEq)] @@ -99,13 +121,21 @@ pub fn snapshot_process_system() -> ProcessSystemSnapshot { /// Collect both resource and system snapshots in one sysinfo refresh. #[inline] pub fn snapshot_process_resource_and_system() -> (ProcessResourceSnapshot, ProcessSystemSnapshot) { + let mut sampler = process_sampler().lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + snapshot_process_resource_and_system_with(&mut sampler) +} + +#[inline] +pub fn snapshot_process_resource_and_system_with( + sampler: &mut ProcessSampler, +) -> (ProcessResourceSnapshot, ProcessSystemSnapshot) { let platform_stats = crate::snapshot_process_platform_stats(); let lock_snapshot = crate::snapshot_process_lock_counts(); - let pid = current_pid(); - let mut sys = process_system().lock().unwrap_or_else(|poisoned| poisoned.into_inner()); - sys.refresh_processes_specifics(ProcessesToUpdate::Some(&[pid]), true, ProcessRefreshKind::everything()); + sampler + .sys + .refresh_processes_specifics(ProcessesToUpdate::Some(&[sampler.pid]), true, ProcessRefreshKind::everything()); - if let Some(process) = sys.process(pid) { + if let Some(process) = sampler.sys.process(sampler.pid) { let disk_usage = process.disk_usage(); let status = ProcessStatusSnapshot::from(process.status()); let uptime_seconds = process.run_time(); @@ -163,4 +193,13 @@ mod tests { let _ = snapshot_process_system(); let _ = snapshot_process_resource_and_system(); } + + #[test] + fn independent_samplers_are_collectable() { + let mut sampler_a = ProcessSampler::new(); + let mut sampler_b = ProcessSampler::new(); + + let _ = snapshot_process_resource_and_system_with(&mut sampler_a); + let _ = snapshot_process_resource_and_system_with(&mut sampler_b); + } } diff --git a/crates/lock/src/fast_lock/manager.rs b/crates/lock/src/fast_lock/manager.rs index d1fc2e2df..b24f96d6f 100644 --- a/crates/lock/src/fast_lock/manager.rs +++ b/crates/lock/src/fast_lock/manager.rs @@ -372,12 +372,17 @@ impl FastObjectLockManager { /// Start background cleanup task fn start_cleanup_task(&self) { + let Ok(handle) = tokio::runtime::Handle::try_current() else { + tracing::debug!("Skipping fast lock cleanup task startup because no Tokio runtime is active"); + return; + }; + let shards = self.shards.clone(); let metrics = self.metrics.clone(); let cleanup_interval = self.config.cleanup_interval; let _max_idle_time = self.config.max_idle_time; - let handle = tokio::spawn(async move { + let handle = handle.spawn(async move { let mut interval = interval(cleanup_interval); loop { @@ -531,6 +536,12 @@ mod tests { manager.shutdown().await; } + #[test] + fn test_manager_construction_without_runtime_does_not_panic() { + let manager = FastObjectLockManager::new(); + assert_eq!(manager.shards.len(), crate::fast_lock::DEFAULT_SHARD_COUNT); + } + #[tokio::test] async fn test_list_locks_reports_held_locks() { let manager = FastObjectLockManager::new(); diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index a6844eec4..1b88d0dc0 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -93,11 +93,12 @@ use crate::metrics::stats_collector::{ collect_bucket_stats, collect_cluster_and_health_stats, collect_cluster_config_stats, collect_cluster_usage_metric_stats, collect_compression_cluster_stats, collect_disk_and_system_drive_stats, collect_erasure_set_stats, collect_host_network_stats, collect_iam_stats, collect_ilm_metric_stats, collect_internode_network_stats, - collect_process_metric_bundle, collect_replication_stats, collect_scanner_metric_stats, + collect_process_metric_bundle_with, collect_replication_stats, collect_scanner_metric_stats, collect_system_cpu_and_memory_stats_with, }; use futures_util::FutureExt; use rustfs_audit::audit_target_metrics; +use rustfs_io_metrics::ProcessSampler; use rustfs_notify::{notification_metrics_snapshot, notification_target_metrics}; use rustfs_utils::get_env_opt_u64; use serde::Serialize; @@ -1274,6 +1275,7 @@ pub fn init_metrics_runtime(token: CancellationToken) { let labels = current_process_metric_labels(); let mut host_system = System::new_all(); let mut host_networks = Networks::new(); + let mut process_sampler = ProcessSampler::new(); let process_interval = config.process_interval; let mut interval = metrics_interval(process_interval, Duration::ZERO); let now = Instant::now(); @@ -1311,7 +1313,7 @@ pub fn init_metrics_runtime(token: CancellationToken) { _ = interval.tick() => { run_metrics_collector_tick(health, MetricsCollectorTaskId::ProcessMetrics, "process_metrics", async { let now = Instant::now(); - let bundle = collect_process_metric_bundle(); + let bundle = collect_process_metric_bundle_with(&mut process_sampler); if now >= next_resource_run { let mut metrics = collect_resource_metrics(&bundle.resource); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 9782c5fff..47956be1e 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -37,7 +37,10 @@ use chrono::Utc; use rustfs_common::heal_channel::HealScanMode; use rustfs_common::metrics::{ScannerMetricsReport, global_metrics}; use rustfs_io_metrics::internode_metrics::global_internode_metrics; -use rustfs_io_metrics::{ProcessStatusSnapshot, snapshot_process_resource_and_system}; +use rustfs_io_metrics::{ + ProcessResourceSnapshot, ProcessSampler, ProcessStatusSnapshot, ProcessSystemSnapshot, snapshot_process_resource_and_system, + snapshot_process_resource_and_system_with, +}; use std::{collections::HashMap, sync::Arc, time::SystemTime}; use sysinfo::{Networks, System}; use tracing::{instrument, warn}; @@ -662,6 +665,19 @@ pub async fn collect_system_drive_stats() -> (Vec, DriveCoun #[inline] pub fn collect_process_metric_bundle() -> ProcessMetricBundle { let (resource_snapshot, process_snapshot) = snapshot_process_resource_and_system(); + process_metric_bundle_from_snapshots(resource_snapshot, process_snapshot) +} + +#[inline] +pub fn collect_process_metric_bundle_with(sampler: &mut ProcessSampler) -> ProcessMetricBundle { + let (resource_snapshot, process_snapshot) = snapshot_process_resource_and_system_with(sampler); + process_metric_bundle_from_snapshots(resource_snapshot, process_snapshot) +} + +fn process_metric_bundle_from_snapshots( + resource_snapshot: ProcessResourceSnapshot, + process_snapshot: ProcessSystemSnapshot, +) -> ProcessMetricBundle { let status = match process_snapshot.status { ProcessStatusSnapshot::Running => ProcessStatusType::Running, ProcessStatusSnapshot::Sleeping => ProcessStatusType::Sleeping, diff --git a/rustfs/src/memory_observability.rs b/rustfs/src/memory_observability.rs index 38193e862..f0a7b5093 100644 --- a/rustfs/src/memory_observability.rs +++ b/rustfs/src/memory_observability.rs @@ -13,8 +13,8 @@ // limitations under the License. use rustfs_io_metrics::{ - AllocatorMemoryObservation, record_allocator_memory_observation, record_cgroup_memory_split, record_cpu_usage, - record_memory_usage, record_process_memory_split, snapshot_process_resource_and_system, + AllocatorMemoryObservation, ProcessSampler, record_allocator_memory_observation, record_cgroup_memory_split, + record_cpu_usage, record_memory_usage, record_process_memory_split, }; use serde::Serialize; use serde_json::Value; @@ -22,7 +22,7 @@ use std::collections::HashMap; #[cfg(not(target_os = "windows"))] use std::ffi::CStr; use std::path::Path; -use std::sync::{Mutex, OnceLock}; +use std::sync::{Arc, Mutex, OnceLock}; use std::time::Duration; use sysinfo::System; use tokio_util::sync::CancellationToken; @@ -368,9 +368,10 @@ pub fn memory_observability_controller_snapshot(ctx: &CancellationToken) -> Memo ) } -async fn record_memory_snapshot() { - match tokio::task::spawn_blocking(|| { - let (resource, process) = snapshot_process_resource_and_system(); +async fn record_memory_snapshot(process_sampler: Arc>) { + match tokio::task::spawn_blocking(move || { + let mut sampler = process_sampler.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let (resource, process) = sampler.snapshot_resource_and_system(); let total_memory = refresh_total_memory(); let cgroup = read_cgroup_memory_snapshot(); let allocator = read_allocator_memory_snapshot(); @@ -407,6 +408,7 @@ async fn record_memory_snapshot() { pub fn init_memory_observability(ctx: CancellationToken) { let interval_secs = configured_memory_observability_interval_secs(); let interval = Duration::from_secs(interval_secs.max(1)); + let process_sampler = Arc::new(Mutex::new(ProcessSampler::new())); tokio::spawn(async move { let mut ticker = tokio::time::interval(interval); @@ -419,7 +421,7 @@ pub fn init_memory_observability(ctx: CancellationToken) { break; } _ = ticker.tick() => { - record_memory_snapshot().await; + record_memory_snapshot(Arc::clone(&process_sampler)).await; } } }