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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

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;
 

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;

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<Mutex<System>> = OnceLock::new();
+static PROCESS_SAMPLER: OnceLock<Mutex<ProcessSampler>> = OnceLock::new();
 
 #[inline]
 fn current_pid() -> Pid {
     Pid::from_u32(std::process::id())
 }
 
-#[inline]
-fn process_system() -> &'static Mutex<System> {
-    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<ProcessSampler> {
+    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);
+    }
 }

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();

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);

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<DriveDetailedStats>, 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,

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<Mutex<ProcessSampler>>) {
+    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;
                 }
             }
         }