From 4283591838d4b2bbc159a9e93ffc9e4c33518dfd Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 21 Aug 2026 09:18:53 +0800 Subject: [PATCH] feat(ecstore): observe PUT commit lock admission (#6319) * feat(ecstore): observe PUT commit lock admission Co-Authored-By: heihutu * update h2 v0.4.18 * test(e2e): box SSE-KMS negative errors Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- Cargo.lock | 4 +- crates/config/README.md | 9 + .../kms_authorization_negative_matrix_test.rs | 7 +- .../ecstore/src/bucket/bucket_target_sys.rs | 18 +- .../replication/replication_resyncer.rs | 16 +- .../replication_target_boundary.rs | 3 +- crates/ecstore/src/set_disk/mod.rs | 380 +++++++++++++++++- crates/io-metrics/src/lib.rs | 95 +++++ crates/s3select-query/src/lib.rs | 2 + 9 files changed, 500 insertions(+), 34 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2b9919484..312fc1376 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4757,9 +4757,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.17" +version = "0.4.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9f877e75f39e9827ec50a572dd592684ac28c029578726c85f1b2aa6ab807449" +checksum = "839c0e8a181239723652be9062bb56ca5bf5f64011f73b623f6f4fc59086a228" dependencies = [ "atomic-waker", "bytes", diff --git a/crates/config/README.md b/crates/config/README.md index 0c703b57c..02cf81d52 100644 --- a/crates/config/README.md +++ b/crates/config/README.md @@ -115,6 +115,15 @@ Current guidance: - enables KMS readiness enforcement for `/health/ready`. - default is `false`. +## Object lock admission environment variables + +- `RUSTFS_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS` + - experimental same-object PUT commit namespace-lock admission budget. + - default is `0`, which disables this override and keeps `RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT` behavior. + - when set, only `put_object_commit` write-lock acquisition is bounded by this millisecond budget; other namespace lock users keep the global object-lock timeout. + - timeout returns S3 `SlowDown`, so clients should use normal SDK retry handling. + - this is not a fdatasync or group-commit switch. Track fdatasync batching separately with `rustfs_s3_put_object_rename_fdatasync_batch_files`. + ## Drive timeout environment variables - `RUSTFS_DRIVE_METADATA_TIMEOUT_SECS` diff --git a/crates/e2e_test/src/kms/kms_authorization_negative_matrix_test.rs b/crates/e2e_test/src/kms/kms_authorization_negative_matrix_test.rs index ac212536d..a2bfedc11 100644 --- a/crates/e2e_test/src/kms/kms_authorization_negative_matrix_test.rs +++ b/crates/e2e_test/src/kms/kms_authorization_negative_matrix_test.rs @@ -39,6 +39,7 @@ use std::time::Duration; use tracing::info; type TestResult = Result<(), Box>; +type S3OperationResult = Result>; const ALLOWED_KEY: &str = "kms-matrix-allowed-key"; const OTHER_KEY: &str = "kms-matrix-other-key"; @@ -130,7 +131,7 @@ fn policy_document(statements: Vec) -> String { serde_json::json!({ "Version": "2012-10-17", "Statement": statements }).to_string() } -async fn put_sse_kms(client: &Client, key: &str, kms_key_id: &str) -> Result<(), Box> { +async fn put_sse_kms(client: &Client, key: &str, kms_key_id: &str) -> S3OperationResult<()> { client .put_object() .bucket(BUCKET) @@ -303,7 +304,7 @@ async fn sse_kms_per_key_authorization_negative_matrix() -> TestResult { .send() .await .map(|_| ()) - .map_err(aws_sdk_s3::Error::from), + .map_err(|err| Box::new(aws_sdk_s3::Error::from(err))), "SSE-KMS read by an identity holding no kms grant", ); @@ -317,7 +318,7 @@ async fn sse_kms_per_key_authorization_negative_matrix() -> TestResult { .send() .await .map(|_| ()) - .map_err(aws_sdk_s3::Error::from), + .map_err(|err| Box::new(aws_sdk_s3::Error::from(err))), "SSE-KMS read by an identity holding kms:GenerateDataKey but not kms:Decrypt", ); diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 45cbb87c6..f5ac0c013 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -90,6 +90,12 @@ use uuid::Uuid; const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16; const REDACTED_CREDENTIAL: &str = ""; +pub type HeadObjectSdkError = Box>; +pub type GetObjectSdkError = Box>; +pub type GetObjectTaggingSdkError = Box>; +pub type PutObjectTaggingSdkError = Box>; +pub type DeleteObjectTaggingSdkError = Box>; + pub static GLOBAL_BUCKET_TARGET_SYS: OnceLock = OnceLock::new(); fn replication_target_versioning_enabled(versioning: Option<&BucketVersioningStatus>) -> bool { @@ -1968,7 +1974,7 @@ impl TargetClient { bucket: &str, object: &str, version_id: Option, - ) -> Result>> { + ) -> Result { // Announce the replication check so a RustFS target returns SSE-C // object metadata (etag/size) without the customer key the replication // worker cannot hold; otherwise SSE-C replicas never converge on HEAD. @@ -2019,7 +2025,7 @@ impl TargetClient { range: Option, part_number: Option, extra_headers: HeaderMap, - ) -> Result>> { + ) -> Result { let headers = proxy_outbound_headers(extra_headers); self.client .head_object() @@ -2048,7 +2054,7 @@ impl TargetClient { range: Option, part_number: Option, extra_headers: HeaderMap, - ) -> Result>> { + ) -> Result { let headers = proxy_outbound_headers(extra_headers); self.client .get_object() @@ -2071,7 +2077,7 @@ impl TargetClient { bucket: &str, object: &str, version_id: Option, - ) -> Result>> { + ) -> Result { let headers = proxy_outbound_headers(HeaderMap::new()); self.client .get_object_tagging() @@ -2093,7 +2099,7 @@ impl TargetClient { object: &str, version_id: Option, tagging: SdkTagging, - ) -> Result>> { + ) -> Result { let headers = proxy_outbound_headers(HeaderMap::new()); self.client .put_object_tagging() @@ -2115,7 +2121,7 @@ impl TargetClient { bucket: &str, object: &str, version_id: Option, - ) -> Result>> { + ) -> Result { let headers = proxy_outbound_headers(HeaderMap::new()); self.client .delete_object_tagging() diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 5ec5bb03e..d921ea411 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -52,8 +52,8 @@ use super::replication_storage_boundary::{ ReplicationObjectIO, ReplicationStorage, StatObjectOptions, StorageObjectInfoOrErr, WalkOptions, }; use super::replication_target_boundary::{ - ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, - SsecPassthroughCapability, SsecPassthroughGate, TargetClient, is_replication_target_offline_error, + ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, + ReplicationTargetStore, SsecPassthroughCapability, SsecPassthroughGate, TargetClient, is_replication_target_offline_error, replication_action_for_target_head, replication_complete_multipart_options, replication_delete_marker_purge_remove_options, replication_delete_remove_options, replication_force_delete_remove_options, replication_object_is_ssec_encrypted, replication_put_object_header_size, replication_put_object_options, replication_target_head_is_newer_null_version, @@ -214,7 +214,7 @@ async fn head_object_for_worker( target_bucket: &str, object: &str, version_id: Option, -) -> std::result::Result>> { +) -> std::result::Result { target_client.head_object(target_bucket, object, version_id).await } @@ -233,7 +233,7 @@ async fn mark_replication_target_offline_if_needed(target_client: &Arc std::result::Result, Box>> { +) -> std::result::Result, HeadObjectSdkError> { match head_object_for_worker(tgt_client, &tgt_client.bucket, object, None).await { Ok(oi) => Ok(Some(oi)), Err(e) if e.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(&e, 404) => Ok(None), @@ -1152,11 +1152,11 @@ fn spawn_resync_walk_task( /// updating the per-object status counters and returning the accounted size /// together with any verification error. async fn verify_resync_head_result( - head_result: std::result::Result>>, + head_result: std::result::Result, roi: &ReplicateObjectInfo, st: &mut TargetReplicationResyncStatus, target_client: &Arc, -) -> (i64, Option>>) { +) -> (i64, Option) { match head_result { Ok(_) => { st.replicated_count += 1; @@ -1275,7 +1275,7 @@ async fn resync_worker_process_object( "Processed resync object" ); } - st.error = err.as_ref().and_then(|err| resync_target_error_detail(err)); + st.error = err.as_ref().and_then(|err| resync_target_error_detail(err.as_ref())); st } @@ -2467,7 +2467,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli Ok(_) => {} Err(e) => { let non_retryable = matches!( - &*e, + e.as_ref(), SdkError::ServiceError(service_err) if is_retryable_delete_replication_head_error( service_err.err().is_not_found(), diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index 87d94779e..455a37ed0 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -36,7 +36,8 @@ use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; pub(crate) use crate::bucket::bucket_target_sys::{ - AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient, resolve_read_api_version_id, + AdvancedPutOptions, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient, + resolve_read_api_version_id, }; #[cfg(test)] pub(crate) use crate::bucket::target::BucketTarget; diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 1373fe272..5625d7583 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -1539,6 +1539,50 @@ fn put_object_commit_lock_timeout_override_enabled(op: &'static str) -> bool { op == "put_object_commit" && get_put_object_commit_lock_acquire_timeout_override_ms() != 0 } +fn put_object_commit_lock_admission_budget_label() -> &'static str { + match get_put_object_commit_lock_acquire_timeout_override_ms() { + 0 => rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_DISABLED, + 1..=250 => rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS, + 251..=500 => rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS, + 501..=1000 => rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_1000MS, + _ => rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_GT_1000MS, + } +} + +fn record_put_object_commit_lock_admission(op: &'static str, outcome: &'static str) { + if op != "put_object_commit" || !rustfs_io_metrics::put_stage_metrics_enabled() { + return; + } + rustfs_io_metrics::record_put_object_commit_lock_admission(put_object_commit_lock_admission_budget_label(), outcome); +} + +fn put_object_commit_lock_acquire_error_outcome(op: &'static str, err: &rustfs_lock::error::LockError) -> &'static str { + if put_object_commit_lock_timeout_override_enabled(op) && matches!(err, rustfs_lock::error::LockError::Timeout { .. }) { + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN + } else { + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_LOCK_ERROR + } +} + +fn resolve_put_object_commit_lock_acquire_result( + set: &SetDisks, + op: &'static str, + bucket: &str, + object: &str, + result: std::result::Result, +) -> Result { + match result { + Ok(guard) => { + record_put_object_commit_lock_admission(op, rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED); + Ok(guard) + } + Err(err) => { + record_put_object_commit_lock_admission(op, put_object_commit_lock_acquire_error_outcome(op, &err)); + Err(map_put_object_commit_lock_acquire_error(set, op, bucket, object, err)) + } + } +} + fn map_put_object_commit_lock_acquire_error( set: &SetDisks, op: &'static str, @@ -3355,10 +3399,13 @@ impl SetDisks { let ns_lock = self.new_ns_lock(bucket, object).await?; let acquire_start = Instant::now(); let acquire_timeout = get_put_object_commit_lock_acquire_timeout(op); - let guard = ns_lock - .get_write_lock(acquire_timeout) - .await - .map_err(|e| map_put_object_commit_lock_acquire_error(self, op, bucket, object, e))?; + let guard = resolve_put_object_commit_lock_acquire_result( + self, + op, + bucket, + object, + ns_lock.get_write_lock(acquire_timeout).await, + )?; Self::record_put_object_commit_namespace_lock_wait(op, acquire_start); let owner = diag_enabled.then(|| ns_lock.owner().to_string()); self.log_object_lock_acquire_if_slow( @@ -3397,17 +3444,22 @@ impl SetDisks { let acquire = ns_lock.get_write_lock(acquire_timeout); tokio::pin!(acquire); let mut on_pending = Some(on_pending); - let guard = futures::future::poll_fn(|cx| match std::future::Future::poll(acquire.as_mut(), cx) { - std::task::Poll::Pending => { - if let Some(on_pending) = on_pending.take() { - on_pending(); + let guard = resolve_put_object_commit_lock_acquire_result( + self, + op, + bucket, + object, + futures::future::poll_fn(|cx| match std::future::Future::poll(acquire.as_mut(), cx) { + std::task::Poll::Pending => { + if let Some(on_pending) = on_pending.take() { + on_pending(); + } + std::task::Poll::Pending } - std::task::Poll::Pending - } - std::task::Poll::Ready(result) => std::task::Poll::Ready(result), - }) - .await - .map_err(|e| map_put_object_commit_lock_acquire_error(self, op, bucket, object, e))?; + std::task::Poll::Ready(result) => std::task::Poll::Ready(result), + }) + .await, + )?; Self::record_put_object_commit_namespace_lock_wait(op, acquire_start); let owner = diag_enabled.then(|| ns_lock.owner().to_string()); self.log_object_lock_acquire_if_slow( @@ -5782,6 +5834,81 @@ mod tests { .sum() } + fn put_object_commit_lock_admission_count( + rows: &[( + metrics_util::CompositeKey, + Option, + Option, + DebugValue, + )], + budget: &'static str, + outcome: &'static str, + ) -> u64 { + rows.iter() + .filter(|(composite, _, _, _)| { + composite.key().name() == "rustfs_s3_put_object_commit_namespace_lock_admission_total" + && composite + .key() + .labels() + .any(|label| label.key() == "budget" && label.value() == budget) + && composite + .key() + .labels() + .any(|label| label.key() == "outcome" && label.value() == outcome) + }) + .map(|(_, _, _, value)| match value { + DebugValue::Counter(count) => *count, + _ => 0, + }) + .sum() + } + + #[test] + #[serial] + fn put_object_commit_lock_admission_budget_labels_are_bounded() { + let cases = [ + ("0", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_DISABLED), + ("250", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS), + ("251", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS), + ("500", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS), + ("501", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_1000MS), + ("1000", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_1000MS), + ("1001", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_GT_1000MS), + ]; + for (timeout_ms, expected) in cases { + temp_env::with_vars( + [(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some(timeout_ms))], + || { + assert_eq!(put_object_commit_lock_admission_budget_label(), expected); + }, + ); + } + } + + #[test] + #[serial] + fn put_object_commit_lock_admission_error_outcomes_are_bounded() { + let timeout = LockError::timeout("bucket/object", Duration::from_millis(1)); + temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("1"))], || { + assert_eq!( + put_object_commit_lock_acquire_error_outcome("put_object_commit", &timeout), + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN + ); + assert_eq!( + put_object_commit_lock_acquire_error_outcome("complete_multipart_upload_commit", &timeout), + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_LOCK_ERROR + ); + }); + + let internal = LockError::internal("simulated lock manager error"); + temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("1"))], || { + assert_eq!( + put_object_commit_lock_acquire_error_outcome("put_object_commit", &internal), + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_LOCK_ERROR + ); + }); + } + #[test] #[serial] fn put_object_commit_namespace_lock_wait_metric_is_wired_to_both_write_lock_paths() { @@ -5905,6 +6032,231 @@ mod tests { }); } + #[test] + #[serial] + fn put_object_commit_lock_admission_records_acquired_and_timeout() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime should start"); + + temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("1"))], || { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + rustfs_io_metrics::set_put_stage_metrics_enabled(true); + runtime.block_on(async { + let ctx = Arc::new(InstanceContext::new()); + ctx.update_erasure_type(SetupType::Erasure).await; + let set = make_test_set_disks_with_ctx(Vec::new(), ctx).await; + let held_guard = set + .acquire_write_lock_diag("put_object_commit", "bucket", "object") + .await + .expect("holder acquire should succeed"); + let err = match set.acquire_write_lock_diag("put_object_commit", "bucket", "object").await { + Ok(_) => panic!("contended PUT commit acquire should return SlowDown"), + Err(err) => err, + }; + assert!(matches!(err, StorageError::SlowDown)); + drop(held_guard); + rustfs_io_metrics::set_put_stage_metrics_enabled(false); + }); + }); + + let rows = snapshotter.snapshot().into_vec(); + assert_eq!( + put_object_commit_lock_admission_count( + &rows, + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS, + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED, + ), + 1 + ); + assert_eq!( + put_object_commit_lock_admission_count( + &rows, + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS, + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN, + ), + 1 + ); + }); + } + + #[test] + #[serial] + fn put_object_commit_lock_admission_records_disabled_budget_acquired() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime should start"); + + temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("0"))], || { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + rustfs_io_metrics::set_put_stage_metrics_enabled(true); + runtime.block_on(async { + let ctx = Arc::new(InstanceContext::new()); + ctx.update_erasure_type(SetupType::Erasure).await; + let set = make_test_set_disks_with_ctx(Vec::new(), ctx).await; + let guard = set + .acquire_write_lock_diag("put_object_commit", "bucket", "object") + .await + .expect("PUT commit acquire should succeed with default timeout"); + drop(guard); + rustfs_io_metrics::set_put_stage_metrics_enabled(false); + }); + }); + + let rows = snapshotter.snapshot().into_vec(); + assert_eq!( + put_object_commit_lock_admission_count( + &rows, + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_DISABLED, + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED, + ), + 1 + ); + }); + } + + #[test] + #[serial] + fn put_object_commit_lock_admission_skips_non_put_commit_ops() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime should start"); + + temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("250"))], || { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + rustfs_io_metrics::set_put_stage_metrics_enabled(true); + runtime.block_on(async { + let ctx = Arc::new(InstanceContext::new()); + ctx.update_erasure_type(SetupType::Erasure).await; + let set = make_test_set_disks_with_ctx(Vec::new(), ctx).await; + let guard = set + .acquire_write_lock_diag("complete_multipart_upload_commit", "bucket", "object") + .await + .expect("non-PUT commit acquire should succeed"); + drop(guard); + rustfs_io_metrics::set_put_stage_metrics_enabled(false); + }); + }); + + let rows = snapshotter.snapshot().into_vec(); + assert_eq!( + rows.iter() + .filter(|(composite, _, _, _)| { + composite.key().name() == "rustfs_s3_put_object_commit_namespace_lock_admission_total" + }) + .count(), + 0 + ); + }); + } + + #[test] + #[serial] + fn put_object_commit_lock_admission_records_lock_error() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime should start"); + + temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("250"))], || { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + rustfs_io_metrics::set_put_stage_metrics_enabled(true); + runtime.block_on(async { + let healthy: Arc = + Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new()))); + let failing: Arc = Arc::new(FailingClient); + let ctx = Arc::new(InstanceContext::new()); + ctx.update_erasure_type(SetupType::DistErasure).await; + let set = make_test_set_disks_with_ctx(vec![healthy, failing], ctx).await; + assert!( + set.acquire_write_lock_diag("put_object_commit", "bucket", "object") + .await + .is_err(), + "one healthy locker must not satisfy the PUT commit write quorum" + ); + rustfs_io_metrics::set_put_stage_metrics_enabled(false); + }); + }); + + let rows = snapshotter.snapshot().into_vec(); + assert_eq!( + put_object_commit_lock_admission_count( + &rows, + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS, + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_LOCK_ERROR, + ), + 1 + ); + assert_eq!( + put_object_commit_lock_admission_count( + &rows, + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS, + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN, + ), + 0 + ); + }); + } + + #[test] + #[serial] + fn put_object_commit_lock_admission_records_pending_hook_acquired() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime should start"); + + temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("500"))], || { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + rustfs_io_metrics::set_put_stage_metrics_enabled(true); + runtime.block_on(async { + let ctx = Arc::new(InstanceContext::new()); + ctx.update_erasure_type(SetupType::Erasure).await; + let set = make_test_set_disks_with_ctx(Vec::new(), ctx).await; + let held_guard = set + .acquire_write_lock_diag("put_object_commit", "bucket", "object") + .await + .expect("holder acquire should succeed"); + let (pending_tx, pending_rx) = tokio::sync::oneshot::channel(); + let pending_acquire = + set.acquire_write_lock_diag_with_pending_hook("put_object_commit", "bucket", "object", move || { + let _ = pending_tx.send(()); + }); + let release_holder = async { + pending_rx.await.expect("pending hook should fire"); + drop(held_guard); + }; + let (pending_guard, ()) = tokio::join!(pending_acquire, release_holder); + drop(pending_guard.expect("pending-hook PUT commit acquire should succeed")); + rustfs_io_metrics::set_put_stage_metrics_enabled(false); + }); + }); + + let rows = snapshotter.snapshot().into_vec(); + assert_eq!( + put_object_commit_lock_admission_count( + &rows, + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS, + rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED, + ), + 2 + ); + }); + } + #[tokio::test] async fn new_ns_lock_shares_clients_without_changing_quorum() { let healthy: Arc = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new()))); diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index a52f5adb6..7c37fc2de 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -121,6 +121,16 @@ pub const PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC: &str = "set_disk_rename_ba pub const PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC: &str = "set_disk_rename_ancestor_dir_fsync"; pub const PUT_STAGE_SET_DISK_RENAME_RENAME_SYSCALL: &str = "set_disk_rename_rename_syscall"; +pub const PUT_COMMIT_LOCK_ADMISSION_BUDGET_DISABLED: &str = "disabled"; +pub const PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS: &str = "le_250ms"; +pub const PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS: &str = "le_500ms"; +pub const PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_1000MS: &str = "le_1000ms"; +pub const PUT_COMMIT_LOCK_ADMISSION_BUDGET_GT_1000MS: &str = "gt_1000ms"; + +pub const PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED: &str = "acquired"; +pub const PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN: &str = "timeout_slowdown"; +pub const PUT_COMMIT_LOCK_ADMISSION_OUTCOME_LOCK_ERROR: &str = "lock_error"; + pub const PUT_RENAME_FDATASYNC_BATCH_MODE_SERIAL: &str = "serial"; pub const PUT_RENAME_FDATASYNC_BATCH_MODE_PARALLEL: &str = "parallel"; pub const PUT_RENAME_FDATASYNC_GROUP_WAIT_ROLE_LEADER: &str = "leader"; @@ -2060,6 +2070,14 @@ pub fn record_put_object_stage_duration_from(stage: &'static str, started_at: Op } } +#[inline(always)] +pub fn record_put_object_commit_lock_admission(budget: &'static str, outcome: &'static str) { + if !put_stage_metrics_enabled() { + return; + } + counter!("rustfs_s3_put_object_commit_namespace_lock_admission_total", "budget" => budget, "outcome" => outcome).increment(1); +} + #[inline(always)] fn put_stage_count_value(value: usize) -> f64 { match u32::try_from(value) { @@ -3204,6 +3222,83 @@ mod tests { assert!(stages.iter().all(|stage| recorded.contains(*stage))); } + #[test] + fn put_commit_lock_admission_labels_are_static_and_gated() { + let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let budgets = [ + PUT_COMMIT_LOCK_ADMISSION_BUDGET_DISABLED, + PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS, + PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS, + PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_1000MS, + PUT_COMMIT_LOCK_ADMISSION_BUDGET_GT_1000MS, + ]; + let outcomes = [ + PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED, + PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN, + PUT_COMMIT_LOCK_ADMISSION_OUTCOME_LOCK_ERROR, + ]; + assert_eq!(budgets.iter().copied().collect::>().len(), budgets.len()); + assert_eq!(outcomes.iter().copied().collect::>().len(), outcomes.len()); + assert!(budgets.iter().chain(outcomes.iter()).all(|label| { + !label.contains('/') + && !label.contains('{') + && !label.contains('}') + && !label.contains(' ') + && label + .chars() + .all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '_') + })); + + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + set_put_stage_metrics_enabled(false); + record_put_object_commit_lock_admission( + PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS, + PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN, + ); + + set_put_stage_metrics_enabled(true); + record_put_object_commit_lock_admission( + PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS, + PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN, + ); + record_put_object_commit_lock_admission( + PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS, + PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED, + ); + set_put_stage_metrics_enabled(false); + }); + + let rows = snapshotter.snapshot().into_vec(); + assert_eq!( + counter_total(&rows, "rustfs_s3_put_object_commit_namespace_lock_admission_total"), + Some(2) + ); + let label_sets = rows + .iter() + .filter(|(composite, _, _, _)| { + composite.kind() == MetricKind::Counter + && composite.key().name() == "rustfs_s3_put_object_commit_namespace_lock_admission_total" + }) + .map(|(composite, _, _, _)| { + composite + .key() + .labels() + .map(|label| (label.key().to_string(), label.value().to_string())) + .collect::>() + }) + .collect::>(); + assert!(label_sets.contains(&HashSet::from([ + ("budget".to_string(), PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS.to_string()), + ("outcome".to_string(), PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN.to_string(),), + ]))); + assert!(label_sets.contains(&HashSet::from([ + ("budget".to_string(), PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS.to_string()), + ("outcome".to_string(), PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED.to_string()), + ]))); + } + #[test] fn put_rename_code_level_metrics_are_static_and_gated() { let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner()); diff --git a/crates/s3select-query/src/lib.rs b/crates/s3select-query/src/lib.rs index cfc0d0702..fade8f9bb 100644 --- a/crates/s3select-query/src/lib.rs +++ b/crates/s3select-query/src/lib.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +#![recursion_limit = "256"] + pub mod data_source; pub mod dispatcher; pub mod execution;