mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-13 09:17:41 +08:00
fix: reconcile bucket generations and observe lock failures (#8421)
* fix: reconcile bucket generations and observe lock failures Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix: fence bucket recovery through cancellation and quota reuse Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(heal): route tests through storage API boundary Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(ecstore): use typed lock errors in recovery Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser
co-authored by
heihutu
zhi22915
heihutu
zhi22915
parent
d386f7741d
commit
8474feaa26
@@ -1153,7 +1153,7 @@ async fn acquire_transaction_lock_with_sys(
|
||||
/// agree on, so renaming it would leave a mixed-version cluster with two
|
||||
/// disjoint keys — and old and new nodes would stop excluding each other on
|
||||
/// the very writes that are serialized today.
|
||||
fn bucket_metadata_transaction_lock_key(bucket: &str) -> String {
|
||||
pub(crate) fn bucket_metadata_transaction_lock_key(bucket: &str) -> String {
|
||||
format!("bucket-targets/{bucket}/transaction.lock")
|
||||
}
|
||||
|
||||
|
||||
@@ -1099,6 +1099,43 @@ async fn load_current_ledger_locked(
|
||||
}
|
||||
}
|
||||
|
||||
/// Remove the single-ledger reservation state for an incarnation after its
|
||||
/// durable bucket retirement has been published. The caller holds the bucket
|
||||
/// lifecycle and metadata transaction write fences, so no live reservation
|
||||
/// owner can outlast this cleanup. Keep the ledger namespace lock as the local
|
||||
/// serialization boundary and delete only the exact V1 object key.
|
||||
pub(crate) async fn cleanup_retired_bucket_ledger(store: Arc<ECStore>, bucket: &str, incarnation: Uuid) -> Result<()> {
|
||||
let object = ledger_object(bucket);
|
||||
let ledger_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &object).await?;
|
||||
let guard = ledger_lock.get_write_lock(get_lock_acquire_timeout()).await?;
|
||||
match load_ledger_locked(Arc::clone(&store), &object).await {
|
||||
Ok(ledger) if ledger.bucket_incarnation == incarnation => {}
|
||||
Ok(ledger) if crate::bucket::retirement::is_retired(store.clone(), bucket, ledger.bucket_incarnation).await? => {}
|
||||
Ok(_) => return Err(StorageError::PreconditionFailed),
|
||||
Err(StorageError::ConfigNotFound) => return Ok(()),
|
||||
Err(error) => return Err(error),
|
||||
}
|
||||
if guard.is_lock_lost() {
|
||||
return Err(StorageError::NamespaceLockQuorumUnavailable {
|
||||
mode: "quota_ledger_retirement",
|
||||
bucket: RUSTFS_META_BUCKET.to_string(),
|
||||
object,
|
||||
required: 1,
|
||||
achieved: 0,
|
||||
});
|
||||
}
|
||||
let mut options = ObjectOptions {
|
||||
no_lock: true,
|
||||
..Default::default()
|
||||
};
|
||||
options.add_namespace_lock_guard(&guard);
|
||||
match store.delete_object(RUSTFS_META_BUCKET, &object, options).await {
|
||||
Ok(_) => Ok(()),
|
||||
Err(error) if is_err_object_not_found(&error) => Ok(()),
|
||||
Err(error) => Err(error),
|
||||
}
|
||||
}
|
||||
|
||||
async fn reap_stale_reservations(store: Arc<ECStore>, bucket: &str, ledger_object: &str) -> Result<()> {
|
||||
let now = now_unix();
|
||||
let (candidates, reconcile_required, next_cursor) = {
|
||||
@@ -1921,6 +1958,8 @@ fn log_deferred_settlement(data: &ReservationLedgerData, state: &'static str, er
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
|
||||
use serial_test::serial;
|
||||
|
||||
fn ledger(accounted_usage: u64) -> QuotaLedger {
|
||||
QuotaLedger::new(Uuid::new_v4(), OffsetDateTime::now_utc(), accounted_usage)
|
||||
@@ -1973,6 +2012,67 @@ mod tests {
|
||||
assert!(matches!(ledger.admitted_usage(), Err(StorageError::PartMissingOrCorrupt)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn retired_bucket_cleanup_removes_stale_v1_reservation_before_new_admission() {
|
||||
let (_directories, store) = metadata_sys::test_support::isolated_store_over_temp_disks().await;
|
||||
metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
|
||||
let bucket = format!("quota-orphan-recovery-{}", Uuid::new_v4().simple());
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create old bucket generation");
|
||||
let old_incarnation = store
|
||||
.bucket_incarnation_id_from_disk(&bucket)
|
||||
.await
|
||||
.expect("old bucket incarnation");
|
||||
let new_incarnation = Uuid::new_v4();
|
||||
let ledger_path = ledger_object(&bucket);
|
||||
let mut stale = QuotaLedger::new(old_incarnation, OffsetDateTime::now_utc(), 0);
|
||||
stale
|
||||
.reserve(
|
||||
Uuid::new_v4(),
|
||||
PersistedReservation {
|
||||
object: "crashed-put".to_string(),
|
||||
old_size: 0,
|
||||
new_size: 100,
|
||||
created_at: now_unix(),
|
||||
pool_index: Some(0),
|
||||
set_index: Some(0),
|
||||
commit_started: false,
|
||||
},
|
||||
)
|
||||
.expect("crash-left reservation is structurally valid");
|
||||
crate::config::com::save_config(Arc::clone(&store), &ledger_path, serde_json::to_vec(&stale).expect("encode ledger"))
|
||||
.await
|
||||
.expect("persist crash-left reservation fixture");
|
||||
|
||||
assert!(matches!(
|
||||
load_current_ledger_locked(Arc::clone(&store), &bucket, &ledger_path, new_incarnation, OffsetDateTime::now_utc())
|
||||
.await,
|
||||
Err(StorageError::PartMissingOrCorrupt)
|
||||
));
|
||||
assert!(matches!(
|
||||
cleanup_retired_bucket_ledger(Arc::clone(&store), &bucket, Uuid::new_v4()).await,
|
||||
Err(StorageError::PreconditionFailed)
|
||||
));
|
||||
crate::bucket::retirement::commit_retirement(Arc::clone(&store), &bucket, old_incarnation, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("publish predecessor generation retirement proof");
|
||||
cleanup_retired_bucket_ledger(Arc::clone(&store), &bucket, new_incarnation)
|
||||
.await
|
||||
.expect("a proven retired predecessor ledger cannot block the replacement generation");
|
||||
assert!(matches!(
|
||||
load_ledger_locked(Arc::clone(&store), &ledger_path).await,
|
||||
Err(StorageError::ConfigNotFound)
|
||||
));
|
||||
let replacement =
|
||||
load_current_ledger_locked(Arc::clone(&store), &bucket, &ledger_path, new_incarnation, OffsetDateTime::now_utc())
|
||||
.await
|
||||
.expect("new quota admission must not inherit a fresh stale reservation");
|
||||
assert!(replacement.reservations.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ledger_accounts_overwrite_delta_and_reserved_growth() {
|
||||
let mut ledger = ledger(10);
|
||||
|
||||
@@ -12,6 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use rustfs_io_metrics::record_object_lock_acquire_failure;
|
||||
use rustfs_lock::{LockError, NamespaceLockGuard, NamespaceLockWrapper};
|
||||
use std::time::{Duration, Instant};
|
||||
use tracing::Span;
|
||||
@@ -21,6 +22,26 @@ const EVENT_OBJECT_NAMESPACE_LOCK: &str = "object_namespace_lock";
|
||||
const LOG_COMPONENT_ECSTORE: &str = "ecstore";
|
||||
const LOG_SUBSYSTEM_OBJECT_LOCK: &str = "object_lock";
|
||||
|
||||
pub(crate) fn record_lock_acquire_failure(op: &'static str, bucket: &str, object: &str, mode: &'static str, error: &LockError) {
|
||||
let resource_class = if bucket == crate::disk::RUSTFS_META_MULTIPART_BUCKET {
|
||||
"multipart_metadata"
|
||||
} else if bucket == crate::disk::RUSTFS_META_BUCKET || bucket == ".minio.sys" {
|
||||
"system_metadata"
|
||||
} else if object.is_empty() || object == bucket {
|
||||
"bucket_namespace"
|
||||
} else {
|
||||
"object_namespace"
|
||||
};
|
||||
let reason = match error {
|
||||
LockError::Timeout { .. } => "timeout",
|
||||
LockError::AlreadyLocked { .. } => "contention",
|
||||
LockError::QuorumNotReached { .. } | LockError::InsufficientNodes { .. } => "quorum",
|
||||
LockError::Network { .. } | LockError::Internal { .. } => "transport",
|
||||
_ => "other",
|
||||
};
|
||||
record_object_lock_acquire_failure(op, mode, resource_class, reason);
|
||||
}
|
||||
|
||||
/// Diagnostic lifetime only; it never owns or changes a namespace lock.
|
||||
/// Keep it alongside the real guard when ownership moves to a commit tail.
|
||||
#[derive(Default)]
|
||||
|
||||
@@ -62,7 +62,7 @@ use crate::diagnostics::get::{
|
||||
classify_storage_error, get_stage_timer_if_enabled, record_get_object_pipeline_failure,
|
||||
record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled,
|
||||
};
|
||||
use crate::diagnostics::object_lock::ObjectLockAttempt;
|
||||
use crate::diagnostics::object_lock::{ObjectLockAttempt, record_lock_acquire_failure};
|
||||
use crate::disk::error_reduce::{
|
||||
BUCKET_OP_IGNORED_ERRS, OBJECT_OP_IGNORED_ERRS, build_write_quorum_failure_summary, count_errs, reduce_read_quorum_errs,
|
||||
reduce_write_quorum_errs,
|
||||
@@ -4715,6 +4715,9 @@ impl SetDisks {
|
||||
let timeout = get_lock_acquire_timeout();
|
||||
let mut attempt = ObjectLockAttempt::start(op, bucket, object, None, &ns_lock, "read", timeout);
|
||||
let result = ns_lock.get_read_lock(timeout).await;
|
||||
if let Err(error) = &result {
|
||||
record_lock_acquire_failure(op, bucket, object, "read", error);
|
||||
}
|
||||
attempt.observe(&result);
|
||||
let guard = result.map_err(|e| self.map_namespace_lock_error(bucket, object, "read", e))?;
|
||||
let owner = diag_enabled.then(|| ns_lock.owner().to_string());
|
||||
@@ -4773,6 +4776,9 @@ impl SetDisks {
|
||||
let acquire_timeout = get_put_object_commit_lock_acquire_timeout(op);
|
||||
let mut attempt = ObjectLockAttempt::start(op, bucket, object, None, &ns_lock, "write", acquire_timeout);
|
||||
let result = ns_lock.get_write_lock(acquire_timeout).await;
|
||||
if let Err(error) = &result {
|
||||
record_lock_acquire_failure(op, bucket, object, "write", error);
|
||||
}
|
||||
attempt.observe(&result);
|
||||
let guard = resolve_put_object_commit_lock_acquire_result(self, op, bucket, object, result)?;
|
||||
Self::record_put_object_commit_namespace_lock_wait(op, acquire_start);
|
||||
@@ -4825,6 +4831,9 @@ impl SetDisks {
|
||||
std::task::Poll::Ready(result) => std::task::Poll::Ready(result),
|
||||
})
|
||||
.await;
|
||||
if let Err(error) = &result {
|
||||
record_lock_acquire_failure(op, bucket, object, "write", error);
|
||||
}
|
||||
attempt.observe(&result);
|
||||
let guard = resolve_put_object_commit_lock_acquire_result(self, op, bucket, object, result)?;
|
||||
Self::record_put_object_commit_namespace_lock_wait(op, acquire_start);
|
||||
|
||||
@@ -34,6 +34,41 @@ pub(crate) enum BucketInfoQuorum {
|
||||
}
|
||||
|
||||
impl SetDisks {
|
||||
/// Require proof from every configured disk before retiring a bucket
|
||||
/// generation whose visible volume was removed outside the storage API.
|
||||
pub(crate) async fn ensure_bucket_volume_absent_on_every_disk(&self, bucket: &str) -> Result<()> {
|
||||
let disks = self.disk_inventory().await;
|
||||
if disks.is_empty() {
|
||||
return Err(Error::ErasureReadQuorum);
|
||||
}
|
||||
|
||||
let futures = disks.into_iter().map(|disk| async move {
|
||||
match disk {
|
||||
Some(disk) => disk.stat_volume(bucket).await.map(|_| ()),
|
||||
None => Err(DiskError::DiskNotFound),
|
||||
}
|
||||
});
|
||||
let results = join_all(futures).await;
|
||||
let mut volume_present = false;
|
||||
let mut first_unknown = None;
|
||||
for result in results {
|
||||
match result {
|
||||
Ok(()) => volume_present = true,
|
||||
Err(DiskError::VolumeNotFound) => {}
|
||||
Err(error) if first_unknown.is_none() => first_unknown = Some(error),
|
||||
Err(_) => {}
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(error) = first_unknown {
|
||||
return Err(error.into());
|
||||
}
|
||||
if volume_present {
|
||||
return Err(Error::BucketExists(bucket.to_owned()));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) async fn stat_bucket_with_quorum(&self, bucket: &str, quorum: BucketInfoQuorum) -> Result<BucketInfo> {
|
||||
let disks = self.disk_inventory().await;
|
||||
let disk_count = disks.len();
|
||||
|
||||
@@ -204,6 +204,52 @@ where
|
||||
.await
|
||||
}
|
||||
|
||||
fn bucket_recovery_lock_unavailable(bucket: &str, object: &str) -> StorageError {
|
||||
StorageError::NamespaceLockQuorumUnavailable {
|
||||
mode: "write",
|
||||
bucket: bucket.to_string(),
|
||||
object: object.to_string(),
|
||||
required: 1,
|
||||
achieved: 0,
|
||||
}
|
||||
}
|
||||
|
||||
async fn await_orphan_bucket_recovery<T, F>(
|
||||
bucket: &str,
|
||||
publication_guard: &rustfs_lock::NamespaceLockGuard,
|
||||
lifecycle_guard: &rustfs_lock::NamespaceLockGuard,
|
||||
metadata_guard: &rustfs_lock::NamespaceLockGuard,
|
||||
namespace_guard: &rustfs_lock::NamespaceLockGuard,
|
||||
future: F,
|
||||
) -> Result<T>
|
||||
where
|
||||
F: Future<Output = Result<T>>,
|
||||
{
|
||||
let metadata_lock_object = crate::bucket::metadata_sys::bucket_metadata_transaction_lock_key(bucket);
|
||||
for ((lock_bucket, lock_object), guard) in [
|
||||
(
|
||||
(bucket, rustfs_common::table_catalog::TABLE_BUCKET_PUBLICATION_LOCK_PATH),
|
||||
publication_guard,
|
||||
),
|
||||
((bucket, BUCKET_LIFECYCLE_LOCK_OBJECT), lifecycle_guard),
|
||||
((RUSTFS_META_BUCKET, metadata_lock_object.as_str()), metadata_guard),
|
||||
((bucket, bucket), namespace_guard),
|
||||
] {
|
||||
if guard.is_lock_lost() {
|
||||
return Err(bucket_recovery_lock_unavailable(lock_bucket, lock_object));
|
||||
}
|
||||
}
|
||||
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = publication_guard.lock_lost_notified() => Err(bucket_recovery_lock_unavailable(bucket, rustfs_common::table_catalog::TABLE_BUCKET_PUBLICATION_LOCK_PATH)),
|
||||
_ = lifecycle_guard.lock_lost_notified() => Err(bucket_recovery_lock_unavailable(bucket, BUCKET_LIFECYCLE_LOCK_OBJECT)),
|
||||
_ = metadata_guard.lock_lost_notified() => Err(bucket_recovery_lock_unavailable(RUSTFS_META_BUCKET, &metadata_lock_object)),
|
||||
_ = namespace_guard.lock_lost_notified() => Err(bucket_recovery_lock_unavailable(bucket, bucket)),
|
||||
result = future => result,
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_bucket_usage_cleanup<F>(guard: Option<&rustfs_lock::NamespaceLockGuard>, bucket: &str, future: F) -> Result<()>
|
||||
where
|
||||
F: Future<Output = Result<()>>,
|
||||
@@ -272,6 +318,20 @@ async fn bucket_delete_local_blocker(
|
||||
}
|
||||
|
||||
impl ECStore {
|
||||
/// Run a multi-step storage mutation on the instance task tracker so its
|
||||
/// caller may stop waiting without releasing guards before the mutation
|
||||
/// finishes.
|
||||
pub async fn run_detached_mutation<F>(&self, mutation: F) -> Result<F::Output>
|
||||
where
|
||||
F: Future + Send + 'static,
|
||||
F::Output: Send + 'static,
|
||||
{
|
||||
self.ctx
|
||||
.run_detached_mutation(mutation)
|
||||
.await
|
||||
.map_err(|error| StorageError::other_with_context("detached ECStore mutation failed", error))
|
||||
}
|
||||
|
||||
fn bucket_sets(&self) -> impl Iterator<Item = (usize, usize, Arc<crate::set_disk::SetDisks>)> + '_ {
|
||||
self.pools.iter().flat_map(|pool| {
|
||||
pool.disk_set
|
||||
@@ -280,6 +340,113 @@ impl ECStore {
|
||||
})
|
||||
}
|
||||
|
||||
async fn ensure_bucket_volume_absent_on_all_sets(&self, bucket: &str) -> Result<()> {
|
||||
let sets: Vec<_> = self.bucket_sets().map(|(_, _, set)| set).collect();
|
||||
if sets.is_empty() {
|
||||
return Err(StorageError::ErasureReadQuorum);
|
||||
}
|
||||
let results = join_all(sets.iter().map(|set| set.ensure_bucket_volume_absent_on_every_disk(bucket))).await;
|
||||
for result in results {
|
||||
result?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Return whether an old bucket generation has a durable retirement record.
|
||||
/// A live matching incarnation is never considered retired, even if a stale
|
||||
/// record exists from a previous generation.
|
||||
pub async fn is_bucket_incarnation_retired(&self, bucket: &str, expected: Uuid) -> Result<bool> {
|
||||
if expected.is_nil() {
|
||||
return Ok(false);
|
||||
}
|
||||
let retirement_store = metadata_sys::object_store_if_initialized_in(&self.ctx).await.ok_or_else(|| {
|
||||
StorageError::InsufficientReadQuorum(RUSTFS_META_BUCKET.to_string(), "bucket-retirements".to_string())
|
||||
})?;
|
||||
if crate::bucket::metadata::load_bucket_incarnation(retirement_store.clone(), bucket).await? == Some(expected) {
|
||||
return Ok(false);
|
||||
}
|
||||
crate::bucket::retirement::is_retired(retirement_store, bucket, expected).await
|
||||
}
|
||||
|
||||
/// Reconcile bucket metadata after an operator removed all physical volume
|
||||
/// directories. The expected generation, all-disk absence proof and durable
|
||||
/// retirement record make retries safe after partial metadata cleanup.
|
||||
pub fn recover_orphaned_bucket<'a>(&'a self, bucket: &'a str, expected: Uuid) -> futures::future::BoxFuture<'a, Result<()>> {
|
||||
Box::pin(async move { Box::pin(self.recover_orphaned_bucket_inner(bucket, expected)).await })
|
||||
}
|
||||
|
||||
async fn recover_orphaned_bucket_inner(&self, bucket: &str, expected: Uuid) -> Result<()> {
|
||||
if let Err(error) = check_valid_bucket_name_strict(bucket) {
|
||||
return Err(StorageError::BucketNameInvalid(error.to_string()));
|
||||
}
|
||||
if is_meta_bucketname(bucket) || expected.is_nil() {
|
||||
return Err(StorageError::InvalidArgument(
|
||||
"RecoverOrphanedBucket".to_string(),
|
||||
"bucket".to_string(),
|
||||
"a user bucket and non-nil incarnation are required".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
// Lock order matches bucket deletion: publication -> lifecycle ->
|
||||
// metadata transaction -> exact namespace.
|
||||
let publication_guard = self.acquire_bucket_publication_write_lock(bucket).await?;
|
||||
let lifecycle_guard = self.acquire_bucket_lifecycle_write_lock(bucket).await?;
|
||||
let metadata_guard = metadata_sys::acquire_bucket_metadata_transaction_lock_in(&self.ctx, bucket).await?;
|
||||
let ns_lock = self.new_ns_lock(bucket, bucket).await?;
|
||||
let ns_guard = ns_lock
|
||||
.get_write_lock(get_lock_acquire_timeout())
|
||||
.await
|
||||
.map_err(|error| match error {
|
||||
rustfs_lock::LockError::QuorumNotReached { required, achieved } => StorageError::NamespaceLockQuorumUnavailable {
|
||||
mode: "write",
|
||||
bucket: bucket.to_string(),
|
||||
object: bucket.to_string(),
|
||||
required,
|
||||
achieved,
|
||||
},
|
||||
other => StorageError::Lock(other),
|
||||
})?;
|
||||
|
||||
let recover = Box::pin(async {
|
||||
self.ensure_bucket_volume_absent_on_all_sets(bucket).await?;
|
||||
let retirement_store = metadata_sys::object_store_if_initialized_in(&self.ctx).await.ok_or_else(|| {
|
||||
StorageError::InsufficientReadQuorum(RUSTFS_META_BUCKET.to_string(), "bucket-retirements".to_string())
|
||||
})?;
|
||||
let current_incarnation = crate::bucket::metadata::load_bucket_incarnation(retirement_store.clone(), bucket).await?;
|
||||
if current_incarnation.is_some_and(|current| current != expected) {
|
||||
return Err(StorageError::PreconditionFailed);
|
||||
}
|
||||
|
||||
let retired = crate::bucket::retirement::is_retired(retirement_store.clone(), bucket, expected).await?;
|
||||
if current_incarnation.is_none() && !retired {
|
||||
return Err(StorageError::PreconditionFailed);
|
||||
}
|
||||
if !retired {
|
||||
let mut record_opts = ObjectOptions {
|
||||
max_parity: true,
|
||||
..Default::default()
|
||||
};
|
||||
record_opts.add_bucket_lifecycle_lock_guard(&lifecycle_guard);
|
||||
record_opts.add_namespace_lock_guard(&ns_guard);
|
||||
let publish_retirement = Box::pin(crate::bucket::retirement::commit_retirement(
|
||||
retirement_store.clone(),
|
||||
bucket,
|
||||
expected,
|
||||
&record_opts,
|
||||
));
|
||||
publish_retirement.await?;
|
||||
}
|
||||
|
||||
crate::bucket::quota::reservation::cleanup_retired_bucket_ledger(retirement_store.clone(), bucket, expected).await?;
|
||||
crate::data_usage::prepare_bucket_usage_for_namespace_change(bucket, Some(&ns_guard)).await?;
|
||||
crate::data_usage::remove_bucket_usage_from_backend_with_guard_fenced(self, bucket, Some(&ns_guard)).await?;
|
||||
self.cleanup_deleted_bucket_metadata_strict(bucket, true).await?;
|
||||
crate::store::list_objects::observe_scanner_namespace_mutations(bucket, 1);
|
||||
Ok(())
|
||||
});
|
||||
await_orphan_bucket_recovery(bucket, &publication_guard, &lifecycle_guard, &metadata_guard, &ns_guard, recover).await
|
||||
}
|
||||
|
||||
pub async fn get_bucket_metadata(&self, bucket: &str) -> Result<Arc<BucketMetadata>> {
|
||||
let sys = metadata_sys::require_bucket_metadata_sys_in(&self.ctx)?;
|
||||
sys.read().await.get(bucket).await
|
||||
@@ -604,6 +771,25 @@ impl ECStore {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn cleanup_deleted_bucket_metadata_strict(&self, bucket: &str, include_deleted_marker: bool) -> Result<()> {
|
||||
let options = ObjectOptions {
|
||||
delete_prefix_object: true,
|
||||
..Default::default()
|
||||
};
|
||||
for prefix in bucket_delete_metadata_cleanup_prefixes(bucket) {
|
||||
self.delete_prefix(RUSTFS_META_BUCKET, &prefix, &options).await?;
|
||||
}
|
||||
|
||||
if include_deleted_marker {
|
||||
let marker_prefix = bucket_deleted_marker_prefix(bucket);
|
||||
self.delete_prefix(RUSTFS_META_BUCKET, &marker_prefix, &options).await?;
|
||||
}
|
||||
|
||||
metadata_sys::remove_bucket_metadata_in(&self.ctx, bucket).await?;
|
||||
runtime_sources::delete_bucket_monitor_entry(bucket);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn cleanup_bucket_usage(&self, bucket: &str, guard: Option<&rustfs_lock::NamespaceLockGuard>) -> Result<()> {
|
||||
run_bucket_usage_cleanup(guard, bucket, async {
|
||||
crate::data_usage::prepare_bucket_usage_for_namespace_change(bucket, guard).await?;
|
||||
@@ -715,6 +901,32 @@ impl ECStore {
|
||||
}
|
||||
};
|
||||
let confirmed_missing = existing_bucket_info.is_none();
|
||||
if confirmed_missing
|
||||
&& !opts.no_lock
|
||||
&& !is_meta_bucketname(bucket)
|
||||
&& let Some(metadata_store) = metadata_sys::object_store_if_initialized_in(&self.ctx).await
|
||||
&& let Some(old_incarnation) = crate::bucket::metadata::load_bucket_incarnation(metadata_store, bucket).await?
|
||||
{
|
||||
// A quorum-level absence is not proof that every disk lost the old
|
||||
// generation. Require the explicit recovery path to verify all disks
|
||||
// and retire this incarnation before a same-name create can mutate
|
||||
// any volume.
|
||||
self.ensure_bucket_volume_absent_on_all_sets(bucket)
|
||||
.await
|
||||
.map_err(|error| match error {
|
||||
StorageError::BucketExists(_) => StorageError::InvalidArgument(
|
||||
"CreateBucket".to_string(),
|
||||
"bucket".to_string(),
|
||||
"an old bucket volume remains on disk; administrator recovery refused".to_string(),
|
||||
),
|
||||
other => other,
|
||||
})?;
|
||||
return Err(StorageError::InvalidArgument(
|
||||
"CreateBucket".to_string(),
|
||||
"bucket".to_string(),
|
||||
format!("orphaned bucket generation {old_incarnation} requires administrator recovery before recreation"),
|
||||
));
|
||||
}
|
||||
let existing_metadata = if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) {
|
||||
let (mut metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
|
||||
if !persisted {
|
||||
@@ -1610,6 +1822,65 @@ mod tests {
|
||||
setup_bucket_quorum_test_env(&[4, 4], None).await
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
#[serial]
|
||||
async fn orphaned_bucket_recovery_requires_every_volume_absent_and_is_retryable() {
|
||||
let (temp_dir, store) = setup_multi_pool_bucket_test_env().await;
|
||||
metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
|
||||
let bucket = format!("orphan-recovery-{}", Uuid::new_v4().simple());
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create bucket before simulated operator removal");
|
||||
let incarnation = store
|
||||
.bucket_incarnation_id_from_disk(&bucket)
|
||||
.await
|
||||
.expect("persisted bucket generation");
|
||||
|
||||
let first_disk_volume = temp_dir.path().join("pool0-disk0").join(&bucket);
|
||||
tokio::fs::remove_dir_all(&first_disk_volume)
|
||||
.await
|
||||
.expect("simulate operator removal on one disk");
|
||||
let blocked = store
|
||||
.recover_orphaned_bucket(&bucket, incarnation)
|
||||
.await
|
||||
.expect_err("recovery must refuse while any configured disk still has the volume");
|
||||
assert!(matches!(blocked, StorageError::BucketExists(_)), "unexpected error: {blocked}");
|
||||
assert_eq!(
|
||||
crate::bucket::metadata::load_bucket_incarnation(store.clone(), &bucket)
|
||||
.await
|
||||
.expect("generation sidecar remains readable"),
|
||||
Some(incarnation),
|
||||
"a refused recovery must preserve the old generation identity"
|
||||
);
|
||||
|
||||
for disk_index in 1..4 {
|
||||
tokio::fs::remove_dir_all(temp_dir.path().join(format!("pool0-disk{disk_index}")).join(&bucket))
|
||||
.await
|
||||
.expect("remove remaining volume from first set");
|
||||
}
|
||||
for disk_index in 0..4 {
|
||||
tokio::fs::remove_dir_all(temp_dir.path().join(format!("pool1-disk{disk_index}")).join(&bucket))
|
||||
.await
|
||||
.expect("remove volume from second set");
|
||||
}
|
||||
|
||||
store
|
||||
.recover_orphaned_bucket(&bucket, incarnation)
|
||||
.await
|
||||
.expect("recovery should retire the exact generation and clean stale metadata");
|
||||
assert!(
|
||||
store
|
||||
.is_bucket_incarnation_retired(&bucket, incarnation)
|
||||
.await
|
||||
.expect("retirement proof remains readable")
|
||||
);
|
||||
store
|
||||
.recover_orphaned_bucket(&bucket, incarnation)
|
||||
.await
|
||||
.expect("retry after metadata cleanup should converge from durable retirement proof");
|
||||
}
|
||||
|
||||
async fn setup_bucket_quorum_test_env(
|
||||
drives_per_pool: &[usize],
|
||||
standard_parity: Option<usize>,
|
||||
|
||||
@@ -45,6 +45,7 @@ use tokio::{
|
||||
};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{debug, error, info, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::{DiskError, Endpoint, HealDiskExt as _, local_disk_map_read};
|
||||
|
||||
@@ -1903,6 +1904,10 @@ impl HealManager {
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn mrf_bucket_incarnation_retired(&self, bucket: &str, expected: Uuid) -> Result<bool> {
|
||||
self.storage.mrf_bucket_incarnation_retired(bucket, expected).await
|
||||
}
|
||||
|
||||
async fn submit_heal_request_with_receipt_alias_and_mrf_notice(
|
||||
&self,
|
||||
mut request: HealRequest,
|
||||
|
||||
@@ -1366,8 +1366,10 @@ impl MrfRuntime {
|
||||
rustfs_common::mrf_channel::consume_recorded_verified_mrf_repair_events_for(bucket.as_ref(), &mut anchors);
|
||||
}
|
||||
let remaining: HashSet<_> = anchors.into_iter().collect();
|
||||
let previous_replay_anchor_count = self.durable_replay_anchors.len();
|
||||
self.durable_replay_anchors.retain(|anchor| remaining.contains(anchor));
|
||||
self.dirty |= self.partial_writes.retain_unproven(&remaining);
|
||||
let replay_anchors_changed = self.durable_replay_anchors.len() != previous_replay_anchor_count;
|
||||
self.dirty |= replay_anchors_changed || self.partial_writes.retain_unproven(&remaining);
|
||||
for bucket in buckets {
|
||||
for event in rustfs_common::mrf_channel::take_mrf_unverified_legacy_events_for(bucket.as_ref()) {
|
||||
if self.partial_writes.park_unverified_legacy(&event.anchor) {
|
||||
@@ -1944,6 +1946,7 @@ async fn run_mrf_consumer(
|
||||
}
|
||||
}
|
||||
_ = flush_tick.tick() => {
|
||||
runtime.partial_writes.note_retired_generation_proofs(manager.as_ref()).await;
|
||||
runtime.discharge_durable_replay_anchors();
|
||||
match tick_action(
|
||||
runtime.dirty,
|
||||
|
||||
@@ -16,10 +16,14 @@ use super::{
|
||||
HealManager, MrfConsumerConfig, MrfDurableRepairAnchor, MrfIntent, MrfLegacyRiskAcceptanceRequest, MrfQueueKey, queue_key,
|
||||
submit_mrf_heal_request,
|
||||
};
|
||||
use rustfs_common::mrf_channel::{MrfDurableAdmissionError, MrfIngressResult, release_mrf_intent, try_rearm_mrf_replay_intent};
|
||||
use rustfs_common::mrf_channel::{
|
||||
MrfDurableAdmissionError, MrfIngressResult, MrfVerifiedRepairDisposition, MrfVerifiedRepairEvent, release_mrf_intent,
|
||||
try_rearm_mrf_replay_intent,
|
||||
};
|
||||
use rustfs_heal_contracts::heal_channel::HealAdmissionResult;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
use tokio::time::Instant;
|
||||
use uuid::Uuid;
|
||||
@@ -29,6 +33,7 @@ const MAX_OPERATOR_ACTOR_BYTES: usize = 256;
|
||||
const MAX_OPERATOR_REFERENCE_BYTES: usize = 256;
|
||||
const LIFECYCLE_CHECKPOINT_RECORD_OVERHEAD_BYTES: usize = 256;
|
||||
const MAX_UNVERIFIED_RETRY_BACKOFF: Duration = Duration::from_secs(60);
|
||||
const MRF_RETIREMENT_PROOF_RECHECK_INTERVAL: Duration = Duration::from_secs(5);
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
|
||||
struct PartialWriteKey {
|
||||
@@ -229,6 +234,7 @@ pub(super) struct PartialWrites {
|
||||
entries: HashMap<PartialWriteKey, Responsibility>,
|
||||
retry_order: VecDeque<PartialWriteKey>,
|
||||
retry_index: HashSet<PartialWriteKey>,
|
||||
retirement_checks: HashMap<(Arc<str>, Uuid), (Instant, bool)>,
|
||||
bytes: usize,
|
||||
}
|
||||
|
||||
@@ -335,6 +341,66 @@ impl PartialWrites {
|
||||
self.entries.values().filter_map(|entry| entry.anchor.as_ref())
|
||||
}
|
||||
|
||||
pub(super) async fn note_retired_generation_proofs(&mut self, manager: &HealManager) {
|
||||
let mut anchors_by_generation: HashMap<(Arc<str>, Uuid), Vec<MrfDurableRepairAnchor>> = HashMap::new();
|
||||
for entry in self.entries.values() {
|
||||
if !entry.persisted {
|
||||
continue;
|
||||
}
|
||||
let Some(source_incarnation) = entry.source_bucket_incarnation_id else {
|
||||
continue;
|
||||
};
|
||||
let Some(anchor) = MrfDurableRepairAnchor::from_intent(&entry.intent, source_incarnation)
|
||||
.filter(|anchor| anchor.bucket_incarnation_id == source_incarnation)
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
anchors_by_generation
|
||||
.entry((anchor.bucket.clone(), source_incarnation))
|
||||
.or_default()
|
||||
.push(anchor);
|
||||
}
|
||||
|
||||
self.retirement_checks
|
||||
.retain(|generation, _| anchors_by_generation.contains_key(generation));
|
||||
let now = Instant::now();
|
||||
for ((bucket, incarnation), anchors) in anchors_by_generation {
|
||||
let retired = match self.retirement_checks.get(&(bucket.clone(), incarnation)) {
|
||||
Some((_, true)) => true,
|
||||
Some((checked_at, false))
|
||||
if now.saturating_duration_since(*checked_at) < MRF_RETIREMENT_PROOF_RECHECK_INTERVAL =>
|
||||
{
|
||||
false
|
||||
}
|
||||
_ => {
|
||||
let retired = manager
|
||||
.mrf_bucket_incarnation_retired(bucket.as_ref(), incarnation)
|
||||
.await
|
||||
.unwrap_or(false);
|
||||
self.retirement_checks.insert((bucket.clone(), incarnation), (now, retired));
|
||||
retired
|
||||
}
|
||||
};
|
||||
if !retired {
|
||||
continue;
|
||||
}
|
||||
|
||||
for anchor in anchors {
|
||||
rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent {
|
||||
kind: anchor.kind,
|
||||
bucket: anchor.bucket.clone(),
|
||||
object: anchor.object.clone(),
|
||||
version_id: anchor.version_id,
|
||||
scope: anchor.scope,
|
||||
delete_marker_purge: anchor.delete_marker_purge.clone(),
|
||||
lease: Some(anchor.lease),
|
||||
bucket_incarnation_id: anchor.bucket_incarnation_id,
|
||||
disposition: MrfVerifiedRepairDisposition::AuthoritativelyAbsent,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn park_unverified_legacy(&mut self, anchor: &MrfDurableRepairAnchor) -> bool {
|
||||
let key = PartialWriteKey::from_anchor(anchor);
|
||||
let Some(entry) = self.entries.get_mut(&key) else {
|
||||
@@ -945,6 +1011,7 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::heal::mrf_queue::MrfLegacyRiskAcceptanceRequest;
|
||||
use crate::heal::storage::{ECStoreHealStorage, HealStorageAPI};
|
||||
use crate::heal::storage_api::storage::{BucketOperations as _, DeleteBucketOptions};
|
||||
use rustfs_common::mrf_channel::{MrfKind, MrfScope, MrfVerifiedRepairDisposition, MrfVerifiedRepairEvent};
|
||||
use serial_test::serial;
|
||||
use std::sync::Arc;
|
||||
@@ -1861,6 +1928,94 @@ mod tests {
|
||||
assert!(runtime.partial_writes.intent_for_responsibility(current_id).is_none());
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
#[serial]
|
||||
async fn durable_mrf_anchor_retires_only_after_exact_bucket_generation_is_deleted() {
|
||||
use super::super::{MrfQueue, MrfRuntime};
|
||||
|
||||
let env = rustfs_test_utils::TestECStoreEnv::builder()
|
||||
.prefix("rustfs_mrf_retirement_proof")
|
||||
.build()
|
||||
.await;
|
||||
let bucket = "mrf-retired-generation";
|
||||
env.make_bucket(bucket, false).await;
|
||||
let incarnation = env
|
||||
.ecstore
|
||||
.bucket_incarnation_id_from_disk(bucket)
|
||||
.await
|
||||
.expect("current generation identity");
|
||||
let storage: Arc<dyn HealStorageAPI> = Arc::new(ECStoreHealStorage::new(env.ecstore.clone()));
|
||||
let manager = Arc::new(HealManager::new_without_root_recovery_for_test(storage, None));
|
||||
|
||||
let mut item = intent("durable-object");
|
||||
item.bucket = Arc::from(bucket);
|
||||
let mut partial_writes = PartialWrites::default();
|
||||
partial_writes
|
||||
.admit_with_source_incarnation(item.clone(), Some(incarnation), 1, 8192)
|
||||
.expect("retain exact old-generation responsibility");
|
||||
partial_writes.mark_persisted();
|
||||
let key = PartialWriteKey::new(&item, Some(incarnation));
|
||||
let anchor = MrfDurableRepairAnchor::from_intent(&item, incarnation).expect("exact retirement anchor");
|
||||
partial_writes.entries.get_mut(&key).expect("durable responsibility").anchor = Some(anchor);
|
||||
|
||||
assert!(
|
||||
!manager
|
||||
.mrf_bucket_incarnation_retired(bucket, incarnation)
|
||||
.await
|
||||
.expect("live generation is queryable"),
|
||||
"a live generation cannot retire its MRF anchors"
|
||||
);
|
||||
env.ecstore
|
||||
.delete_bucket(bucket, &DeleteBucketOptions::default())
|
||||
.await
|
||||
.expect("normal bucket deletion publishes durable retirement evidence");
|
||||
|
||||
partial_writes.note_retired_generation_proofs(manager.as_ref()).await;
|
||||
let config = MrfConsumerConfig::default();
|
||||
let mut runtime = MrfRuntime {
|
||||
partial_writes,
|
||||
queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes),
|
||||
config,
|
||||
checkpoint_owner: Uuid::new_v4(),
|
||||
next_checkpoint_sequence: 1,
|
||||
new_since_flush: 0,
|
||||
dirty: false,
|
||||
journal_on_disk: true,
|
||||
retain_replay_journal: false,
|
||||
durable_replay_anchors: Vec::new(),
|
||||
replay_cleanup: None,
|
||||
runtime_checkpoint: None,
|
||||
backoff_until: None,
|
||||
};
|
||||
runtime.discharge_durable_replay_anchors();
|
||||
assert_eq!(
|
||||
runtime.partial_writes.depth(),
|
||||
0,
|
||||
"retirement proof should discharge the exact old anchor"
|
||||
);
|
||||
assert!(runtime.dirty, "discharging durable responsibility must publish a successor checkpoint");
|
||||
|
||||
let manually_removed_bucket = "mrf-manually-removed-volume";
|
||||
env.make_bucket(manually_removed_bucket, false).await;
|
||||
let manual_incarnation = env
|
||||
.ecstore
|
||||
.bucket_incarnation_id_from_disk(manually_removed_bucket)
|
||||
.await
|
||||
.expect("manual-removal bucket generation identity");
|
||||
for disk_path in &env.disk_paths {
|
||||
tokio::fs::remove_dir_all(disk_path.join(manually_removed_bucket))
|
||||
.await
|
||||
.expect("simulate out-of-band volume removal");
|
||||
}
|
||||
assert!(
|
||||
!manager
|
||||
.mrf_bucket_incarnation_retired(manually_removed_bucket, manual_incarnation)
|
||||
.await
|
||||
.expect("missing retirement proof is authoritative absence of proof"),
|
||||
"out-of-band disk removal must not be mistaken for a durable retirement record"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lifecycle_listing_uses_stable_bounded_cursors() {
|
||||
let mut writes = PartialWrites::default();
|
||||
|
||||
@@ -459,6 +459,13 @@ pub trait HealStorageAPI: Send + Sync {
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
/// Whether the exact old bucket generation has a durable deletion record.
|
||||
/// Alternate backends must fail closed unless they can validate the same
|
||||
/// deployment, bucket and incarnation identity.
|
||||
async fn mrf_bucket_incarnation_retired(&self, _bucket: &str, _expected: Uuid) -> Result<bool> {
|
||||
Ok(false)
|
||||
}
|
||||
|
||||
/// Aggregate usage-cache baselines for the requested buckets.
|
||||
async fn erasure_set_usage_baseline(&self, _buckets: &[String]) -> Result<Option<HealBucketUsageBaseline>> {
|
||||
Ok(None)
|
||||
@@ -1178,6 +1185,13 @@ impl HealStorageAPI for ECStoreHealStorage {
|
||||
.map_err(Error::Storage)
|
||||
}
|
||||
|
||||
async fn mrf_bucket_incarnation_retired(&self, bucket: &str, expected: Uuid) -> Result<bool> {
|
||||
self.ecstore
|
||||
.is_bucket_incarnation_retired(bucket, expected)
|
||||
.await
|
||||
.map_err(Error::Storage)
|
||||
}
|
||||
|
||||
async fn erasure_set_usage_baseline(&self, buckets: &[String]) -> Result<Option<HealBucketUsageBaseline>> {
|
||||
if buckets.is_empty() {
|
||||
return Ok(None);
|
||||
|
||||
@@ -53,6 +53,9 @@ pub(crate) mod storage {
|
||||
BucketInfo, BucketOperations, DiskSetSelector, HealOperations, ListOperations, ObjectIO, ObjectOperations,
|
||||
StorageAdminApi,
|
||||
};
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) use super::storage_contracts::DeleteBucketOptions;
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -281,8 +281,9 @@ 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_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,
|
||||
record_object_lock_acquire_failure, 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::{
|
||||
|
||||
@@ -62,6 +62,25 @@ pub fn record_contention_event() {
|
||||
counter!("rustfs_lock_contentions").increment(1);
|
||||
}
|
||||
|
||||
/// Record a failed object namespace lock acquisition with bounded labels.
|
||||
#[inline(always)]
|
||||
pub fn record_object_lock_acquire_failure(
|
||||
op: &'static str,
|
||||
mode: &'static str,
|
||||
resource_class: &'static str,
|
||||
reason: &'static str,
|
||||
) {
|
||||
use metrics::counter;
|
||||
counter!(
|
||||
"rustfs_object_lock_acquire_failures_total",
|
||||
"op" => op,
|
||||
"mode" => mode,
|
||||
"resource_class" => resource_class,
|
||||
"reason" => reason
|
||||
)
|
||||
.increment(1);
|
||||
}
|
||||
|
||||
/// Record a remote lock RPC that exceeded its caller's deadline.
|
||||
#[inline(always)]
|
||||
pub fn record_remote_lock_rpc_timeout(peer: &str, op: &'static str) {
|
||||
@@ -244,6 +263,7 @@ mod tests {
|
||||
record_remote_lock_rpc_detached("lock", "detached");
|
||||
record_remote_lock_rpc_late_completion("lock", "success");
|
||||
record_remote_lock_late_release("released");
|
||||
record_object_lock_acquire_failure("put_object_commit", "write", "system_metadata", "timeout");
|
||||
});
|
||||
|
||||
let emitted: std::collections::HashSet<String> = snapshotter
|
||||
@@ -267,10 +287,35 @@ mod tests {
|
||||
"rustfs_remote_lock_rpc_detached_total",
|
||||
"rustfs_remote_lock_rpc_late_completions_total",
|
||||
"rustfs_remote_lock_late_releases_total",
|
||||
"rustfs_object_lock_acquire_failures_total",
|
||||
] {
|
||||
assert!(emitted.contains(expected), "{expected} must be emitted by its record helper");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn object_lock_acquire_failure_metric_is_recorded_without_resource_labels() {
|
||||
let recorder = SeenMetricsRecorder::default();
|
||||
metrics::with_local_recorder(&recorder, || {
|
||||
record_object_lock_acquire_failure("put_object_commit", "write", "system_metadata", "timeout");
|
||||
});
|
||||
|
||||
let keys = recorder.counters.lock().expect("counter keys should be lockable");
|
||||
let key = keys
|
||||
.iter()
|
||||
.find(|key| key.name() == "rustfs_object_lock_acquire_failures_total")
|
||||
.expect("object lock failure counter should be emitted");
|
||||
let labels = key.labels().map(|label| (label.key(), label.value())).collect::<Vec<_>>();
|
||||
assert_eq!(
|
||||
labels,
|
||||
[
|
||||
("op", "put_object_commit"),
|
||||
("mode", "write"),
|
||||
("resource_class", "system_metadata"),
|
||||
("reason", "timeout"),
|
||||
]
|
||||
);
|
||||
}
|
||||
use metrics::{Counter, CounterFn, Gauge, GaugeFn, Histogram, HistogramFn, Key, KeyName, Metadata, SharedString, Unit};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
|
||||
@@ -29,6 +29,7 @@ pub type S3HttpCompletionObserver = fn(S3Operation, Duration, bool);
|
||||
pub type S3HttpCompletionObserverEnabled = fn() -> bool;
|
||||
|
||||
const METRIC: &str = "rustfs_s3_http_requests_total";
|
||||
const LOCK_FAILURE_METRIC: &str = "rustfs_s3_lock_failures_total";
|
||||
const METHODS: [&str; 10] = [
|
||||
"GET", "PUT", "POST", "DELETE", "HEAD", "OPTIONS", "PATCH", "CONNECT", "TRACE", "OTHER",
|
||||
];
|
||||
@@ -112,6 +113,21 @@ pub(crate) fn observe_s3_http_operation(op: S3Operation) {
|
||||
});
|
||||
}
|
||||
|
||||
/// Attribute a lock-related S3 failure to the request operation without
|
||||
/// exporting bucket, object, owner, or request identifiers as metric labels.
|
||||
pub fn record_s3_http_lock_failure(resource_class: &'static str, mode: &'static str, reason: &'static str) {
|
||||
let _ = CURRENT_OPERATION.try_with(|current| {
|
||||
counter!(
|
||||
LOCK_FAILURE_METRIC,
|
||||
"op" => operation_label(current.get()),
|
||||
"resource_class" => resource_class,
|
||||
"mode" => mode,
|
||||
"reason" => reason
|
||||
)
|
||||
.increment(1);
|
||||
});
|
||||
}
|
||||
|
||||
/// An external request is counted exactly once: at response headers, at a
|
||||
/// service error, or when its future is dropped before producing a response.
|
||||
/// Body-stream failures after headers use the existing streaming metrics.
|
||||
@@ -196,7 +212,7 @@ pub fn s3_http_metrics_snapshot() -> Vec<S3HttpMetricSnapshot> {
|
||||
mod tests {
|
||||
use super::*;
|
||||
use metrics::with_local_recorder;
|
||||
use metrics_util::debugging::DebuggingRecorder;
|
||||
use metrics_util::debugging::{DebugValue, DebuggingRecorder};
|
||||
|
||||
#[test]
|
||||
fn completion_observer_sees_each_dispatched_request_once() {
|
||||
@@ -303,4 +319,42 @@ mod tests {
|
||||
assert_eq!(other.method, METHODS.len() - 1);
|
||||
assert_eq!(other.operation, UNKNOWN_OPERATION);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lock_failure_metric_uses_bounded_request_labels() {
|
||||
let recorder = DebuggingRecorder::new();
|
||||
with_local_recorder(&recorder, || {
|
||||
let mut request = S3HttpRequestGuard::new("PUT");
|
||||
request.in_scope(|| {
|
||||
observe_s3_http_operation(S3Operation::PutObject);
|
||||
record_s3_http_lock_failure("system_metadata", "write", "timeout");
|
||||
});
|
||||
record_s3_http_lock_failure("object_namespace", "read", "quorum");
|
||||
});
|
||||
|
||||
let exported = recorder
|
||||
.snapshotter()
|
||||
.snapshot()
|
||||
.into_vec()
|
||||
.into_iter()
|
||||
.filter(|(key, _, _, _)| key.key().name() == LOCK_FAILURE_METRIC)
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(exported.len(), 1, "background lock errors must not be attributed to an S3 request");
|
||||
let (key, _, _, value) = &exported[0];
|
||||
assert!(matches!(value, DebugValue::Counter(1)));
|
||||
let labels = key
|
||||
.key()
|
||||
.labels()
|
||||
.map(|label| (label.key(), label.value()))
|
||||
.collect::<std::collections::BTreeMap<_, _>>();
|
||||
assert_eq!(
|
||||
labels,
|
||||
std::collections::BTreeMap::from([
|
||||
("mode", "write"),
|
||||
("op", S3Operation::PutObject.as_str()),
|
||||
("reason", "timeout"),
|
||||
("resource_class", "system_metadata"),
|
||||
])
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -527,6 +527,9 @@ pub enum AdminAction {
|
||||
ImportBucketMetadataAction,
|
||||
#[strum(serialize = "admin:ExportBucketMetadata")]
|
||||
ExportBucketMetadataAction,
|
||||
/// Reconcile metadata left after an operator removed all physical bucket volumes.
|
||||
#[strum(serialize = "admin:RecoverOrphanedBucket")]
|
||||
RecoverOrphanedBucketAction,
|
||||
#[strum(serialize = "admin:GetTableCatalog")]
|
||||
GetTableCatalogAction,
|
||||
#[strum(serialize = "admin:MigrateTableCatalog")]
|
||||
@@ -682,6 +685,7 @@ impl AdminAction {
|
||||
| AdminAction::GetReplicationMetricsAction
|
||||
| AdminAction::ImportBucketMetadataAction
|
||||
| AdminAction::ExportBucketMetadataAction
|
||||
| AdminAction::RecoverOrphanedBucketAction
|
||||
| AdminAction::GetTableCatalogAction
|
||||
| AdminAction::MigrateTableCatalogAction
|
||||
| AdminAction::GetTableBucketAction
|
||||
@@ -909,6 +913,14 @@ mod tests {
|
||||
assert!(AdminAction::GetMetricsAction.is_valid());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn orphaned_bucket_recovery_admin_action_is_valid() {
|
||||
let action = AdminAction::try_from("admin:RecoverOrphanedBucket").expect("parse recovery action");
|
||||
assert_eq!(action, AdminAction::RecoverOrphanedBucketAction);
|
||||
assert!(action.is_valid());
|
||||
assert_eq!(<&str>::from(action), "admin:RecoverOrphanedBucket");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_bucket_on_demand_migration_admin_actions_are_valid() {
|
||||
let set_action = AdminAction::try_from("admin:SetBucketOnDemandMigration").expect("parse set action");
|
||||
|
||||
@@ -26,3 +26,17 @@ Alternatively, use `PUT /rustfs/admin/v3/set-remote-target?bucket=<bucket>&repla
|
||||
RustFS validates the replacement target, then reads and updates the latest persisted target set under the bucket metadata transaction lock. The flag authorizes discarding that set only if it is still unreadable at this point. A readable set, including a repair already committed by another node, is preserved and merged using the ordinary target-create identity and conflict checks. Success is returned after persistence, and an actual discard is audited after the commit. Without this opt-in, target writes retain the unreadable-configuration refusal.
|
||||
|
||||
Do not submit the diagnostic archive itself to the import endpoint. Copy only reviewed replacement entries into an ordinary import archive.
|
||||
|
||||
## Recover metadata after out-of-band volume removal
|
||||
|
||||
Removing bucket directories directly from storage disks bypasses bucket deletion, including its lifecycle fences and durable incarnation-retirement record. S3 `DeleteBucket` does not repair this state. If an old incarnation sidecar remains while the bucket is absent by quorum, ordinary `CreateBucket` refuses to create a replacement until an administrator reconciles the orphan.
|
||||
|
||||
After confirming that the bucket's data was intentionally removed, call the authenticated `POST /rustfs/admin/v3/recover-orphaned-bucket/{bucket}` endpoint with the dedicated `RecoverOrphanedBucketAction` permission and a JSON body:
|
||||
|
||||
```json
|
||||
{"expectedIncarnationId":"<non-nil bucket incarnation UUID>"}
|
||||
```
|
||||
|
||||
Supply the exact old incarnation UUID from the bucket incarnation sidecar or a trusted metadata backup. The operation checks every configured disk in every pool. A remaining volume, offline disk, unreadable retirement metadata, missing identity, or identity mismatch stops recovery. Once all disks confirm absence, RustFS publishes durable retirement evidence before cleaning usage state, bucket metadata, table-catalog metadata, and deleted-bucket markers. Repeating the request after an interrupted cleanup is safe when the same identity is supplied.
|
||||
|
||||
This endpoint is local to one deployment and is refused while site replication is enabled; reconcile the affected bucket across sites before retrying. It is not a substitute for `DeleteBucket`: use the normal API whenever the bucket still exists, and do not delete volume directories manually as a routine operation.
|
||||
|
||||
@@ -0,0 +1,148 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//! Explicit, generation-fenced recovery for orphaned bucket metadata.
|
||||
|
||||
use crate::admin::auth::authorize_admin_request;
|
||||
use crate::admin::router::{AdminOperation, Operation, S3Router};
|
||||
use crate::admin::runtime_sources::current_object_store_handle;
|
||||
use crate::server::ADMIN_PREFIX;
|
||||
use crate::site_replication::{site_replication_enabled, with_site_replication_bucket_mutation_lock};
|
||||
use hyper::{Method, StatusCode};
|
||||
use matchit::Params;
|
||||
use rustfs_policy::policy::action::{Action, AdminAction};
|
||||
use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tracing::info;
|
||||
use uuid::Uuid;
|
||||
|
||||
const EVENT_ADMIN_ORPHAN_BUCKET_RECOVERY: &str = "admin_orphan_bucket_recovery";
|
||||
const LOG_COMPONENT_ADMIN: &str = "admin";
|
||||
const LOG_SUBSYSTEM_BUCKET: &str = "bucket_recovery";
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
#[serde(rename_all = "camelCase", deny_unknown_fields)]
|
||||
struct RecoverOrphanedBucketRequest {
|
||||
expected_incarnation_id: Uuid,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct RecoverOrphanedBucketResponse {
|
||||
bucket: String,
|
||||
retired_incarnation_id: Uuid,
|
||||
}
|
||||
|
||||
pub struct RecoverOrphanedBucketHandler;
|
||||
|
||||
pub fn register_bucket_recovery_route(router: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
|
||||
router.insert(
|
||||
Method::POST,
|
||||
format!("{}{}", ADMIN_PREFIX, "/v3/recover-orphaned-bucket/{bucket}").as_str(),
|
||||
AdminOperation(&RecoverOrphanedBucketHandler {}),
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Operation for RecoverOrphanedBucketHandler {
|
||||
async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::RecoverOrphanedBucketAction)]).await?;
|
||||
let secret_key = req
|
||||
.credentials
|
||||
.as_ref()
|
||||
.map(|credentials| credentials.secret_key.clone())
|
||||
.ok_or_else(|| s3_error!(InvalidRequest, "authentication required"))?;
|
||||
let bucket = params
|
||||
.get("bucket")
|
||||
.filter(|bucket| !bucket.is_empty())
|
||||
.ok_or_else(|| s3_error!(InvalidRequest, "bucket name is required"))?;
|
||||
let body = req
|
||||
.input
|
||||
.store_all_limited(rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE)
|
||||
.await
|
||||
.map_err(|_| s3_error!(InvalidRequest, "failed to read recovery request body"))?;
|
||||
let request: RecoverOrphanedBucketRequest = serde_json::from_slice(&body)
|
||||
.map_err(|_| s3_error!(InvalidRequest, "expected JSON with a non-nil expectedIncarnationId"))?;
|
||||
if request.expected_incarnation_id.is_nil() {
|
||||
return Err(s3_error!(InvalidRequest, "expectedIncarnationId must be non-nil"));
|
||||
}
|
||||
let Some(store) = current_object_store_handle() else {
|
||||
return Err(s3_error!(InternalError, "object store is not initialized"));
|
||||
};
|
||||
|
||||
let operation_bucket = bucket.to_owned();
|
||||
let lock_store = store.clone();
|
||||
let operation_store = store.clone();
|
||||
let expected_incarnation = request.expected_incarnation_id;
|
||||
let completed = store
|
||||
.run_detached_mutation(async move {
|
||||
let lock_bucket = operation_bucket.clone();
|
||||
with_site_replication_bucket_mutation_lock(lock_store, &lock_bucket, move || async move {
|
||||
if site_replication_enabled().await? {
|
||||
return Err(s3_error!(
|
||||
OperationAborted,
|
||||
"orphaned bucket recovery is local; reconcile every site before recreating this bucket"
|
||||
));
|
||||
}
|
||||
operation_store
|
||||
.recover_orphaned_bucket(&operation_bucket, expected_incarnation)
|
||||
.await
|
||||
.map_err(|error| s3s::S3Error::from(crate::error::ApiError::from(error)))
|
||||
})
|
||||
.await??;
|
||||
Ok::<(), s3s::S3Error>(())
|
||||
})
|
||||
.await
|
||||
.map_err(|error| s3s::S3Error::from(crate::error::ApiError::from(error)))?;
|
||||
completed?;
|
||||
|
||||
info!(
|
||||
event = EVENT_ADMIN_ORPHAN_BUCKET_RECOVERY,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_BUCKET,
|
||||
state = "recovered",
|
||||
bucket = %bucket,
|
||||
incarnation_id = %request.expected_incarnation_id,
|
||||
"orphaned bucket metadata recovery completed"
|
||||
);
|
||||
|
||||
super::admin_json_response(
|
||||
req.uri.path(),
|
||||
secret_key.expose(),
|
||||
StatusCode::OK,
|
||||
&RecoverOrphanedBucketResponse {
|
||||
bucket: bucket.to_owned(),
|
||||
retired_incarnation_id: request.expected_incarnation_id,
|
||||
},
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn recovery_request_rejects_missing_or_nil_generation_identity() {
|
||||
assert!(serde_json::from_slice::<RecoverOrphanedBucketRequest>(b"{}").is_err());
|
||||
let nil = format!(r#"{{"expectedIncarnationId":"{}"}}"#, Uuid::nil());
|
||||
assert!(
|
||||
serde_json::from_str::<RecoverOrphanedBucketRequest>(&nil)
|
||||
.expect("nil UUID parses")
|
||||
.expected_incarnation_id
|
||||
.is_nil()
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -19,6 +19,7 @@ pub mod audit;
|
||||
mod audit_runtime_config;
|
||||
pub mod batch_job;
|
||||
pub mod bucket_meta;
|
||||
pub mod bucket_recovery;
|
||||
pub mod cluster_snapshot;
|
||||
pub mod config_admin;
|
||||
pub mod diagnostics;
|
||||
@@ -147,6 +148,7 @@ mod tests {
|
||||
let _storage_info_handler = system::StorageInfoHandler {};
|
||||
let _data_usage_handler = system::DataUsageInfoHandler {};
|
||||
let _metrics_handler = realtime::MetricsHandler {};
|
||||
let _recover_orphaned_bucket = bucket_recovery::RecoverOrphanedBucketHandler {};
|
||||
let _profile_handler = profile_admin::ProfileHandler {};
|
||||
let _profile_status_handler = profile_admin::ProfileStatusHandler {};
|
||||
let _tls_status_handler = tls_debug::TlsStatusHandler {};
|
||||
|
||||
@@ -34,11 +34,12 @@ mod kms_contract;
|
||||
mod route_registration_test;
|
||||
|
||||
use handlers::{
|
||||
account, audit, batch_job, bucket_meta, cluster_snapshot, config_admin, diagnostics, durability as durability_handler,
|
||||
extensions, gateway_key_inventory, heal, health, idp_compat, ilm_transition, inspect_archive, integrity, kms, mfa,
|
||||
module_switch, object_data_cache, object_zip_download, oidc, on_demand_migration, plugins_catalog, plugins_instances, pools,
|
||||
profile_admin, quota as quota_handler, rebalance, replication as replication_handler, scanner, site_replication, sts, system,
|
||||
table_catalog, tier, tls_debug, usage_prefix, user,
|
||||
account, audit, batch_job, bucket_meta, bucket_recovery, cluster_snapshot, config_admin, diagnostics,
|
||||
durability as durability_handler, extensions, gateway_key_inventory, heal, health, idp_compat, ilm_transition,
|
||||
inspect_archive, integrity, kms, mfa, module_switch, object_data_cache, object_zip_download, oidc, on_demand_migration,
|
||||
plugins_catalog, plugins_instances, pools, profile_admin, quota as quota_handler, rebalance,
|
||||
replication as replication_handler, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, usage_prefix,
|
||||
user,
|
||||
};
|
||||
use router::{AdminOperation, S3Router};
|
||||
use s3s::route::S3Route;
|
||||
@@ -77,6 +78,7 @@ fn register_admin_routes(r: &mut S3Router<AdminOperation>) -> std::io::Result<()
|
||||
|
||||
quota_handler::register_quota_route(r)?;
|
||||
durability_handler::register_durability_route(r)?;
|
||||
bucket_recovery::register_bucket_recovery_route(r)?;
|
||||
on_demand_migration::register_on_demand_migration_route(r)?;
|
||||
bucket_meta::register_bucket_meta_route(r)?;
|
||||
config_admin::register_config_route(r)?;
|
||||
|
||||
@@ -55,6 +55,7 @@ const GET_TABLE_NAMESPACE: AdminActionRef = AdminActionRef::new("GetTableNamespa
|
||||
const HEAL: AdminActionRef = AdminActionRef::new("HealAdminAction");
|
||||
const HEALTH_INFO: AdminActionRef = AdminActionRef::new("HealthInfoAdminAction");
|
||||
const IMPORT_BUCKET_METADATA: AdminActionRef = AdminActionRef::new("ImportBucketMetadataAction");
|
||||
const RECOVER_ORPHANED_BUCKET: AdminActionRef = AdminActionRef::new("RecoverOrphanedBucketAction");
|
||||
const IMPORT_IAM: AdminActionRef = AdminActionRef::new("ImportIAMAction");
|
||||
const INSPECT_DATA: AdminActionRef = AdminActionRef::new("InspectDataAction");
|
||||
const KMS_BACKUP: AdminActionRef = AdminActionRef::new("kms:Backup");
|
||||
@@ -470,6 +471,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
|
||||
IMPORT_BUCKET_METADATA,
|
||||
RouteRiskLevel::High,
|
||||
),
|
||||
admin(
|
||||
HttpMethod::Post,
|
||||
"/rustfs/admin/v3/recover-orphaned-bucket/{bucket}",
|
||||
RECOVER_ORPHANED_BUCKET,
|
||||
RouteRiskLevel::High,
|
||||
),
|
||||
admin(HttpMethod::Get, "/rustfs/admin/v3/get-config-kv", CONFIG_UPDATE, RouteRiskLevel::High),
|
||||
admin(HttpMethod::Put, "/rustfs/admin/v3/set-config-kv", CONFIG_UPDATE, RouteRiskLevel::High),
|
||||
admin(HttpMethod::Delete, "/rustfs/admin/v3/del-config-kv", CONFIG_UPDATE, RouteRiskLevel::High),
|
||||
|
||||
@@ -288,6 +288,11 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
|
||||
admin_route(Method::GET, "/v3/export-bucket-metadata"),
|
||||
admin_route(Method::PUT, "/import-bucket-metadata"),
|
||||
admin_route(Method::PUT, "/v3/import-bucket-metadata"),
|
||||
admin_route_sample(
|
||||
Method::POST,
|
||||
"/v3/recover-orphaned-bucket/{bucket}",
|
||||
"/v3/recover-orphaned-bucket/test-bucket",
|
||||
),
|
||||
admin_route(Method::GET, "/v3/get-config-kv"),
|
||||
admin_route(Method::PUT, "/v3/set-config-kv"),
|
||||
admin_route(Method::DELETE, "/v3/del-config-kv"),
|
||||
@@ -1389,6 +1394,7 @@ fn test_register_routes_cover_representative_admin_paths() {
|
||||
assert_route(&router, Method::GET, &admin_path("/v3/export-bucket-metadata"));
|
||||
assert_route(&router, Method::PUT, &admin_path("/import-bucket-metadata"));
|
||||
assert_route(&router, Method::PUT, &admin_path("/v3/import-bucket-metadata"));
|
||||
assert_route(&router, Method::POST, &admin_path("/v3/recover-orphaned-bucket/test-bucket"));
|
||||
assert_route(&router, Method::GET, &admin_path("/v3/list-remote-targets"));
|
||||
assert_route(&router, Method::PUT, &admin_path("/v3/set-remote-target"));
|
||||
assert_route(&router, Method::POST, &admin_path("/v3/replication/diff"));
|
||||
|
||||
@@ -16,6 +16,7 @@ use crate::storage_api::error::contract::{StorageErrorCode, range::HTTPRangeErro
|
||||
use crate::storage_api::error::{PoolMetadataError, QuotaError, StorageError, unreadable_config_refusal};
|
||||
use http::StatusCode;
|
||||
use rustfs_kms::KmsUnavailableError;
|
||||
use rustfs_lock::LockError;
|
||||
use s3s::{S3Error, S3ErrorCode};
|
||||
|
||||
const MAX_VERSIONS_EXCEEDED_CODE: &str = "MaxVersionsExceeded";
|
||||
@@ -26,6 +27,58 @@ const SLOW_DOWN_READ_MESSAGE: &str = "Resource requested is unreadable, please r
|
||||
/// S3 error code for a request that names a KMS key the KMS does not hold.
|
||||
pub const KMS_KEY_NOT_FOUND_ERROR_CODE: &str = "KMS.NotFoundException";
|
||||
|
||||
fn lock_resource_class(resource: Option<&str>) -> &'static str {
|
||||
let Some(resource) = resource else {
|
||||
return "unknown";
|
||||
};
|
||||
if resource == ".rustfs.sys/multipart"
|
||||
|| resource.starts_with(".rustfs.sys/multipart/")
|
||||
|| resource == ".minio.sys/multipart"
|
||||
|| resource.starts_with(".minio.sys/multipart/")
|
||||
{
|
||||
return "multipart_metadata";
|
||||
}
|
||||
if resource.starts_with(".rustfs.sys/") || resource.starts_with(".minio.sys/") {
|
||||
return "system_metadata";
|
||||
}
|
||||
let Some((bucket, object)) = resource.split_once('/') else {
|
||||
return "unknown";
|
||||
};
|
||||
let object = object.split('@').next().unwrap_or(object);
|
||||
if object.is_empty() || object == bucket {
|
||||
"bucket_namespace"
|
||||
} else {
|
||||
"object_namespace"
|
||||
}
|
||||
}
|
||||
|
||||
fn record_s3_lock_failure(err: &StorageError) {
|
||||
let record = |resource: Option<&str>, mode: &'static str, reason: &'static str| {
|
||||
rustfs_io_metrics::s3_http_metrics::record_s3_http_lock_failure(lock_resource_class(resource), mode, reason);
|
||||
};
|
||||
match err {
|
||||
StorageError::Lock(lock_error) => match lock_error {
|
||||
LockError::Timeout { resource, .. } => record(Some(resource), "unknown", "timeout"),
|
||||
LockError::AlreadyLocked { resource, .. } => record(Some(resource), "unknown", "contention"),
|
||||
LockError::QuorumNotReached { .. } | LockError::InsufficientNodes { .. } => record(None, "unknown", "quorum"),
|
||||
LockError::Network { .. } | LockError::Internal { .. } => record(None, "unknown", "transport"),
|
||||
_ => record(None, "unknown", "other"),
|
||||
},
|
||||
StorageError::NamespaceLockQuorumUnavailable {
|
||||
mode, bucket, object, ..
|
||||
} => {
|
||||
let resource = format!("{bucket}/{object}");
|
||||
let mode = match *mode {
|
||||
"read" => "read",
|
||||
"write" => "write",
|
||||
_ => "other",
|
||||
};
|
||||
record(Some(&resource), mode, "quorum");
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
/// Map a KMS failure that surfaced on the S3 data path to its S3 error code.
|
||||
///
|
||||
/// The contract deliberately differs from the admin lifecycle handlers
|
||||
@@ -525,6 +578,7 @@ impl From<ApiError> for S3Error {
|
||||
|
||||
impl From<StorageError> for ApiError {
|
||||
fn from(err: StorageError) -> Self {
|
||||
record_s3_lock_failure(&err);
|
||||
if err.pool_metadata_failure().is_some() {
|
||||
return ApiError {
|
||||
code: S3ErrorCode::ServiceUnavailable,
|
||||
@@ -824,9 +878,52 @@ impl From<QuotaError> for ApiError {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use metrics::with_local_recorder;
|
||||
use metrics_util::debugging::{DebugValue, DebuggingRecorder};
|
||||
use rustfs_io_metrics::s3_http_metrics::S3HttpRequestGuard;
|
||||
use rustfs_s3_ops::S3Operation;
|
||||
use s3s::{S3Error, S3ErrorCode};
|
||||
use std::io::{Error as IoError, ErrorKind};
|
||||
|
||||
#[test]
|
||||
fn lock_503_mapping_records_operation_and_resource_class_without_resource_name() {
|
||||
let recorder = DebuggingRecorder::new();
|
||||
with_local_recorder(&recorder, || {
|
||||
let mut request = S3HttpRequestGuard::new("PUT");
|
||||
request.in_scope(|| {
|
||||
rustfs_io_metrics::record_s3_op(S3Operation::PutObject);
|
||||
let error = ApiError::from(StorageError::Lock(LockError::timeout(
|
||||
".rustfs.sys/scanner/durable-dirty-producer-replay.json",
|
||||
std::time::Duration::from_secs(5),
|
||||
)));
|
||||
assert_eq!(error.code, S3ErrorCode::ServiceUnavailable);
|
||||
});
|
||||
});
|
||||
|
||||
let metrics = recorder.snapshotter().snapshot().into_vec();
|
||||
let (key, _, _, value) = metrics
|
||||
.iter()
|
||||
.find(|(key, _, _, _)| key.key().name() == "rustfs_s3_lock_failures_total")
|
||||
.expect("S3 lock failures should be observable");
|
||||
assert!(matches!(value, DebugValue::Counter(1)));
|
||||
let labels = key
|
||||
.key()
|
||||
.labels()
|
||||
.map(|label| (label.key(), label.value()))
|
||||
.collect::<std::collections::BTreeMap<_, _>>();
|
||||
assert_eq!(labels["op"], S3Operation::PutObject.as_str());
|
||||
assert_eq!(labels["mode"], "unknown");
|
||||
assert_eq!(labels["reason"], "timeout");
|
||||
assert_eq!(labels["resource_class"], "system_metadata");
|
||||
assert!(!labels.values().any(|value| value.contains("durable-dirty-producer-replay")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn multipart_lock_resources_have_a_distinct_bounded_class() {
|
||||
assert_eq!(lock_resource_class(Some(".rustfs.sys/multipart/upload-1")), "multipart_metadata");
|
||||
assert_eq!(lock_resource_class(Some(".minio.sys/multipart/upload-1")), "multipart_metadata");
|
||||
}
|
||||
|
||||
/// rustfs/backlog#1734: a refusal to act on a stored bucket config that
|
||||
/// cannot be parsed is recoverable once an operator repairs the bytes, so
|
||||
/// it answers 503 naming the bucket, config and stored length, not a
|
||||
|
||||
Reference in New Issue
Block a user