mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-22 04:16:38 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| da3fd83aa7 | |||
| a2e7036cb1 |
@@ -585,9 +585,12 @@ impl VersionsHistogram {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Replication statistics for a single target
|
/// Replication statistics for a single target.
|
||||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
///
|
||||||
pub struct ReplicationStats {
|
/// Renamed from `ReplicationStats`; serde field names are preserved
|
||||||
|
/// byte-identically to maintain wire compatibility with existing snapshots.
|
||||||
|
#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||||
|
pub struct ReplicationTargetUsage {
|
||||||
pub pending_size: u64,
|
pub pending_size: u64,
|
||||||
pub replicated_size: u64,
|
pub replicated_size: u64,
|
||||||
pub failed_size: u64,
|
pub failed_size: u64,
|
||||||
@@ -600,7 +603,7 @@ pub struct ReplicationStats {
|
|||||||
pub replicated_count: u64,
|
pub replicated_count: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl ReplicationStats {
|
impl ReplicationTargetUsage {
|
||||||
pub fn is_empty(&self) -> bool {
|
pub fn is_empty(&self) -> bool {
|
||||||
let Self {
|
let Self {
|
||||||
pending_size,
|
pending_size,
|
||||||
@@ -636,7 +639,7 @@ impl ReplicationStats {
|
|||||||
/// Replication statistics for all targets
|
/// Replication statistics for all targets
|
||||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
||||||
pub struct ReplicationAllStats {
|
pub struct ReplicationAllStats {
|
||||||
pub targets: HashMap<String, ReplicationStats>,
|
pub targets: HashMap<String, ReplicationTargetUsage>,
|
||||||
pub replica_size: u64,
|
pub replica_size: u64,
|
||||||
pub replica_count: u64,
|
pub replica_count: u64,
|
||||||
}
|
}
|
||||||
@@ -649,7 +652,7 @@ impl ReplicationAllStats {
|
|||||||
targets,
|
targets,
|
||||||
} = self;
|
} = self;
|
||||||
|
|
||||||
*replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationStats::is_empty)
|
*replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationTargetUsage::is_empty)
|
||||||
}
|
}
|
||||||
|
|
||||||
#[deprecated(note = "use is_empty instead")]
|
#[deprecated(note = "use is_empty instead")]
|
||||||
@@ -2466,7 +2469,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn replication_stats_empty_checks_every_field() {
|
fn replication_stats_empty_checks_every_field() {
|
||||||
type SetField = fn(&mut ReplicationStats);
|
type SetField = fn(&mut ReplicationTargetUsage);
|
||||||
|
|
||||||
let cases: [(&str, SetField); 10] = [
|
let cases: [(&str, SetField); 10] = [
|
||||||
("pending_size", |stats| stats.pending_size = 1),
|
("pending_size", |stats| stats.pending_size = 1),
|
||||||
@@ -2481,9 +2484,9 @@ mod tests {
|
|||||||
("replicated_count", |stats| stats.replicated_count = 1),
|
("replicated_count", |stats| stats.replicated_count = 1),
|
||||||
];
|
];
|
||||||
|
|
||||||
assert!(ReplicationStats::default().is_empty());
|
assert!(ReplicationTargetUsage::default().is_empty());
|
||||||
for (field, set_nonzero) in cases {
|
for (field, set_nonzero) in cases {
|
||||||
let mut stats = ReplicationStats::default();
|
let mut stats = ReplicationTargetUsage::default();
|
||||||
set_nonzero(&mut stats);
|
set_nonzero(&mut stats);
|
||||||
assert!(!stats.is_empty(), "{field} must make replication stats non-empty");
|
assert!(!stats.is_empty(), "{field} must make replication stats non-empty");
|
||||||
}
|
}
|
||||||
@@ -2514,17 +2517,17 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
let empty_targets = ReplicationAllStats {
|
let empty_targets = ReplicationAllStats {
|
||||||
targets: HashMap::from([("arn:test:empty".to_string(), ReplicationStats::default())]),
|
targets: HashMap::from([("arn:test:empty".to_string(), ReplicationTargetUsage::default())]),
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
assert!(empty_targets.is_empty(), "all-empty targets must keep aggregate stats empty");
|
assert!(empty_targets.is_empty(), "all-empty targets must keep aggregate stats empty");
|
||||||
|
|
||||||
let stats = ReplicationAllStats {
|
let stats = ReplicationAllStats {
|
||||||
targets: HashMap::from([
|
targets: HashMap::from([
|
||||||
("arn:test:empty".to_string(), ReplicationStats::default()),
|
("arn:test:empty".to_string(), ReplicationTargetUsage::default()),
|
||||||
(
|
(
|
||||||
"arn:test:non-empty".to_string(),
|
"arn:test:non-empty".to_string(),
|
||||||
ReplicationStats {
|
ReplicationTargetUsage {
|
||||||
pending_count: 1,
|
pending_count: 1,
|
||||||
..Default::default()
|
..Default::default()
|
||||||
},
|
},
|
||||||
@@ -2565,7 +2568,7 @@ mod tests {
|
|||||||
replication_stats: Some(ReplicationAllStats {
|
replication_stats: Some(ReplicationAllStats {
|
||||||
targets: HashMap::from([(
|
targets: HashMap::from([(
|
||||||
"arn:test:pending".to_string(),
|
"arn:test:pending".to_string(),
|
||||||
ReplicationStats {
|
ReplicationTargetUsage {
|
||||||
pending_count: 1,
|
pending_count: 1,
|
||||||
..Default::default()
|
..Default::default()
|
||||||
},
|
},
|
||||||
@@ -2714,7 +2717,7 @@ mod tests {
|
|||||||
targets: HashMap::from([
|
targets: HashMap::from([
|
||||||
(
|
(
|
||||||
"arn:self-only".to_string(),
|
"arn:self-only".to_string(),
|
||||||
ReplicationStats {
|
ReplicationTargetUsage {
|
||||||
pending_size: 7,
|
pending_size: 7,
|
||||||
pending_count: 1,
|
pending_count: 1,
|
||||||
..Default::default()
|
..Default::default()
|
||||||
@@ -2722,7 +2725,7 @@ mod tests {
|
|||||||
),
|
),
|
||||||
(
|
(
|
||||||
"arn:shared".to_string(),
|
"arn:shared".to_string(),
|
||||||
ReplicationStats {
|
ReplicationTargetUsage {
|
||||||
failed_size: 3,
|
failed_size: 3,
|
||||||
failed_count: 1,
|
failed_count: 1,
|
||||||
missed_threshold_size: 2,
|
missed_threshold_size: 2,
|
||||||
@@ -2741,7 +2744,7 @@ mod tests {
|
|||||||
targets: HashMap::from([
|
targets: HashMap::from([
|
||||||
(
|
(
|
||||||
"arn:shared".to_string(),
|
"arn:shared".to_string(),
|
||||||
ReplicationStats {
|
ReplicationTargetUsage {
|
||||||
failed_size: 5,
|
failed_size: 5,
|
||||||
failed_count: 2,
|
failed_count: 2,
|
||||||
after_threshold_size: 4,
|
after_threshold_size: 4,
|
||||||
@@ -2751,7 +2754,7 @@ mod tests {
|
|||||||
),
|
),
|
||||||
(
|
(
|
||||||
"arn:other-only".to_string(),
|
"arn:other-only".to_string(),
|
||||||
ReplicationStats {
|
ReplicationTargetUsage {
|
||||||
replicated_size: 11,
|
replicated_size: 11,
|
||||||
replicated_count: 3,
|
replicated_count: 3,
|
||||||
..Default::default()
|
..Default::default()
|
||||||
@@ -2993,7 +2996,9 @@ mod tests {
|
|||||||
fn replication_target_deserialization_preserves_large_historical_maps() {
|
fn replication_target_deserialization_preserves_large_historical_maps() {
|
||||||
let mut stats = ReplicationAllStats::default();
|
let mut stats = ReplicationAllStats::default();
|
||||||
for index in 0..=1024 {
|
for index in 0..=1024 {
|
||||||
stats.targets.insert(format!("target-{index}"), ReplicationStats::default());
|
stats
|
||||||
|
.targets
|
||||||
|
.insert(format!("target-{index}"), ReplicationTargetUsage::default());
|
||||||
}
|
}
|
||||||
let encoded = rmp_serde::to_vec_named(&stats).expect("large replication target fixture should encode");
|
let encoded = rmp_serde::to_vec_named(&stats).expect("large replication target fixture should encode");
|
||||||
let decoded = rmp_serde::from_slice::<ReplicationAllStats>(&encoded)
|
let decoded = rmp_serde::from_slice::<ReplicationAllStats>(&encoded)
|
||||||
@@ -3002,6 +3007,47 @@ mod tests {
|
|||||||
assert_eq!(decoded.targets.len(), stats.targets.len());
|
assert_eq!(decoded.targets.len(), stats.targets.len());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Round-trip test: encoding a [`ReplicationTargetUsage`] and decoding it back
|
||||||
|
/// must produce the exact same value. This guards against accidental serde
|
||||||
|
/// field-name drift during the `ReplicationStats` -> `ReplicationTargetUsage`
|
||||||
|
/// rename. Wire-level field names are the serialized Rust field identifiers,
|
||||||
|
/// which must remain byte-identical.
|
||||||
|
#[test]
|
||||||
|
fn replication_target_usage_rmp_round_trip() {
|
||||||
|
let original = ReplicationTargetUsage {
|
||||||
|
pending_size: 100,
|
||||||
|
replicated_size: 2_000,
|
||||||
|
failed_size: 50,
|
||||||
|
failed_count: 3,
|
||||||
|
pending_count: 7,
|
||||||
|
missed_threshold_size: 11,
|
||||||
|
after_threshold_size: 22,
|
||||||
|
missed_threshold_count: 1,
|
||||||
|
after_threshold_count: 2,
|
||||||
|
replicated_count: 99,
|
||||||
|
};
|
||||||
|
|
||||||
|
let buf = rmp_serde::to_vec_named(&original).expect("encode ReplicationTargetUsage to msgpack");
|
||||||
|
let decoded: ReplicationTargetUsage = rmp_serde::from_slice(&buf).expect("decode ReplicationTargetUsage from msgpack");
|
||||||
|
assert_eq!(original, decoded, "round-trip through rmp must preserve every field");
|
||||||
|
|
||||||
|
// Also verify that encoding as an unnamed sequence and then decoding
|
||||||
|
// with named fields produces the correct mapping (this catches reordering).
|
||||||
|
let named_buf = rmp_serde::to_vec_named(&original).expect("re-encode for field-name pinning");
|
||||||
|
// Spot-check that known field names appear in the named encoding.
|
||||||
|
let named_str = String::from_utf8_lossy(&named_buf);
|
||||||
|
assert!(named_str.contains("pending_size"), "field 'pending_size' must survive the rename");
|
||||||
|
assert!(named_str.contains("replicated_size"), "field 'replicated_size' must survive the rename");
|
||||||
|
assert!(
|
||||||
|
named_str.contains("missed_threshold_size"),
|
||||||
|
"field 'missed_threshold_size' must survive the rename"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
named_str.contains("after_threshold_count"),
|
||||||
|
"field 'after_threshold_count' must survive the rename"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn checked_merge_rejects_noncanonical_histograms_without_mutation() {
|
fn checked_merge_rejects_noncanonical_histograms_without_mutation() {
|
||||||
let mut entry = DataUsageEntry {
|
let mut entry = DataUsageEntry {
|
||||||
|
|||||||
@@ -36,8 +36,7 @@ use crate::disk::error::DiskError;
|
|||||||
use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
|
use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
|
||||||
use crate::error::{Error, Result};
|
use crate::error::{Error, Result};
|
||||||
use crate::error::{
|
use crate::error::{
|
||||||
StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_object_not_found, is_err_operation_canceled,
|
StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found,
|
||||||
is_err_version_not_found,
|
|
||||||
};
|
};
|
||||||
use crate::layout::endpoints::EndpointServerPools;
|
use crate::layout::endpoints::EndpointServerPools;
|
||||||
use crate::object_api::{GetObjectReader, ObjectOptions};
|
use crate::object_api::{GetObjectReader, ObjectOptions};
|
||||||
@@ -774,76 +773,7 @@ async fn load_decommission_entry_exact_versions(
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn resolve_decommission_check_after_list_result(list_result: Result<()>, entry_error: Option<Error>) -> Result<()> {
|
fn resolve_decommission_check_after_list_result(list_result: Result<()>, entry_error: Option<Error>) -> Result<()> {
|
||||||
match list_result {
|
if let Some(err) = entry_error { Err(err) } else { list_result }
|
||||||
Ok(()) => entry_error.map_or(Ok(()), Err),
|
|
||||||
Err(list_err) => resolve_decommission_listing_error(Some(list_err), entry_error).map_or(Ok(()), Err),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn resolve_decommission_listing_error(listing_error: Option<Error>, entry_error: Option<Error>) -> Option<Error> {
|
|
||||||
match (listing_error, entry_error) {
|
|
||||||
(Some(listing_error), Some(entry_error)) if is_err_operation_canceled(&listing_error) => Some(entry_error),
|
|
||||||
(Some(listing_error), Some(entry_error)) if is_err_operation_canceled(&entry_error) => Some(listing_error),
|
|
||||||
(Some(listing_error), _) => Some(listing_error),
|
|
||||||
(None, entry_error) => entry_error,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn decommission_unresolved_listing_error(
|
|
||||||
bucket: &str,
|
|
||||||
prefix: &str,
|
|
||||||
candidate: Option<&str>,
|
|
||||||
candidate_count: usize,
|
|
||||||
disk_error_count: usize,
|
|
||||||
pool_index: usize,
|
|
||||||
set_index: usize,
|
|
||||||
) -> Error {
|
|
||||||
let location = candidate.unwrap_or(prefix);
|
|
||||||
Error::other(format!(
|
|
||||||
"decommission listing could not resolve metadata for {bucket}/{location} on pool {pool_index} set {set_index} ({candidate_count} candidate(s), {disk_error_count} disk error(s))"
|
|
||||||
))
|
|
||||||
}
|
|
||||||
|
|
||||||
fn resolve_decommission_partial_listing_entry(
|
|
||||||
entries: MetaCacheEntries,
|
|
||||||
resolver: MetadataResolutionParams,
|
|
||||||
bucket: &str,
|
|
||||||
prefix: &str,
|
|
||||||
disk_error_count: usize,
|
|
||||||
pool_index: usize,
|
|
||||||
set_index: usize,
|
|
||||||
) -> Result<MetaCacheEntry> {
|
|
||||||
let candidate_count = entries.as_ref().iter().flatten().count();
|
|
||||||
if let Some(entry) = entries.resolve(resolver) {
|
|
||||||
return Ok(entry);
|
|
||||||
}
|
|
||||||
|
|
||||||
let candidate = entries.as_ref().iter().flatten().map(|entry| entry.name.as_str()).next();
|
|
||||||
Err(decommission_unresolved_listing_error(
|
|
||||||
bucket,
|
|
||||||
prefix,
|
|
||||||
candidate,
|
|
||||||
candidate_count,
|
|
||||||
disk_error_count,
|
|
||||||
pool_index,
|
|
||||||
set_index,
|
|
||||||
))
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn record_decommission_entry_error(
|
|
||||||
entry_error: &Arc<tokio::sync::Mutex<Option<Error>>>,
|
|
||||||
rx: &CancellationToken,
|
|
||||||
err: Error,
|
|
||||||
) {
|
|
||||||
if rx.is_cancelled() {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
let mut first_err = entry_error.lock().await;
|
|
||||||
if first_err.is_none() && !rx.is_cancelled() {
|
|
||||||
*first_err = Some(err);
|
|
||||||
rx.cancel();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
fn resolve_decommission_pool_meta_reload_result(result: Result<()>, stage: &str) -> Result<()> {
|
fn resolve_decommission_pool_meta_reload_result(result: Result<()>, stage: &str) -> Result<()> {
|
||||||
@@ -3608,7 +3538,6 @@ impl ECStore {
|
|||||||
let rx_clone = rx.clone();
|
let rx_clone = rx.clone();
|
||||||
let bi = bi.clone();
|
let bi = bi.clone();
|
||||||
let set_id = set_idx;
|
let set_id = set_idx;
|
||||||
let listing_entry_error = entry_error.clone();
|
|
||||||
let worker = tokio::spawn(async move {
|
let worker = tokio::spawn(async move {
|
||||||
let _listing_permit = listing_permit;
|
let _listing_permit = listing_permit;
|
||||||
run_decommission_listing_with_retry(
|
run_decommission_listing_with_retry(
|
||||||
@@ -3622,11 +3551,7 @@ impl ECStore {
|
|||||||
let set = set.clone();
|
let set = set.clone();
|
||||||
let rx = rx_clone.clone();
|
let rx = rx_clone.clone();
|
||||||
let bucket = bi.clone();
|
let bucket = bi.clone();
|
||||||
let entry_error = listing_entry_error.clone();
|
async move { set.list_objects_to_decommission(rx, bucket, callback).await }
|
||||||
async move {
|
|
||||||
set.list_objects_to_decommission(rx, bucket, callback, entry_error.clone(), idx, set_id)
|
|
||||||
.await
|
|
||||||
}
|
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
@@ -3656,7 +3581,11 @@ impl ECStore {
|
|||||||
|
|
||||||
wait_decommission_worker_drain(&workers, worker_limit).await?;
|
wait_decommission_worker_drain(&workers, worker_limit).await?;
|
||||||
|
|
||||||
if let Some(err) = resolve_decommission_listing_error(listing_worker_error, entry_error.lock().await.clone()) {
|
if let Some(err) = listing_worker_error {
|
||||||
|
return Err(err);
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some(err) = entry_error.lock().await.clone() {
|
||||||
return Err(err);
|
return Err(err);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -4262,7 +4191,7 @@ impl ECStore {
|
|||||||
let buckets = self.get_buckets_to_decommission().await?;
|
let buckets = self.get_buckets_to_decommission().await?;
|
||||||
let pool = self.pools[idx].clone();
|
let pool = self.pools[idx].clone();
|
||||||
|
|
||||||
for (set_index, set) in pool.disk_set.iter().enumerate() {
|
for set in &pool.disk_set {
|
||||||
for bucket_info in &buckets {
|
for bucket_info in &buckets {
|
||||||
let mut lifecycle_config = None;
|
let mut lifecycle_config = None;
|
||||||
let mut object_lock_config = None;
|
let mut object_lock_config = None;
|
||||||
@@ -4357,7 +4286,7 @@ impl ECStore {
|
|||||||
});
|
});
|
||||||
|
|
||||||
let list_result = set
|
let list_result = set
|
||||||
.list_objects_to_decommission(callback_rx, bucket_info.clone(), callback, entry_error.clone(), idx, set_index)
|
.list_objects_to_decommission(callback_rx, bucket_info.clone(), callback)
|
||||||
.await;
|
.await;
|
||||||
let entry_error = entry_error.lock().await.clone();
|
let entry_error = entry_error.lock().await.clone();
|
||||||
resolve_decommission_check_after_list_result(list_result, entry_error)?;
|
resolve_decommission_check_after_list_result(list_result, entry_error)?;
|
||||||
@@ -5092,15 +5021,12 @@ mod tests {
|
|||||||
pub type ListCallback = Arc<dyn Fn(MetaCacheEntry) -> BoxFuture<'static, ()> + Send + Sync + 'static>;
|
pub type ListCallback = Arc<dyn Fn(MetaCacheEntry) -> BoxFuture<'static, ()> + Send + Sync + 'static>;
|
||||||
|
|
||||||
impl SetDisks {
|
impl SetDisks {
|
||||||
#[tracing::instrument(skip(self, rx, cb_func, entry_error))]
|
#[tracing::instrument(skip(self, rx, cb_func))]
|
||||||
async fn list_objects_to_decommission(
|
async fn list_objects_to_decommission(
|
||||||
self: &Arc<Self>,
|
self: &Arc<Self>,
|
||||||
rx: CancellationToken,
|
rx: CancellationToken,
|
||||||
bucket_info: DecomBucketInfo,
|
bucket_info: DecomBucketInfo,
|
||||||
cb_func: ListCallback,
|
cb_func: ListCallback,
|
||||||
entry_error: Arc<tokio::sync::Mutex<Option<Error>>>,
|
|
||||||
pool_index: usize,
|
|
||||||
set_index: usize,
|
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
let (disks, _) = self.get_online_disks_with_healing(false).await;
|
let (disks, _) = self.get_online_disks_with_healing(false).await;
|
||||||
ensure_decommission_listing_disks_available(!disks.is_empty(), &bucket_info.name)?;
|
ensure_decommission_listing_disks_available(!disks.is_empty(), &bucket_info.name)?;
|
||||||
@@ -5115,12 +5041,6 @@ impl SetDisks {
|
|||||||
};
|
};
|
||||||
|
|
||||||
let cb1 = cb_func.clone();
|
let cb1 = cb_func.clone();
|
||||||
let unresolved_error = entry_error.clone();
|
|
||||||
let unresolved_rx = rx.clone();
|
|
||||||
let unresolved_bucket = bucket_info.name.clone();
|
|
||||||
let unresolved_prefix = bucket_info.prefix.clone();
|
|
||||||
let unresolved_pool_index = pool_index;
|
|
||||||
let unresolved_set_index = set_index;
|
|
||||||
|
|
||||||
list_path_raw(
|
list_path_raw(
|
||||||
rx,
|
rx,
|
||||||
@@ -5133,51 +5053,20 @@ impl SetDisks {
|
|||||||
skip_walkdir_total_timeout: true,
|
skip_walkdir_total_timeout: true,
|
||||||
walkdir_stall_timeout: Some(DECOMMISSION_BACKGROUND_WALKDIR_STALL_TIMEOUT),
|
walkdir_stall_timeout: Some(DECOMMISSION_BACKGROUND_WALKDIR_STALL_TIMEOUT),
|
||||||
agreed: Some(Box::new(move |entry: MetaCacheEntry| Box::pin(cb1(entry)))),
|
agreed: Some(Box::new(move |entry: MetaCacheEntry| Box::pin(cb1(entry)))),
|
||||||
partial: Some(Box::new(move |entries: MetaCacheEntries, errs: &[Option<DiskError>]| {
|
partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
|
||||||
let resolver = resolver.clone();
|
let resolver = resolver.clone();
|
||||||
let cb_func = cb_func.clone();
|
let cb_func = cb_func.clone();
|
||||||
let bucket = unresolved_bucket.clone();
|
match entries.resolve(resolver) {
|
||||||
let prefix = unresolved_prefix.clone();
|
Some(entry) => {
|
||||||
let unresolved_error = unresolved_error.clone();
|
|
||||||
let unresolved_rx = unresolved_rx.clone();
|
|
||||||
let pool_index = unresolved_pool_index;
|
|
||||||
let set_index = unresolved_set_index;
|
|
||||||
let disk_error_count = errs.iter().flatten().count();
|
|
||||||
if unresolved_rx.is_cancelled() {
|
|
||||||
return Box::pin(async {});
|
|
||||||
}
|
|
||||||
|
|
||||||
match resolve_decommission_partial_listing_entry(
|
|
||||||
entries,
|
|
||||||
resolver,
|
|
||||||
&bucket,
|
|
||||||
&prefix,
|
|
||||||
disk_error_count,
|
|
||||||
pool_index,
|
|
||||||
set_index,
|
|
||||||
) {
|
|
||||||
Ok(entry) => {
|
|
||||||
warn!("decommission_pool: list_objects_to_decommission get {}", &entry.name);
|
warn!("decommission_pool: list_objects_to_decommission get {}", &entry.name);
|
||||||
Box::pin(async move {
|
Box::pin(async move {
|
||||||
cb_func(entry).await;
|
cb_func(entry).await;
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
Err(err) => Box::pin(async move {
|
None => {
|
||||||
if unresolved_rx.is_cancelled() {
|
warn!("decommission_pool: list_objects_to_decommission get none");
|
||||||
return;
|
Box::pin(async {})
|
||||||
}
|
}
|
||||||
warn!(
|
|
||||||
event = EVENT_DECOMMISSION_BUCKET,
|
|
||||||
component = LOG_COMPONENT_ECSTORE,
|
|
||||||
subsystem = LOG_SUBSYSTEM_POOLS,
|
|
||||||
bucket = %bucket,
|
|
||||||
prefix = %prefix,
|
|
||||||
state = "unresolved_entry",
|
|
||||||
error = %err,
|
|
||||||
"Decommission listing failed closed on unresolved metadata"
|
|
||||||
);
|
|
||||||
record_decommission_entry_error(&unresolved_error, &unresolved_rx, err).await;
|
|
||||||
}),
|
|
||||||
}
|
}
|
||||||
})),
|
})),
|
||||||
..Default::default()
|
..Default::default()
|
||||||
@@ -5185,10 +5074,6 @@ impl SetDisks {
|
|||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
if let Some(err) = entry_error.lock().await.clone() {
|
|
||||||
return Err(err);
|
|
||||||
}
|
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -5394,12 +5279,11 @@ mod pools_tests {
|
|||||||
has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_requested,
|
has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_requested,
|
||||||
load_decommission_entry_versions, local_decommission_queue_prefix, mark_decommission_bucket_done,
|
load_decommission_entry_versions, local_decommission_queue_prefix, mark_decommission_bucket_done,
|
||||||
merge_pool_status_refresh, missing_decommission_worker_prefix, observe_decommission_terminal_reload_result,
|
merge_pool_status_refresh, missing_decommission_worker_prefix, observe_decommission_terminal_reload_result,
|
||||||
pool_meta_has_active_decommission, record_decommission_entry_error, require_decommission_store,
|
pool_meta_has_active_decommission, require_decommission_store, resolve_decommission_bucket_done_save_result,
|
||||||
resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state,
|
resolve_decommission_bucket_state, resolve_decommission_check_after_list_result,
|
||||||
resolve_decommission_check_after_list_result, resolve_decommission_entry_cleanup_delete_result,
|
resolve_decommission_entry_cleanup_delete_result, resolve_decommission_entry_exact_versions,
|
||||||
resolve_decommission_entry_exact_versions, resolve_decommission_entry_reload_result, resolve_decommission_listing_error,
|
resolve_decommission_entry_reload_result, resolve_decommission_listing_worker_result,
|
||||||
resolve_decommission_listing_worker_result, resolve_decommission_optional_bucket_config_result,
|
resolve_decommission_optional_bucket_config_result, resolve_decommission_pool_meta_reload_result,
|
||||||
resolve_decommission_partial_listing_entry, resolve_decommission_pool_meta_reload_result,
|
|
||||||
resolve_decommission_preflight_heal_result, resolve_decommission_progress_save_result,
|
resolve_decommission_preflight_heal_result, resolve_decommission_progress_save_result,
|
||||||
resolve_decommission_spawn_failure_result, resolve_decommission_terminal_mark_after_error_result,
|
resolve_decommission_spawn_failure_result, resolve_decommission_terminal_mark_after_error_result,
|
||||||
resolve_decommission_terminal_mark_result, resolve_decommission_update_after_result,
|
resolve_decommission_terminal_mark_result, resolve_decommission_update_after_result,
|
||||||
@@ -5418,9 +5302,7 @@ mod pools_tests {
|
|||||||
use crate::error::{Error, StorageError};
|
use crate::error::{Error, StorageError};
|
||||||
use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
|
use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
|
||||||
use crate::services::rebalance::{RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats};
|
use crate::services::rebalance::{RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats};
|
||||||
use rustfs_filemeta::{
|
use rustfs_filemeta::{FileInfo, FileInfoVersions, MetaCacheEntry, ObjectPartInfo};
|
||||||
FileInfo, FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo,
|
|
||||||
};
|
|
||||||
use rustfs_rio::Index;
|
use rustfs_rio::Index;
|
||||||
use std::sync::{
|
use std::sync::{
|
||||||
Arc,
|
Arc,
|
||||||
@@ -6439,65 +6321,6 @@ mod pools_tests {
|
|||||||
assert!(matches!(err, Error::SlowDown));
|
assert!(matches!(err, Error::SlowDown));
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn test_resolve_decommission_partial_listing_entry_rejects_unresolved_metadata() {
|
|
||||||
let err = resolve_decommission_partial_listing_entry(
|
|
||||||
MetaCacheEntries(vec![None]),
|
|
||||||
MetadataResolutionParams {
|
|
||||||
dir_quorum: 2,
|
|
||||||
obj_quorum: 2,
|
|
||||||
bucket: "bucket-a".to_string(),
|
|
||||||
..Default::default()
|
|
||||||
},
|
|
||||||
"bucket-a",
|
|
||||||
"prefix/",
|
|
||||||
1,
|
|
||||||
2,
|
|
||||||
3,
|
|
||||||
)
|
|
||||||
.expect_err("unresolved partial listing must fail closed");
|
|
||||||
|
|
||||||
let message = err.to_string();
|
|
||||||
assert!(message.contains("decommission listing could not resolve metadata"));
|
|
||||||
assert!(message.contains("bucket-a/prefix/"));
|
|
||||||
assert!(message.contains("pool 2 set 3"));
|
|
||||||
assert!(message.contains("1 disk error(s)"));
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
|
||||||
async fn test_record_decommission_entry_error_cancels_listing_and_preserves_first_error() {
|
|
||||||
let entry_error = Arc::new(tokio::sync::Mutex::new(None));
|
|
||||||
let rx = CancellationToken::new();
|
|
||||||
|
|
||||||
record_decommission_entry_error(&entry_error, &rx, Error::SlowDown).await;
|
|
||||||
record_decommission_entry_error(&entry_error, &rx, Error::OperationCanceled).await;
|
|
||||||
|
|
||||||
assert!(rx.is_cancelled());
|
|
||||||
assert!(matches!(*entry_error.lock().await, Some(Error::SlowDown)));
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
|
||||||
async fn test_record_decommission_entry_error_ignores_already_canceled_listing() {
|
|
||||||
let entry_error = Arc::new(tokio::sync::Mutex::new(None));
|
|
||||||
let rx = CancellationToken::new();
|
|
||||||
rx.cancel();
|
|
||||||
|
|
||||||
record_decommission_entry_error(&entry_error, &rx, Error::SlowDown).await;
|
|
||||||
|
|
||||||
assert!(entry_error.lock().await.is_none());
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn test_resolve_decommission_listing_error_preserves_real_listing_failure() {
|
|
||||||
let err = resolve_decommission_listing_error(Some(Error::SlowDown), Some(Error::OperationCanceled))
|
|
||||||
.expect("listing failure should be returned");
|
|
||||||
assert!(matches!(err, Error::SlowDown));
|
|
||||||
|
|
||||||
let err = resolve_decommission_listing_error(Some(Error::OperationCanceled), Some(Error::SlowDown))
|
|
||||||
.expect("entry failure should be returned");
|
|
||||||
assert!(matches!(err, Error::SlowDown));
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_resolve_decommission_check_after_list_result_returns_list_result_without_entry_error() {
|
fn test_resolve_decommission_check_after_list_result_returns_list_result_without_entry_error() {
|
||||||
let err = resolve_decommission_check_after_list_result(Err(Error::OperationCanceled), None)
|
let err = resolve_decommission_check_after_list_result(Err(Error::OperationCanceled), None)
|
||||||
|
|||||||
@@ -16,7 +16,7 @@ use super::persistence::DataUsageCacheLoadAttempt;
|
|||||||
use super::*;
|
use super::*;
|
||||||
use crate::storage_api::scanner_io::{HTTPRangeSpec, ObjectIO};
|
use crate::storage_api::scanner_io::{HTTPRangeSpec, ObjectIO};
|
||||||
use crate::{ScannerGetObjectReader, ScannerPutObjReader};
|
use crate::{ScannerGetObjectReader, ScannerPutObjReader};
|
||||||
use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};
|
use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage};
|
||||||
use serde_json::Value;
|
use serde_json::Value;
|
||||||
use std::io::Cursor;
|
use std::io::Cursor;
|
||||||
use std::pin::Pin;
|
use std::pin::Pin;
|
||||||
@@ -1636,7 +1636,7 @@ fn size_recursive_prunes_empty_and_preserves_threshold_replication_stats() {
|
|||||||
replication_stats: Some(ReplicationAllStats {
|
replication_stats: Some(ReplicationAllStats {
|
||||||
targets: HashMap::from([(
|
targets: HashMap::from([(
|
||||||
"arn:test:threshold".to_string(),
|
"arn:test:threshold".to_string(),
|
||||||
ReplicationStats {
|
ReplicationTargetUsage {
|
||||||
after_threshold_count: 1,
|
after_threshold_count: 1,
|
||||||
..Default::default()
|
..Default::default()
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -13,7 +13,7 @@
|
|||||||
// limitations under the License.
|
// limitations under the License.
|
||||||
|
|
||||||
use super::*;
|
use super::*;
|
||||||
use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};
|
use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage};
|
||||||
|
|
||||||
const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]);
|
const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]);
|
||||||
|
|
||||||
@@ -271,7 +271,7 @@ fn completed_data_usage_info_flattens_nested_bucket_entries() {
|
|||||||
replication_stats: Some(ReplicationAllStats {
|
replication_stats: Some(ReplicationAllStats {
|
||||||
targets: HashMap::from([(
|
targets: HashMap::from([(
|
||||||
"arn:target".to_string(),
|
"arn:target".to_string(),
|
||||||
ReplicationStats {
|
ReplicationTargetUsage {
|
||||||
replicated_size: 2048,
|
replicated_size: 2048,
|
||||||
replicated_count: 2,
|
replicated_count: 2,
|
||||||
..Default::default()
|
..Default::default()
|
||||||
|
|||||||
@@ -206,13 +206,6 @@ def check_runner_selection(root: Path) -> list[str]:
|
|||||||
return errors
|
return errors
|
||||||
|
|
||||||
|
|
||||||
def check_s3_tests_runner(root: Path) -> list[str]:
|
|
||||||
runner = (root / "scripts/s3-tests/run.sh").read_text()
|
|
||||||
if "--showlocals" in runner:
|
|
||||||
return ["scripts/s3-tests/run.sh: pytest failure diagnostics must not dump local values"]
|
|
||||||
return []
|
|
||||||
|
|
||||||
|
|
||||||
def profile_selection(root: Path, profile: str) -> str:
|
def profile_selection(root: Path, profile: str) -> str:
|
||||||
if not re.fullmatch(r"e2e-[a-z0-9-]+", profile):
|
if not re.fullmatch(r"e2e-[a-z0-9-]+", profile):
|
||||||
raise ValueError(f"invalid e2e profile name: {profile}")
|
raise ValueError(f"invalid e2e profile name: {profile}")
|
||||||
@@ -279,7 +272,6 @@ def validate(root: Path) -> list[str]:
|
|||||||
errors.extend(check_e2e_modules(root))
|
errors.extend(check_e2e_modules(root))
|
||||||
errors.extend(check_fuzz_targets(root))
|
errors.extend(check_fuzz_targets(root))
|
||||||
errors.extend(check_runner_selection(root))
|
errors.extend(check_runner_selection(root))
|
||||||
errors.extend(check_s3_tests_runner(root))
|
|
||||||
errors.extend(check_profile_definitions(root))
|
errors.extend(check_profile_definitions(root))
|
||||||
return errors
|
return errors
|
||||||
|
|
||||||
@@ -349,23 +341,6 @@ class SelfTests(unittest.TestCase):
|
|||||||
)
|
)
|
||||||
self.assertEqual(len(check_fuzz_targets(root)), 1)
|
self.assertEqual(len(check_fuzz_targets(root)), 1)
|
||||||
|
|
||||||
def test_s3_runner_rejects_unbounded_failure_locals(self) -> None:
|
|
||||||
with tempfile.TemporaryDirectory() as tmp:
|
|
||||||
root = Path(tmp)
|
|
||||||
runner = root / "scripts/s3-tests/run.sh"
|
|
||||||
runner.parent.mkdir(parents=True)
|
|
||||||
runner.write_text("tox -- -vv -ra --tb=long\n")
|
|
||||||
self.assertEqual(check_s3_tests_runner(root), [])
|
|
||||||
runner.write_text("tox -- -vv -ra --showlocals --tb=long\n")
|
|
||||||
self.assertEqual(len(check_s3_tests_runner(root)), 1)
|
|
||||||
with (
|
|
||||||
mock.patch(__name__ + ".check_e2e_modules", return_value=[]),
|
|
||||||
mock.patch(__name__ + ".check_fuzz_targets", return_value=[]),
|
|
||||||
mock.patch(__name__ + ".check_runner_selection", return_value=[]),
|
|
||||||
mock.patch(__name__ + ".check_profile_definitions", return_value=[]),
|
|
||||||
):
|
|
||||||
self.assertEqual(len(validate(root)), 1)
|
|
||||||
|
|
||||||
def test_profile_listing_enforces_selection(self) -> None:
|
def test_profile_listing_enforces_selection(self) -> None:
|
||||||
with tempfile.TemporaryDirectory() as tmp:
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
root = Path(tmp)
|
root = Path(tmp)
|
||||||
@@ -436,7 +411,7 @@ def main() -> int:
|
|||||||
for error in errors:
|
for error in errors:
|
||||||
print(f"ERROR: {error}", file=sys.stderr)
|
print(f"ERROR: {error}", file=sys.stderr)
|
||||||
return 1
|
return 1
|
||||||
print("OK: e2e modules, runner selection, fuzz matrices, profiles, and bounded diagnostics are wired")
|
print("OK: e2e modules, runner selection, fuzz matrices, and profile guards are wired")
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1028,11 +1028,10 @@ else
|
|||||||
fi
|
fi
|
||||||
|
|
||||||
# Run tests from s3tests/functional
|
# Run tests from s3tests/functional
|
||||||
# Failure locals can contain multi-MiB request bodies; keep tracebacks without expanding local values.
|
|
||||||
set +e
|
set +e
|
||||||
S3TEST_CONF="${CONF_OUTPUT_PATH}" \
|
S3TEST_CONF="${CONF_OUTPUT_PATH}" \
|
||||||
tox -- \
|
tox -- \
|
||||||
-vv -ra --tb=long \
|
-vv -ra --showlocals --tb=long \
|
||||||
--maxfail="${MAXFAIL}" \
|
--maxfail="${MAXFAIL}" \
|
||||||
--timeout="${TEST_TIMEOUT}" \
|
--timeout="${TEST_TIMEOUT}" \
|
||||||
--junitxml="${ARTIFACTS_DIR}/junit.xml" \
|
--junitxml="${ARTIFACTS_DIR}/junit.xml" \
|
||||||
|
|||||||
Reference in New Issue
Block a user