diff --git a/.docker/observability/grafana/dashboards/rustfs.json b/.docker/observability/grafana/dashboards/rustfs.json index 86f595180..2830693ee 100644 --- a/.docker/observability/grafana/dashboards/rustfs.json +++ b/.docker/observability/grafana/dashboards/rustfs.json @@ -4551,7 +4551,7 @@ "index": 0, "text": "INACTIVE" }, - "to": 1e-09 + "to": 1e-9 }, "type": "range" } @@ -5281,7 +5281,10 @@ "id": 102, "options": { "legend": { - "calcs": ["mean", "max"], + "calcs": [ + "mean", + "max" + ], "displayMode": "table", "placement": "bottom" }, @@ -5368,7 +5371,10 @@ "id": 103, "options": { "legend": { - "calcs": ["mean", "max"], + "calcs": [ + "mean", + "max" + ], "displayMode": "table", "placement": "bottom" }, @@ -5455,7 +5461,10 @@ "id": 104, "options": { "legend": { - "calcs": ["mean", "max"], + "calcs": [ + "mean", + "max" + ], "displayMode": "table", "placement": "bottom" }, @@ -5533,7 +5542,9 @@ "id": 105, "options": { "legend": { - "calcs": ["mean"], + "calcs": [ + "mean" + ], "displayMode": "table", "placement": "bottom" }, @@ -5629,7 +5640,10 @@ "id": 106, "options": { "legend": { - "calcs": ["mean", "max"], + "calcs": [ + "mean", + "max" + ], "displayMode": "table", "placement": "bottom" }, @@ -5725,7 +5739,10 @@ "id": 107, "options": { "legend": { - "calcs": ["mean", "max"], + "calcs": [ + "mean", + "max" + ], "displayMode": "table", "placement": "bottom" }, @@ -5803,7 +5820,10 @@ "id": 108, "options": { "legend": { - "calcs": ["mean", "max"], + "calcs": [ + "mean", + "max" + ], "displayMode": "table", "placement": "bottom" }, @@ -8071,6 +8091,374 @@ "x": 0, "y": 195 }, + "id": 150, + "panels": [], + "title": "Object Lock Diagnostics", + "type": "row" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + }, + { + "color": "orange", + "value": 1 + }, + { + "color": "red", + "value": 5 + } + ] + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 196 + }, + "id": 151, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (op, mode) (increase(rustfs_object_lock_diag_slow_acquire_total{job=~\"$job\"}[$__rate_interval]))", + "legendFormat": "{{op}} | {{mode}}", + "range": true, + "refId": "A" + } + ], + "title": "Slow Object Lock Acquires", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + }, + { + "color": "orange", + "value": 1 + }, + { + "color": "red", + "value": 5 + } + ] + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 196 + }, + "id": 152, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (op, mode) (increase(rustfs_object_lock_diag_slow_hold_total{job=~\"$job\"}[$__rate_interval]))", + "legendFormat": "{{op}} | {{mode}}", + "range": true, + "refId": "A" + } + ], + "title": "Slow Object Lock Holds", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "ms" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 204 + }, + "id": 153, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "1000 * histogram_quantile(0.95, sum by (le, op, mode) (rate(rustfs_object_lock_diag_acquire_duration_seconds_bucket{job=~\"$job\"}[5m])))", + "legendFormat": "{{op}} | {{mode}}", + "range": true, + "refId": "A" + } + ], + "title": "Object Lock Acquire Duration P95", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "ms" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 204 + }, + "id": 154, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "1000 * histogram_quantile(0.95, sum by (le, op, mode) (rate(rustfs_object_lock_diag_hold_duration_seconds_bucket{job=~\"$job\"}[5m])))", + "legendFormat": "{{op}} | {{mode}}", + "range": true, + "refId": "A" + } + ], + "title": "Object Lock Hold Duration P95", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [ + { + "options": { + "0": { + "text": "Disabled" + }, + "1": { + "text": "Enabled" + } + }, + "type": "value" + } + ], + "max": 1, + "min": 0, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "red", + "value": 0 + }, + { + "color": "green", + "value": 1 + } + ] + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 24, + "x": 0, + "y": 212 + }, + "id": 155, + "options": { + "colorMode": "background", + "graphMode": "none", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "max(rustfs_object_lock_diag_enabled{job=~\"$job\"})", + "legendFormat": "__auto", + "range": true, + "refId": "A" + } + ], + "title": "Object Lock Diagnostics Enabled", + "type": "stat" + }, + { + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 213 + }, "id": 137, "panels": [], "title": "Debug / Raw Explorer", @@ -8138,7 +8526,7 @@ "h": 8, "w": 12, "x": 0, - "y": 196 + "y": 214 }, "id": 138, "options": { @@ -8235,7 +8623,7 @@ "h": 8, "w": 12, "x": 12, - "y": 196 + "y": 214 }, "id": 139, "options": { @@ -8332,7 +8720,7 @@ "h": 8, "w": 12, "x": 0, - "y": 204 + "y": 222 }, "id": 140, "options": { @@ -8429,7 +8817,7 @@ "h": 8, "w": 12, "x": 12, - "y": 204 + "y": 222 }, "id": 141, "options": { @@ -8526,7 +8914,7 @@ "h": 8, "w": 12, "x": 0, - "y": 212 + "y": 230 }, "id": 142, "options": { @@ -8623,7 +9011,7 @@ "h": 8, "w": 12, "x": 12, - "y": 212 + "y": 230 }, "id": 143, "options": { @@ -8720,7 +9108,7 @@ "h": 8, "w": 12, "x": 0, - "y": 220 + "y": 238 }, "id": 144, "options": { @@ -8817,7 +9205,7 @@ "h": 8, "w": 12, "x": 12, - "y": 220 + "y": 238 }, "id": 145, "options": { @@ -8914,7 +9302,7 @@ "h": 8, "w": 12, "x": 0, - "y": 228 + "y": 246 }, "id": 146, "options": { @@ -9011,7 +9399,7 @@ "h": 8, "w": 12, "x": 12, - "y": 228 + "y": 246 }, "id": 147, "options": { @@ -9108,7 +9496,7 @@ "h": 8, "w": 12, "x": 0, - "y": 236 + "y": 254 }, "id": 148, "options": { @@ -9205,7 +9593,7 @@ "h": 8, "w": 12, "x": 12, - "y": 236 + "y": 254 }, "id": 149, "options": { @@ -9382,5 +9770,5 @@ "timezone": "browser", "title": "RustFS", "uid": "rustfs-s3", - "version": 13 + "version": 14 } diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index 0a2aa05b0..d77189431 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -281,6 +281,40 @@ pub const ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT: &str = "RUSTFS_OBJECT_LOCK_ACQUIRE_TI /// Default lock acquisition timeout: 5 seconds. pub const DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT: u64 = 5; +/// Environment variable to enable object namespace lock diagnostics. +/// +/// When enabled, RustFS emits slow lock acquisition and long lock hold +/// warnings for object-level namespace locks. This is intended for +/// production debugging of contention on hot object keys. +/// +/// Default: false (disabled, can be overridden by `RUSTFS_OBJECT_LOCK_DIAG_ENABLE`). +pub const ENV_OBJECT_LOCK_DIAG_ENABLE: &str = "RUSTFS_OBJECT_LOCK_DIAG_ENABLE"; + +/// Default: object lock diagnostics are disabled. +pub const DEFAULT_OBJECT_LOCK_DIAG_ENABLE: bool = false; + +/// Environment variable for the slow object lock acquisition threshold in milliseconds. +/// +/// When a read or write namespace lock takes at least this long to acquire, +/// RustFS emits a warning with the operation name and object key. +/// +/// Default: 500 milliseconds. +pub const ENV_OBJECT_LOCK_DIAG_SLOW_ACQUIRE_MS: &str = "RUSTFS_OBJECT_LOCK_DIAG_SLOW_ACQUIRE_MS"; + +/// Default slow object lock acquisition threshold: 500 milliseconds. +pub const DEFAULT_OBJECT_LOCK_DIAG_SLOW_ACQUIRE_MS: u64 = 500; + +/// Environment variable for the long object lock hold threshold in milliseconds. +/// +/// When a namespace lock guard is held for at least this long, RustFS emits +/// a warning when the guard is dropped. +/// +/// Default: 1000 milliseconds. +pub const ENV_OBJECT_LOCK_DIAG_SLOW_HOLD_MS: &str = "RUSTFS_OBJECT_LOCK_DIAG_SLOW_HOLD_MS"; + +/// Default long object lock hold threshold: 1000 milliseconds. +pub const DEFAULT_OBJECT_LOCK_DIAG_SLOW_HOLD_MS: u64 = 1000; + // ============================================================================ // I/O priority scheduling configuration // ============================================================================ diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 5ef636323..c47a2ec0b 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -73,6 +73,10 @@ use rustfs_filemeta::{ FileInfo, FileMeta, FileMetaShallowVersion, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo, RawFileInfo, ReplicateDecision, ReplicationStatusType, VersionPurgeStatusType, file_info_from_raw, merge_file_meta_versions, }; +use rustfs_io_metrics::{ + record_object_lock_diag_acquire_duration, record_object_lock_diag_enabled, record_object_lock_diag_hold_duration, + record_object_lock_diag_slow_acquire, record_object_lock_diag_slow_hold, +}; use rustfs_lock::LockClient; use rustfs_lock::fast_lock::types::LockResult; use rustfs_lock::local_lock::LocalLock; @@ -102,6 +106,7 @@ use s3s::header::{X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OB use sha2::{Digest, Sha256}; use std::hash::Hash; use std::mem::{self}; +use std::sync::OnceLock; use std::time::{Instant, SystemTime, UNIX_EPOCH}; use std::{ collections::{HashMap, HashSet}, @@ -131,6 +136,66 @@ pub(crate) const RUSTFS_MULTIPART_BUCKET_KEY: &str = "x-rustfs-internal-multipar pub(crate) const RUSTFS_MULTIPART_OBJECT_KEY: &str = "x-rustfs-internal-multipart-object"; const ENV_ISSUE3031_DIAG_ENABLE: &str = "RUSTFS_ISSUE3031_DIAG_ENABLE"; +struct ObjectLockDiagGuard { + guard: NamespaceLockGuard, + enabled: bool, + op: &'static str, + bucket: Option, + object: Option, + owner: Option, + mode: &'static str, + acquired_at: Instant, +} + +impl ObjectLockDiagGuard { + fn new( + guard: NamespaceLockGuard, + enabled: bool, + op: &'static str, + bucket: Option, + object: Option, + owner: Option, + mode: &'static str, + ) -> Self { + Self { + guard, + enabled, + op, + bucket, + object, + owner, + mode, + acquired_at: Instant::now(), + } + } +} + +impl Drop for ObjectLockDiagGuard { + fn drop(&mut self) { + if !self.enabled || self.guard.is_released() { + return; + } + + let hold = self.acquired_at.elapsed(); + record_object_lock_diag_hold_duration(self.op, self.mode, hold); + let threshold = get_object_lock_diag_slow_hold_threshold(); + if hold >= threshold { + record_object_lock_diag_slow_hold(self.op, self.mode); + warn!( + target: "rustfs_ecstore::object_lock_diag", + op = self.op, + bucket = %self.bucket.as_deref().unwrap_or_default(), + object = %self.object.as_deref().unwrap_or_default(), + mode = self.mode, + owner = %self.owner.as_deref().unwrap_or_default(), + hold_ms = hold.as_millis(), + threshold_ms = threshold.as_millis(), + "object namespace lock held longer than threshold" + ); + } + } +} + pub(crate) fn strip_internal_multipart_metadata(metadata: &mut HashMap) { metadata.remove(RUSTFS_MULTIPART_BUCKET_KEY); metadata.remove(RUSTFS_MULTIPART_OBJECT_KEY); @@ -189,6 +254,7 @@ pub fn get_duplex_buffer_size() -> usize { } const DISK_ONLINE_TIMEOUT: Duration = Duration::from_secs(1); const DISK_HEALTH_CACHE_TTL: Duration = Duration::from_millis(750); +static OBJECT_LOCK_DIAG_ENABLED: OnceLock = OnceLock::new(); mod heal; mod list; @@ -208,6 +274,31 @@ pub fn get_lock_acquire_timeout() -> Duration { )) } +pub fn is_object_lock_diag_enabled() -> bool { + *OBJECT_LOCK_DIAG_ENABLED.get_or_init(|| { + let enabled = rustfs_utils::get_env_bool( + rustfs_config::ENV_OBJECT_LOCK_DIAG_ENABLE, + rustfs_config::DEFAULT_OBJECT_LOCK_DIAG_ENABLE, + ); + record_object_lock_diag_enabled(enabled); + enabled + }) +} + +pub fn get_object_lock_diag_slow_acquire_threshold() -> Duration { + Duration::from_millis(rustfs_utils::get_env_u64( + rustfs_config::ENV_OBJECT_LOCK_DIAG_SLOW_ACQUIRE_MS, + rustfs_config::DEFAULT_OBJECT_LOCK_DIAG_SLOW_ACQUIRE_MS, + )) +} + +pub fn get_object_lock_diag_slow_hold_threshold() -> Duration { + Duration::from_millis(rustfs_utils::get_env_u64( + rustfs_config::ENV_OBJECT_LOCK_DIAG_SLOW_HOLD_MS, + rustfs_config::DEFAULT_OBJECT_LOCK_DIAG_SLOW_HOLD_MS, + )) +} + /// Check if lock optimization is enabled. /// When enabled, read locks are released after metadata read instead of /// being held for the entire data transfer duration. @@ -342,6 +433,89 @@ impl DiskHealthEntry { } impl SetDisks { + async fn acquire_read_lock_diag(&self, op: &'static str, bucket: &str, object: &str) -> Result { + let diag_enabled = is_object_lock_diag_enabled(); + let ns_lock = self.new_ns_lock(bucket, object).await?; + let acquire_start = Instant::now(); + let guard = ns_lock + .get_read_lock(get_lock_acquire_timeout()) + .await + .map_err(|e| self.map_namespace_lock_error(bucket, object, "read", e))?; + let owner = diag_enabled.then(|| ns_lock.owner().to_string()); + self.log_object_lock_acquire_if_slow(op, bucket, object, "read", owner.as_deref(), acquire_start.elapsed(), diag_enabled); + Ok(ObjectLockDiagGuard::new( + guard, + diag_enabled, + op, + diag_enabled.then(|| bucket.to_string()), + diag_enabled.then(|| object.to_string()), + owner, + "read", + )) + } + + async fn acquire_write_lock_diag(&self, op: &'static str, bucket: &str, object: &str) -> Result { + let diag_enabled = is_object_lock_diag_enabled(); + let ns_lock = self.new_ns_lock(bucket, object).await?; + let acquire_start = Instant::now(); + let guard = ns_lock + .get_write_lock(get_lock_acquire_timeout()) + .await + .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?; + let owner = diag_enabled.then(|| ns_lock.owner().to_string()); + self.log_object_lock_acquire_if_slow( + op, + bucket, + object, + "write", + owner.as_deref(), + acquire_start.elapsed(), + diag_enabled, + ); + Ok(ObjectLockDiagGuard::new( + guard, + diag_enabled, + op, + diag_enabled.then(|| bucket.to_string()), + diag_enabled.then(|| object.to_string()), + owner, + "write", + )) + } + + #[allow(clippy::too_many_arguments)] + fn log_object_lock_acquire_if_slow( + &self, + op: &'static str, + bucket: &str, + object: &str, + mode: &'static str, + owner: Option<&str>, + elapsed: Duration, + diag_enabled: bool, + ) { + if !diag_enabled { + return; + } + + let threshold = get_object_lock_diag_slow_acquire_threshold(); + record_object_lock_diag_acquire_duration(op, mode, elapsed); + if elapsed >= threshold { + record_object_lock_diag_slow_acquire(op, mode); + warn!( + target: "rustfs_ecstore::object_lock_diag", + op, + bucket, + object, + mode, + owner = owner.unwrap_or_default(), + acquire_ms = elapsed.as_millis(), + threshold_ms = threshold.as_millis(), + "object namespace lock acquisition exceeded threshold" + ); + } + } + #[allow(clippy::too_many_arguments)] pub async fn new( locker_owner: String, @@ -628,12 +802,7 @@ impl ObjectIO for SetDisks { ); } - let guard = self - .new_ns_lock(bucket, object) - .await? - .get_read_lock(get_lock_acquire_timeout()) - .await - .map_err(|e| self.map_namespace_lock_error(bucket, object, "read", e))?; + let guard = self.acquire_read_lock_diag("get_object", bucket, object).await?; // Record lock acquisition for deadlock detection let _lock_id = record_lock_acquire(bucket, object, "read"); @@ -761,12 +930,9 @@ impl ObjectIO for SetDisks { if opts.http_preconditions.is_some() { if !opts.no_lock { - let ns_lock = self.new_ns_lock(bucket, object).await?; object_lock_guard = Some( - ns_lock - .get_write_lock(get_lock_acquire_timeout()) - .await - .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, + self.acquire_write_lock_diag("put_object_precondition", bucket, object) + .await?, ); } @@ -1002,13 +1168,7 @@ impl ObjectIO for SetDisks { drop(writers); // drop writers to close all files, this is to prevent FileAccessDenied errors when renaming data if !opts.no_lock && object_lock_guard.is_none() { - let ns_lock = self.new_ns_lock(bucket, object).await?; - object_lock_guard = Some( - ns_lock - .get_write_lock(get_lock_acquire_timeout()) - .await - .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, - ); + object_lock_guard = Some(self.acquire_write_lock_diag("put_object_commit", bucket, object).await?); } let (online_disks, _, op_old_dir) = Self::rename_data( @@ -1428,11 +1588,8 @@ impl ObjectOperations for SetDisks { None } else { Some( - self.new_ns_lock(dst_bucket, dst_object) - .await? - .get_write_lock(get_lock_acquire_timeout()) - .await - .map_err(|e| self.map_namespace_lock_error(dst_bucket, dst_object, "write", e))?, + self.acquire_write_lock_diag("copy_object_metadata", dst_bucket, dst_object) + .await?, ) }; @@ -1822,13 +1979,7 @@ impl ObjectOperations for SetDisks { async fn delete_object(&self, bucket: &str, object: &str, mut opts: ObjectOptions) -> Result { // Guard lock for single object delete let _lock_guard = if (!opts.delete_prefix || opts.delete_prefix_object) && !opts.no_lock { - Some( - self.new_ns_lock(bucket, object) - .await? - .get_write_lock(get_lock_acquire_timeout()) - .await - .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, - ) + Some(self.acquire_write_lock_diag("delete_object", bucket, object).await?) } else { None }; @@ -1963,13 +2114,7 @@ impl ObjectOperations for SetDisks { async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { // Acquire a shared read-lock to protect consistency during info fetch let _read_lock_guard = if !opts.no_lock { - Some( - self.new_ns_lock(bucket, object) - .await? - .get_read_lock(get_lock_acquire_timeout()) - .await - .map_err(|e| self.map_namespace_lock_error(bucket, object, "read", e))?, - ) + Some(self.acquire_read_lock_diag("get_object_info", bucket, object).await?) } else { None }; @@ -2017,13 +2162,7 @@ impl ObjectOperations for SetDisks { // Guard lock for metadata update let _lock_guard = if !opts.no_lock { - Some( - self.new_ns_lock(bucket, object) - .await? - .get_write_lock(get_lock_acquire_timeout()) - .await - .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, - ) + Some(self.acquire_write_lock_diag("put_object_metadata", bucket, object).await?) } else { None }; @@ -3090,12 +3229,9 @@ impl MultipartOperations for SetDisks { if opts.http_preconditions.is_some() { if !opts.no_lock { - let ns_lock = self.new_ns_lock(bucket, object).await?; _object_lock_guard = Some( - ns_lock - .get_write_lock(get_lock_acquire_timeout()) - .await - .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, + self.acquire_write_lock_diag("new_multipart_upload_precondition", bucket, object) + .await?, ); } @@ -3269,12 +3405,9 @@ impl MultipartOperations for SetDisks { if opts.http_preconditions.is_some() { if !opts.no_lock { - let ns_lock = self.new_ns_lock(bucket, object).await?; object_lock_guard = Some( - ns_lock - .get_write_lock(get_lock_acquire_timeout()) - .await - .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, + self.acquire_write_lock_diag("complete_multipart_upload_precondition", bucket, object) + .await?, ); } @@ -3619,12 +3752,9 @@ impl MultipartOperations for SetDisks { } if !opts.no_lock && object_lock_guard.is_none() { - let ns_lock = self.new_ns_lock(bucket, object).await?; object_lock_guard = Some( - ns_lock - .get_write_lock(get_lock_acquire_timeout()) - .await - .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, + self.acquire_write_lock_diag("complete_multipart_upload_commit", bucket, object) + .await?, ); } diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 0187120f4..0a5e5dff2 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -13,16 +13,25 @@ // limitations under the License. use super::*; -use crate::set_disk::{get_lock_acquire_timeout, is_lock_optimization_enabled}; +use crate::set_disk::{ + get_lock_acquire_timeout, get_object_lock_diag_slow_acquire_threshold, get_object_lock_diag_slow_hold_threshold, + is_lock_optimization_enabled, is_object_lock_diag_enabled, +}; +use rustfs_io_metrics::{ + record_object_lock_diag_acquire_duration, record_object_lock_diag_hold_duration, record_object_lock_diag_slow_acquire, + record_object_lock_diag_slow_hold, +}; use std::{ + fmt, pin::Pin, task::{Context, Poll}, + time::{Duration, Instant}, }; use tokio::io::{AsyncRead, ReadBuf}; struct LockGuardedReader { inner: Box, - guard: Option, + guard: Option, } impl AsyncRead for LockGuardedReader { @@ -37,6 +46,118 @@ impl AsyncRead for LockGuardedReader { } } +#[derive(Clone, Copy, Debug)] +enum ObjectLockDiagMode { + Read, + Write, +} + +impl ObjectLockDiagMode { + fn as_str(self) -> &'static str { + match self { + Self::Read => "read", + Self::Write => "write", + } + } +} + +impl fmt::Display for ObjectLockDiagMode { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +struct ObjectLockDiagGuard { + guard: rustfs_lock::NamespaceLockGuard, + enabled: bool, + op: &'static str, + bucket: Option, + object: Option, + owner: Option, + mode: ObjectLockDiagMode, + acquired_at: Instant, +} + +impl ObjectLockDiagGuard { + fn new( + guard: rustfs_lock::NamespaceLockGuard, + enabled: bool, + op: &'static str, + bucket: Option, + object: Option, + owner: Option, + mode: ObjectLockDiagMode, + ) -> Self { + Self { + guard, + enabled, + op, + bucket, + object, + owner, + mode, + acquired_at: Instant::now(), + } + } +} + +impl Drop for ObjectLockDiagGuard { + fn drop(&mut self) { + if !self.enabled || self.guard.is_released() { + return; + } + + let hold = self.acquired_at.elapsed(); + record_object_lock_diag_hold_duration(self.op, self.mode.as_str(), hold); + let threshold = get_object_lock_diag_slow_hold_threshold(); + if hold >= threshold { + record_object_lock_diag_slow_hold(self.op, self.mode.as_str()); + warn!( + target: "rustfs_ecstore::object_lock_diag", + op = self.op, + bucket = %self.bucket.as_deref().unwrap_or_default(), + object = %self.object.as_deref().unwrap_or_default(), + mode = %self.mode, + owner = %self.owner.as_deref().unwrap_or_default(), + hold_ms = hold.as_millis(), + threshold_ms = threshold.as_millis(), + "object namespace lock held longer than threshold" + ); + } + } +} + +fn log_object_lock_acquire_if_slow( + op: &'static str, + bucket: &str, + object: &str, + owner: Option<&str>, + mode: ObjectLockDiagMode, + elapsed: Duration, + diag_enabled: bool, +) { + if !diag_enabled { + return; + } + + let threshold = get_object_lock_diag_slow_acquire_threshold(); + record_object_lock_diag_acquire_duration(op, mode.as_str(), elapsed); + if elapsed >= threshold { + record_object_lock_diag_slow_acquire(op, mode.as_str()); + warn!( + target: "rustfs_ecstore::object_lock_diag", + op, + bucket, + object, + mode = %mode, + owner = owner.unwrap_or_default(), + acquire_ms = elapsed.as_millis(), + threshold_ms = threshold.as_millis(), + "object namespace lock acquisition exceeded threshold" + ); + } +} + fn select_data_movement_target_pool( existing_pool_idx: Result, src_pool_idx: usize, @@ -128,45 +249,87 @@ impl ECStore { async fn acquire_object_write_lock_if_needed( &self, + op: &'static str, bucket: &str, object: &str, opts: &mut ObjectOptions, - ) -> Result> { + ) -> Result> { if opts.no_lock { return Ok(None); } + let diag_enabled = is_object_lock_diag_enabled(); let ns_lock = self.handle_new_ns_lock(bucket, object).await?; + let acquire_start = Instant::now(); let guard = ns_lock .get_write_lock(get_lock_acquire_timeout()) .await .map_err(|err| Self::map_namespace_lock_error(bucket, object, "write", err))?; + let owner = diag_enabled.then(|| ns_lock.owner().to_string()); + log_object_lock_acquire_if_slow( + op, + bucket, + object, + owner.as_deref(), + ObjectLockDiagMode::Write, + acquire_start.elapsed(), + diag_enabled, + ); opts.no_lock = true; - Ok(Some(guard)) + Ok(Some(ObjectLockDiagGuard::new( + guard, + diag_enabled, + op, + diag_enabled.then(|| bucket.to_string()), + diag_enabled.then(|| object.to_string()), + owner, + ObjectLockDiagMode::Write, + ))) } async fn acquire_object_read_lock_if_needed( &self, + op: &'static str, bucket: &str, object: &str, opts: &mut ObjectOptions, - ) -> Result> { + ) -> Result> { if opts.no_lock { return Ok(None); } + let diag_enabled = is_object_lock_diag_enabled(); let ns_lock = self.handle_new_ns_lock(bucket, object).await?; + let acquire_start = Instant::now(); let guard = ns_lock .get_read_lock(get_lock_acquire_timeout()) .await .map_err(|err| Self::map_namespace_lock_error(bucket, object, "read", err))?; + let owner = diag_enabled.then(|| ns_lock.owner().to_string()); + log_object_lock_acquire_if_slow( + op, + bucket, + object, + owner.as_deref(), + ObjectLockDiagMode::Read, + acquire_start.elapsed(), + diag_enabled, + ); opts.no_lock = true; - Ok(Some(guard)) + Ok(Some(ObjectLockDiagGuard::new( + guard, + diag_enabled, + op, + diag_enabled.then(|| bucket.to_string()), + diag_enabled.then(|| object.to_string()), + owner, + ObjectLockDiagMode::Read, + ))) } - fn attach_read_lock_guard(mut reader: GetObjectReader, guard: Option) -> GetObjectReader { + fn attach_read_lock_guard(mut reader: GetObjectReader, guard: Option) -> GetObjectReader { if is_lock_optimization_enabled() { return reader; } @@ -286,7 +449,9 @@ impl ECStore { let object = encode_dir_object(object); let mut opts = opts.clone(); - let read_lock_guard = self.acquire_object_read_lock_if_needed(bucket, &object, &mut opts).await?; + let read_lock_guard = self + .acquire_object_read_lock_if_needed("get_object", bucket, &object, &mut opts) + .await?; let reader = if self.single_pool() { self.pools[0] @@ -346,7 +511,9 @@ impl ECStore { let object = encode_dir_object(object); let mut opts = opts.clone(); - let _object_lock_guard = self.acquire_object_read_lock_if_needed(bucket, &object, &mut opts).await?; + let _object_lock_guard = self + .acquire_object_read_lock_if_needed("get_object_info", bucket, &object, &mut opts) + .await?; let info = if self.single_pool() { self.pools[0].get_object_info(bucket, object.as_str(), &opts).await? @@ -381,7 +548,7 @@ impl ECStore { let mut dst_opts = dst_opts.clone(); let _dst_lock_guard = if cp_src_dst_same { - self.acquire_object_write_lock_if_needed(dst_bucket, &dst_object, &mut dst_opts) + self.acquire_object_write_lock_if_needed("copy_object", dst_bucket, &dst_object, &mut dst_opts) .await? } else { None @@ -464,7 +631,9 @@ impl ECStore { return Ok(ObjectInfo::default()); } - let _object_lock_guard = self.acquire_object_write_lock_if_needed(bucket, object, &mut opts).await?; + let _object_lock_guard = self + .acquire_object_write_lock_if_needed("delete_object", bucket, object, &mut opts) + .await?; if opts.delete_prefix { self.delete_prefix(bucket, object, &opts).await?; @@ -1116,6 +1285,15 @@ mod tests { .get_read_lock(key.clone(), "reader", Duration::from_secs(1)) .await .expect("read lock should be acquired"); + let read_guard = ObjectLockDiagGuard::new( + read_guard, + true, + "test_get_object", + Some("bucket".to_string()), + Some("object".to_string()), + Some("reader".to_string()), + ObjectLockDiagMode::Read, + ); let reader = GetObjectReader { stream: Box::new(Cursor::new(Vec::::new())), object_info: ObjectInfo::default(), @@ -1145,6 +1323,15 @@ mod tests { .get_read_lock(key.clone(), "reader", Duration::from_secs(1)) .await .expect("read lock should be acquired"); + let read_guard = ObjectLockDiagGuard::new( + read_guard, + true, + "test_get_object", + Some("bucket".to_string()), + Some("object".to_string()), + Some("reader".to_string()), + ObjectLockDiagMode::Read, + ); let reader = GetObjectReader { stream: Box::new(Cursor::new(vec![1, 2, 3])), object_info: ObjectInfo::default(), diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 21943aad1..ff742c69f 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -112,7 +112,8 @@ pub use deadlock_metrics::{ // Lock metrics exports pub use lock_metrics::{ LockMetricsSummary, record_contention_event, record_early_release, record_lock_hold_time, record_lock_optimization_enabled, - record_spin_attempt, record_spin_count_change, + record_object_lock_diag_acquire_duration, record_object_lock_diag_enabled, record_object_lock_diag_hold_duration, + record_object_lock_diag_slow_acquire, record_object_lock_diag_slow_hold, record_spin_attempt, record_spin_count_change, }; pub use process_lock_metrics::{ diff --git a/crates/io-metrics/src/lock_metrics.rs b/crates/io-metrics/src/lock_metrics.rs index da1df38f6..663510b06 100644 --- a/crates/io-metrics/src/lock_metrics.rs +++ b/crates/io-metrics/src/lock_metrics.rs @@ -62,6 +62,61 @@ pub fn record_contention_event() { counter!("rustfs_lock_contentions").increment(1); } +/// Record object namespace lock diagnostics being enabled. +#[inline(always)] +pub fn record_object_lock_diag_enabled(enabled: bool) { + use metrics::gauge; + gauge!("rustfs_object_lock_diag_enabled").set(if enabled { 1.0 } else { 0.0 }); +} + +/// Record object namespace lock acquire duration. +#[inline(always)] +pub fn record_object_lock_diag_acquire_duration(op: &'static str, mode: &'static str, duration: Duration) { + use metrics::histogram; + histogram!( + "rustfs_object_lock_diag_acquire_duration_seconds", + "op" => op, + "mode" => mode + ) + .record(duration.as_secs_f64()); +} + +/// Record object namespace lock hold duration. +#[inline(always)] +pub fn record_object_lock_diag_hold_duration(op: &'static str, mode: &'static str, duration: Duration) { + use metrics::histogram; + histogram!( + "rustfs_object_lock_diag_hold_duration_seconds", + "op" => op, + "mode" => mode + ) + .record(duration.as_secs_f64()); +} + +/// Record an object namespace lock slow-acquire event. +#[inline(always)] +pub fn record_object_lock_diag_slow_acquire(op: &'static str, mode: &'static str) { + use metrics::counter; + counter!( + "rustfs_object_lock_diag_slow_acquire_total", + "op" => op, + "mode" => mode + ) + .increment(1); +} + +/// Record an object namespace lock slow-hold event. +#[inline(always)] +pub fn record_object_lock_diag_slow_hold(op: &'static str, mode: &'static str) { + use metrics::counter; + counter!( + "rustfs_object_lock_diag_slow_hold_total", + "op" => op, + "mode" => mode + ) + .increment(1); +} + /// Lock statistics summary. #[derive(Debug, Clone, Default)] pub struct LockMetricsSummary { @@ -108,6 +163,97 @@ impl LockMetricsSummary { #[cfg(test)] mod tests { use super::*; + use metrics::{Counter, CounterFn, Gauge, GaugeFn, Histogram, HistogramFn, Key, KeyName, Metadata, SharedString, Unit}; + use std::sync::{Arc, Mutex}; + + #[derive(Clone, Default)] + struct SeenMetricsRecorder { + counters: Arc>>, + gauges: Arc>>, + histograms: Arc>>, + } + + impl SeenMetricsRecorder { + fn saw_counter_named(&self, name: &str) -> bool { + self.counters + .lock() + .expect("counter key collection should be lockable") + .iter() + .any(|key| key.name() == name) + } + + fn saw_gauge_named(&self, name: &str) -> bool { + self.gauges + .lock() + .expect("gauge key collection should be lockable") + .iter() + .any(|key| key.name() == name) + } + + fn saw_histogram_named(&self, name: &str) -> bool { + self.histograms + .lock() + .expect("histogram key collection should be lockable") + .iter() + .any(|key| key.name() == name) + } + } + + impl metrics::Recorder for SeenMetricsRecorder { + fn describe_counter(&self, _key: KeyName, _unit: Option, _description: SharedString) {} + + fn describe_gauge(&self, _key: KeyName, _unit: Option, _description: SharedString) {} + + fn describe_histogram(&self, _key: KeyName, _unit: Option, _description: SharedString) {} + + fn register_counter(&self, key: &Key, _metadata: &Metadata<'_>) -> Counter { + self.counters + .lock() + .expect("counter key collection should be lockable") + .push(key.clone()); + Counter::from_arc(Arc::new(NoopCounter)) + } + + fn register_gauge(&self, key: &Key, _metadata: &Metadata<'_>) -> Gauge { + self.gauges + .lock() + .expect("gauge key collection should be lockable") + .push(key.clone()); + Gauge::from_arc(Arc::new(NoopGauge)) + } + + fn register_histogram(&self, key: &Key, _metadata: &Metadata<'_>) -> Histogram { + self.histograms + .lock() + .expect("histogram key collection should be lockable") + .push(key.clone()); + Histogram::from_arc(Arc::new(NoopHistogram)) + } + } + + struct NoopCounter; + + impl CounterFn for NoopCounter { + fn increment(&self, _value: u64) {} + + fn absolute(&self, _value: u64) {} + } + + struct NoopGauge; + + impl GaugeFn for NoopGauge { + fn increment(&self, _value: f64) {} + + fn decrement(&self, _value: f64) {} + + fn set(&self, _value: f64) {} + } + + struct NoopHistogram; + + impl HistogramFn for NoopHistogram { + fn record(&self, _value: f64) {} + } #[test] fn test_record_lock_optimization_enabled() { @@ -143,6 +289,60 @@ mod tests { record_contention_event(); } + #[test] + fn test_record_object_lock_diag_enabled() { + let recorder = SeenMetricsRecorder::default(); + metrics::with_local_recorder(&recorder, || { + record_object_lock_diag_enabled(true); + record_object_lock_diag_enabled(false); + }); + assert!( + recorder.saw_gauge_named("rustfs_object_lock_diag_enabled"), + "expected object lock diagnostics enabled gauge to be emitted" + ); + } + + #[test] + fn test_record_object_lock_diag_acquire_duration() { + let recorder = SeenMetricsRecorder::default(); + metrics::with_local_recorder(&recorder, || { + record_object_lock_diag_acquire_duration("get_object", "read", Duration::from_millis(10)); + }); + assert!( + recorder.saw_histogram_named("rustfs_object_lock_diag_acquire_duration_seconds"), + "expected object lock diagnostics acquire histogram to be emitted" + ); + } + + #[test] + fn test_record_object_lock_diag_hold_duration() { + let recorder = SeenMetricsRecorder::default(); + metrics::with_local_recorder(&recorder, || { + record_object_lock_diag_hold_duration("put_object_commit", "write", Duration::from_millis(20)); + }); + assert!( + recorder.saw_histogram_named("rustfs_object_lock_diag_hold_duration_seconds"), + "expected object lock diagnostics hold histogram to be emitted" + ); + } + + #[test] + fn test_record_object_lock_diag_slow_events() { + let recorder = SeenMetricsRecorder::default(); + metrics::with_local_recorder(&recorder, || { + record_object_lock_diag_slow_acquire("get_object_info", "read"); + record_object_lock_diag_slow_hold("complete_multipart_upload_commit", "write"); + }); + assert!( + recorder.saw_counter_named("rustfs_object_lock_diag_slow_acquire_total"), + "expected object lock diagnostics slow-acquire counter to be emitted" + ); + assert!( + recorder.saw_counter_named("rustfs_object_lock_diag_slow_hold_total"), + "expected object lock diagnostics slow-hold counter to be emitted" + ); + } + #[test] fn test_lock_metrics_summary() { let mut summary = LockMetricsSummary::new(); diff --git a/crates/lock/src/namespace/mod.rs b/crates/lock/src/namespace/mod.rs index 690993aeb..601e6a35e 100644 --- a/crates/lock/src/namespace/mod.rs +++ b/crates/lock/src/namespace/mod.rs @@ -52,6 +52,11 @@ impl NamespaceLockWrapper { Self { lock, resource, owner } } + /// Get lock owner identifier used for acquisition. + pub fn owner(&self) -> &str { + &self.owner + } + /// Acquire write lock (exclusive lock) with timeout /// Returns the guard if acquisition succeeds, or an error if it fails pub async fn get_write_lock(&self, timeout: Duration) -> std::result::Result {