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

refactor(logging): standardize concurrency and trusted proxy events (#3417)

Browse Source 
* refactor(logging): standardize concurrency and proxy events

* chore(logging): extend guardrails for concurrency and proxies

* feat(skill): add rustfs logging governance skill
This commit is contained in:
houseme
2026-06-14 01:00:26 +08:00
committed by GitHub
parent 88d2c6ad5c
commit 9059a9c68d
19 changed files with 1363 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/concurrency/src/deadlock.rs
+18 -2
View File
@@ -102,7 +102,15 @@ impl DeadlockManager {
*running = true;
drop(running);
tracing::info!("Deadlock detection started");
tracing::info!(
event = "deadlock_monitor.lifecycle",
component = "concurrency",
subsystem = "deadlock",
state = "started",
check_interval_ms = self.config.check_interval.as_millis(),
hang_threshold_ms = self.config.hang_threshold.as_millis(),
"deadlock monitor state changed"
);
}
/// Stop the deadlock detection
@@ -110,7 +118,15 @@ impl DeadlockManager {
let mut running = self.running.lock().await;
*running = false;
tracing::info!("Deadlock detection stopped");
tracing::info!(
event = "deadlock_monitor.lifecycle",
component = "concurrency",
subsystem = "deadlock",
state = "stopped",
check_interval_ms = self.config.check_interval.as_millis(),
hang_threshold_ms = self.config.hang_threshold.as_millis(),
"deadlock monitor state changed"
);
}
/// Create a request tracker
crates/concurrency/src/lock.rs
+10 -2
View File
@@ -138,9 +138,13 @@ impl<G> OptimizedLockGuard<G> {
lock_metrics::record_lock_hold_time(hold_time);
tracing::debug!(
event = "lock_guard.release",
component = "concurrency",
subsystem = "lock",
release_mode = "early",
resource = %self.resource,
hold_time_ms = hold_time.as_millis(),
"Lock released early (optimization active)"
"lock guard released"
);
}
@@ -161,9 +165,13 @@ impl<G> Drop for OptimizedLockGuard<G> {
lock_metrics::record_lock_hold_time(hold_time);
tracing::debug!(
event = "lock_guard.release",
component = "concurrency",
subsystem = "lock",
release_mode = "drop",
resource = %self.resource,
hold_time_ms = hold_time.as_millis(),
"Lock released on drop (normal release)"
"lock guard released"
);
}
}
crates/concurrency/src/manager.rs
+17 -7
View File
@@ -233,12 +233,16 @@ impl ConcurrencyManager {
}
tracing::info!(
"Concurrency manager started (timeout={}, lock={}, deadlock={}, backpressure={}, scheduler={})",
self.is_timeout_enabled(),
self.is_lock_enabled(),
self.is_deadlock_enabled(),
self.is_backpressure_enabled(),
self.is_scheduler_enabled()
event = "concurrency_manager.lifecycle",
component = "concurrency",
subsystem = "manager",
state = "started",
timeout_enabled = self.is_timeout_enabled(),
lock_enabled = self.is_lock_enabled(),
deadlock_enabled = self.is_deadlock_enabled(),
backpressure_enabled = self.is_backpressure_enabled(),
scheduler_enabled = self.is_scheduler_enabled(),
"concurrency manager state changed"
);
}
@@ -249,7 +253,13 @@ impl ConcurrencyManager {
self.deadlock.stop().await;
}
tracing::info!("Concurrency manager stopped");
tracing::info!(
event = "concurrency_manager.lifecycle",
component = "concurrency",
subsystem = "manager",
state = "stopped",
"concurrency manager state changed"
);
}
}
crates/concurrency/src/workers.rs
+50 -4
View File
@@ -16,7 +16,7 @@
use std::sync::Arc;
use tokio::sync::{Mutex, Notify};
use tracing::info;
use tracing::{debug, trace};
/// Cooperative worker-slot controller for async tasks.
pub struct Workers {
@@ -43,12 +43,30 @@ impl Workers {
pub async fn take(&self) {
loop {
let mut available = self.available.lock().await;
info!("worker take, {}", *available);
if *available == 0 {
trace!(
event = "worker_slot.acquire",
component = "concurrency",
subsystem = "workers",
state = "waiting",
available_slots = *available,
total_slots = self.limit,
"worker slot pending"
);
drop(available);
self.notify.notified().await;
} else {
*available -= 1;
trace!(
event = "worker_slot.acquire",
component = "concurrency",
subsystem = "workers",
state = "granted",
available_slots = *available,
total_slots = self.limit,
permits_in_use = self.limit.saturating_sub(*available),
"worker slot updated"
);
break;
}
}
@@ -57,8 +75,17 @@ impl Workers {
/// Release a worker slot.
pub async fn give(&self) {
let mut available = self.available.lock().await;
info!("worker give, {}", *available);
*available = (*available).saturating_add(1).min(self.limit); // avoid over-release beyond limit
trace!(
event = "worker_slot.release",
component = "concurrency",
subsystem = "workers",
state = "released",
available_slots = *available,
total_slots = self.limit,
permits_in_use = self.limit.saturating_sub(*available),
"worker slot updated"
);
self.notify.notify_one(); // Notify a waiting task
}
@@ -70,11 +97,30 @@ impl Workers {
if *available == self.limit {
break;
}
trace!(
event = "worker_slot.wait",
component = "concurrency",
subsystem = "workers",
state = "waiting",
available_slots = *available,
total_slots = self.limit,
permits_in_use = self.limit.saturating_sub(*available),
"worker drain pending"
);
}
// Wait until all slots are freed
self.notify.notified().await;
}
info!("worker wait end");
debug!(
event = "worker_slot.wait",
component = "concurrency",
subsystem = "workers",
state = "drained",
available_slots = self.limit,
total_slots = self.limit,
permits_in_use = 0,
"worker drain complete"
);
}
/// Return the current number of available worker slots.