gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-21 11:56:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(ecstore): observe PUT commit lock admission

Browse Source 
Co-Authored-By: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-21 01:13:45 +08:00
parent cee0d5cf9b
commit bae3d2022f
16 changed files with 517 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/common/src/metrics.rs
+1 -1
View File
@@ -729,7 +729,7 @@ fn timestamp_elapsed_seconds_since(now: Timestamp, earlier: Timestamp) -> u64 {
return 0;
}
u64::try_from(duration.as_secs()).map_or(u64::MAX, |seconds| seconds)
u64::try_from(duration.as_secs()).unwrap_or(u64::MAX)
}
#[derive(Clone, Copy, Debug, Default)]
crates/config/README.md
+9
View File
@@ -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`
crates/ecstore/src/bucket/bucket_target_sys.rs
+18 -7
View File
@@ -90,6 +90,12 @@ use uuid::Uuid;
const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16;
const REDACTED_CREDENTIAL: &str = "<redacted>";
pub type HeadObjectSdkError = Box<SdkError<HeadObjectError>>;
pub type GetObjectSdkError = Box<SdkError<GetObjectError>>;
pub type GetObjectTaggingSdkError = Box<SdkError<GetObjectTaggingError>>;
pub type PutObjectTaggingSdkError = Box<SdkError<PutObjectTaggingError>>;
pub type DeleteObjectTaggingSdkError = Box<SdkError<DeleteObjectTaggingError>>;
pub static GLOBAL_BUCKET_TARGET_SYS: OnceLock<BucketTargetSys> = OnceLock::new();
fn replication_target_versioning_enabled(versioning: Option<&BucketVersioningStatus>) -> bool {
@@ -1968,7 +1974,7 @@ impl TargetClient {
bucket: &str,
object: &str,
version_id: Option<String>,
) -> Result<HeadObjectOutput, SdkError<HeadObjectError>> {
) -> Result<HeadObjectOutput, HeadObjectSdkError> {
// 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.
@@ -2001,7 +2007,7 @@ impl TargetClient {
.await
{
Ok(res) => Ok(res),
Err(e) => Err(e),
Err(e) => Err(Box::new(e)),
}
}
@@ -2023,7 +2029,7 @@ impl TargetClient {
range: Option<String>,
part_number: Option<i32>,
extra_headers: HeaderMap,
) -> Result<HeadObjectOutput, SdkError<HeadObjectError>> {
) -> Result<HeadObjectOutput, HeadObjectSdkError> {
let headers = proxy_outbound_headers(extra_headers);
self.client
.head_object()
@@ -2036,6 +2042,7 @@ impl TargetClient {
.map_request(move |req| apply_extra_headers(req, &headers))
.send()
.await
.map_err(Box::new)
}
/// GET used by the read-proxy path (MinIO `proxyGetToReplicationTarget`).
@@ -2051,7 +2058,7 @@ impl TargetClient {
range: Option<String>,
part_number: Option<i32>,
extra_headers: HeaderMap,
) -> Result<GetObjectOutput, SdkError<GetObjectError>> {
) -> Result<GetObjectOutput, GetObjectSdkError> {
let headers = proxy_outbound_headers(extra_headers);
self.client
.get_object()
@@ -2064,6 +2071,7 @@ impl TargetClient {
.map_request(move |req| apply_extra_headers(req, &headers))
.send()
.await
.map_err(Box::new)
}
/// GetObjectTagging for the tagging read-proxy path
@@ -2073,7 +2081,7 @@ impl TargetClient {
bucket: &str,
object: &str,
version_id: Option<String>,
) -> Result<GetObjectTaggingOutput, SdkError<GetObjectTaggingError>> {
) -> Result<GetObjectTaggingOutput, GetObjectTaggingSdkError> {
let headers = proxy_outbound_headers(HeaderMap::new());
self.client
.get_object_tagging()
@@ -2084,6 +2092,7 @@ impl TargetClient {
.map_request(move |req| apply_extra_headers(req, &headers))
.send()
.await
.map_err(Box::new)
}
/// PutObjectTagging for the tagging proxy path
@@ -2094,7 +2103,7 @@ impl TargetClient {
object: &str,
version_id: Option<String>,
tagging: SdkTagging,
) -> Result<PutObjectTaggingOutput, SdkError<PutObjectTaggingError>> {
) -> Result<PutObjectTaggingOutput, PutObjectTaggingSdkError> {
let headers = proxy_outbound_headers(HeaderMap::new());
self.client
.put_object_tagging()
@@ -2106,6 +2115,7 @@ impl TargetClient {
.map_request(move |req| apply_extra_headers(req, &headers))
.send()
.await
.map_err(Box::new)
}
/// DeleteObjectTagging for the tagging proxy path
@@ -2115,7 +2125,7 @@ impl TargetClient {
bucket: &str,
object: &str,
version_id: Option<String>,
) -> Result<DeleteObjectTaggingOutput, SdkError<DeleteObjectTaggingError>> {
) -> Result<DeleteObjectTaggingOutput, DeleteObjectTaggingSdkError> {
let headers = proxy_outbound_headers(HeaderMap::new());
self.client
.delete_object_tagging()
@@ -2126,6 +2136,7 @@ impl TargetClient {
.map_request(move |req| apply_extra_headers(req, &headers))
.send()
.await
.map_err(Box::new)
}
/// On success returns the version id the target assigned (from
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+2 -2
View File
@@ -2180,7 +2180,7 @@ pub async fn recover_manual_transition_jobs_once(
if limit == 0 {
return Err(Error::other("manual transition job recovery limit must be greater than zero"));
}
let list_limit = i32::try_from(limit).map_or(i32::MAX, |value| value);
let list_limit = i32::try_from(limit).unwrap_or(i32::MAX);
let page = api
.clone()
.list_objects_v2(
@@ -2386,7 +2386,7 @@ async fn replay_manual_transition_pending_tasks(
version_id: task.version_id,
etag: task.etag,
mod_time,
size: task.size.map_or(0, |size| size),
size: task.size.unwrap_or(0),
is_latest: task.is_latest.unwrap_or(false),
..Default::default()
};
crates/ecstore/src/bucket/lifecycle/transition_transaction.rs
+1 -1
View File
@@ -1016,7 +1016,7 @@ pub async fn recover_transition_transaction_records(
return Err(Error::other("transition transaction recovery limit must be greater than zero"));
}
let list_limit = i32::try_from(limit).map_or(i32::MAX, |value| value);
let list_limit = i32::try_from(limit).unwrap_or(i32::MAX);
let list = api
.clone()
.list_objects_v2(
crates/ecstore/src/bucket/replication/replication_resyncer.rs
+8 -8
View File
@@ -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<String>,
) -> std::result::Result<HeadObjectOutput, SdkError<HeadObjectError>> {
) -> std::result::Result<HeadObjectOutput, HeadObjectSdkError> {
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<TargetCli
async fn head_object_fallback(
tgt_client: &TargetClient,
object: &str,
) -> std::result::Result<Option<HeadObjectOutput>, SdkError<HeadObjectError>> {
) -> std::result::Result<Option<HeadObjectOutput>, 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<S: ReplicationStorage>(
/// 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<HeadObjectOutput, SdkError<HeadObjectError>>,
head_result: std::result::Result<HeadObjectOutput, HeadObjectSdkError>,
roi: &ReplicateObjectInfo,
st: &mut TargetReplicationResyncStatus,
target_client: &Arc<TargetClient>,
) -> (i64, Option<SdkError<HeadObjectError>>) {
) -> (i64, Option<HeadObjectSdkError>) {
match head_result {
Ok(_) => {
st.replicated_count += 1;
@@ -1275,7 +1275,7 @@ async fn resync_worker_process_object<S: ReplicationStorage>(
"Processed resync object"
);
}
st.error = err.as_ref().and_then(resync_target_error_detail);
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(),
crates/ecstore/src/bucket/replication/replication_target_boundary.rs
+2 -1
View File
@@ -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;
crates/ecstore/src/services/tier/tier_mutation_intent.rs
+1 -1
View File
@@ -657,7 +657,7 @@ where
prefix,
marker,
None,
i32::try_from(limit).map_or(i32::MAX, |value| value),
i32::try_from(limit).unwrap_or(i32::MAX),
false,
None,
false,
crates/ecstore/src/set_disk/mod.rs
+366 -14
View File
@@ -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<rustfs_lock::namespace::NamespaceLockGuard, rustfs_lock::error::LockError>,
) -> Result<rustfs_lock::namespace::NamespaceLockGuard> {
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<metrics::Unit>,
Option<metrics::SharedString>,
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<dyn LockClient> =
Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
let failing: Arc<dyn LockClient> = 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<dyn LockClient> = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
crates/ecstore/src/set_disk/ops/object.rs
+8 -8
View File
@@ -3766,7 +3766,7 @@ pub(crate) async fn complete_transition_upload<Remote, Producer>(
producer: Producer,
expected_size: u64,
consumed: Arc<AtomicU64>,
) -> std::result::Result<TransitionUploadCompletion, TransitionUploadFailure>
) -> std::result::Result<TransitionUploadCompletion, Box<TransitionUploadFailure>>
where
Remote: Future<Output = std::result::Result<String, std::io::Error>>,
Producer: Future<Output = Result<u64>>,
@@ -3784,23 +3784,23 @@ where
Err(_) => StorageError::Unexpected,
Ok(Ok(_)) => StorageError::Io(remote_error),
};
return Err(TransitionUploadFailure { error, candidate: None });
return Err(Box::new(TransitionUploadFailure { error, candidate: None }));
}
};
let candidate = TransitionUploadCandidate::from_put_response(remote_version);
let produced = match producer_result {
Ok(Ok(produced)) => produced,
Ok(Err(error)) => {
return Err(TransitionUploadFailure {
return Err(Box::new(TransitionUploadFailure {
error,
candidate: Some(candidate),
});
}));
}
Err(_) => {
return Err(TransitionUploadFailure {
return Err(Box::new(TransitionUploadFailure {
error: StorageError::Unexpected,
candidate: Some(candidate),
});
}));
}
};
let consumed = consumed.load(Ordering::Acquire);
@@ -3810,10 +3810,10 @@ where
} else {
StorageError::MoreData
};
return Err(TransitionUploadFailure {
return Err(Box::new(TransitionUploadFailure {
error,
candidate: Some(candidate),
});
}));
}
Ok(TransitionUploadCompletion {
candidate,
crates/heal/src/heal/task/tests.rs
+1 -1
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::super::{DiskOption, DiskStore, Endpoint, HealDiskExt as _, new_disk};
use super::super::{DiskOption, DiskStore, Endpoint, new_disk};
use super::*;
use crate::heal::storage::{HealListItem, HealObjectInfo};
use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, TraceSubscription, TraceVal, subscribe_trace_events};
crates/io-metrics/src/lib.rs
+95
View File
@@ -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::<HashSet<_>>().len(), budgets.len());
assert_eq!(outcomes.iter().copied().collect::<HashSet<_>>().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::<HashSet<_>>()
})
.collect::<Vec<_>>();
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());
crates/obs/src/metrics/stats_collector.rs
+1 -1
View File
@@ -337,7 +337,7 @@ fn timestamp_elapsed_seconds_since(now: Timestamp, earlier: Timestamp) -> u64 {
return 0;
}
u64::try_from(duration.as_secs()).map_or(u64::MAX, |seconds| seconds)
u64::try_from(duration.as_secs()).unwrap_or(u64::MAX)
}
fn scanner_scan_mode_code(scan_mode: &str) -> u64 {
crates/s3select-query/src/lib.rs
+2
View File
@@ -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;
crates/scanner/src/remote_scanner/stream.rs
+1 -1
View File
@@ -22,7 +22,7 @@ use crate::scanner_io::{
use crate::storage_api::owner::NS_SCANNER_PROTOCOL_VERSION;
use crate::{
DATA_USAGE_CACHE_NAME, DataUsageCache, DataUsageCachePrepareOutcome, DataUsageCacheSource, DataUsageEntryInfo,
DataUsageScanPlanDigest, Disk, ScannerDiskExt as _, ScannerError, StorageError, resolve_scanner_object_store_handle,
DataUsageScanPlanDigest, Disk, ScannerError, StorageError, resolve_scanner_object_store_handle,
};
use hmac::{Hmac, KeyInit, Mac};
use rustfs_common::heal_channel::HealScanMode;
rustfs/src/admin/router.rs
+1 -1
View File
@@ -2694,7 +2694,7 @@ async fn ssec_passthrough_probe_object(
let head = target_client
.head_object(target_bucket, probe_key, head_version)
.await
.map_err(S3ClientError::from)?;
.map_err(|err| S3ClientError::from(*err))?;
Ok(ReplicationSsecProbeOutcome {
evidence_present: head.sse_customer_algorithm().is_some_and(|algorithm| !algorithm.is_empty()),