mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-06 12:09:12 +00:00
test(scanner): verify default scoped entry fallback walks (#7241)
This commit is contained in:
houseme
@@ -39,6 +39,8 @@ use temp_env::with_var;
|
||||
use time::OffsetDateTime;
|
||||
use uuid::Uuid;
|
||||
|
||||
mod scoped_entry_fallback;
|
||||
|
||||
#[derive(Clone)]
|
||||
struct FixedWorkloadProvider {
|
||||
snapshot: WorkloadAdmissionRegistrySnapshot,
|
||||
|
||||
@@ -0,0 +1,289 @@
|
||||
// Copyright 2026 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.
|
||||
|
||||
use super::*;
|
||||
use crate::data_usage_define::{DATA_USAGE_OBJ_NAME_PATH, read_config_with_revision};
|
||||
use crate::storage_api::owner::EcstoreDiskAPI;
|
||||
|
||||
type DriveIdentities = HashMap<String, (Uuid, DataUsageCacheSource)>;
|
||||
type WalkCounts = HashMap<(String, String, String), u64>;
|
||||
|
||||
async fn drive_identities(store: &ECStore) -> DriveIdentities {
|
||||
let mut identities = HashMap::new();
|
||||
let mut ids = HashSet::new();
|
||||
for set in store.all_set_disks() {
|
||||
let source = DataUsageCacheSource::new(set.pool_index, set.set_index);
|
||||
for disk in scanner_set_disk_inventory(set.as_ref()).await {
|
||||
let id = EcstoreDiskAPI::get_disk_id(disk.as_ref())
|
||||
.await
|
||||
.expect("fixture disk identity should be readable")
|
||||
.expect("fixture disk must have a durable identity");
|
||||
assert!(!id.is_nil());
|
||||
assert!(ids.insert(id), "fixture disk identities must be unique");
|
||||
let path = crate::ScannerDiskExt::path(disk.as_ref()).to_string_lossy().into_owned();
|
||||
assert!(identities.insert(path, (id, source)).is_none());
|
||||
}
|
||||
}
|
||||
assert_eq!(identities.len(), 8);
|
||||
identities
|
||||
}
|
||||
|
||||
fn walk_counts(drives: &DriveIdentities) -> WalkCounts {
|
||||
rustfs_scanner_metrics::metrics::global_metrics()
|
||||
.scanner_runtime_details_report()
|
||||
.bucket_drive_results
|
||||
.into_iter()
|
||||
.filter(|result| drives.contains_key(&result.drive))
|
||||
.map(|result| ((result.bucket, result.drive, result.result), result.count))
|
||||
.collect()
|
||||
}
|
||||
|
||||
async fn put_and_settle(store: &ECStore, bucket: &str, object: &str) {
|
||||
let set = &store.pools[0].disk_set[0];
|
||||
let mut reader = ScannerPutObjReader::from_vec(b"object".to_vec());
|
||||
set.put_object(bucket, object, &mut reader, &ScannerObjectOptions::default())
|
||||
.await
|
||||
.expect("fixture object should persist");
|
||||
let lock = set.new_ns_lock(bucket, object).await.expect("fixture namespace lock");
|
||||
let _settled = lock
|
||||
.get_write_lock(Duration::from_secs(30))
|
||||
.await
|
||||
.expect("quorum-ACK rename tail must settle before taking the activity baseline");
|
||||
}
|
||||
|
||||
async fn create_bucket(store: &ECStore, bucket: &str) {
|
||||
store
|
||||
.make_bucket(bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("fixture bucket should be created");
|
||||
put_and_settle(store, bucket, "initial").await;
|
||||
}
|
||||
|
||||
async fn persist_baseline(store: &Arc<ECStore>, baseline: &DataUsageInfo) {
|
||||
let mut baseline = baseline.clone();
|
||||
baseline.usage_snapshot_converged = Some(true);
|
||||
crate::save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_OBJ_NAME_PATH.as_str(),
|
||||
serde_json::to_vec(&baseline).expect("baseline should encode"),
|
||||
)
|
||||
.await
|
||||
.expect("fixture baseline should persist");
|
||||
}
|
||||
|
||||
// Every invocation uses the production default scope. The expected walker set
|
||||
// comes from storage's per-source inventory, not the resolver's selected names.
|
||||
async fn run_entry(store: &Arc<ECStore>, cycle: u64, selected: Option<&str>, expect_walks: bool) -> DataUsageInfo {
|
||||
let drives = drive_identities(store).await;
|
||||
let inventory = store
|
||||
.list_bucket_for_scanner(&BucketOptions::default())
|
||||
.await
|
||||
.expect("fixture inventory should be complete");
|
||||
assert!(inventory.topology_complete);
|
||||
let expected_walks = if expect_walks {
|
||||
inventory
|
||||
.set_buckets
|
||||
.into_iter()
|
||||
.flat_map(|set| {
|
||||
let source = DataUsageCacheSource::new(set.pool_index, set.set_index);
|
||||
set.buckets.into_iter().map(move |bucket| ((source, bucket.name), 1_u64))
|
||||
})
|
||||
.collect::<HashMap<_, _>>()
|
||||
} else {
|
||||
HashMap::new()
|
||||
};
|
||||
let root_before = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("root baseline should be readable");
|
||||
let dirty_before = dirty_usage_buckets_for_tests();
|
||||
let generation_before = dirty_usage_generation();
|
||||
let before = walk_counts(&drives);
|
||||
let ctx = CancellationToken::new();
|
||||
let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
|
||||
let (updates, mut receiver) = mpsc::channel(1);
|
||||
let (observer, observed) = tokio::sync::oneshot::channel();
|
||||
let result = tokio::time::timeout(
|
||||
Duration::from_secs(30),
|
||||
nsscanner_with_storage_status_scoped(
|
||||
store.as_ref(),
|
||||
ScannerCycleRequest {
|
||||
ctx,
|
||||
budget,
|
||||
updates,
|
||||
want_cycle: cycle,
|
||||
leader_epoch: 11,
|
||||
scan_mode: HealScanMode::Normal,
|
||||
scan_scope: ScannerBucketScanScope::default(),
|
||||
persisted_usage_baseline: root_before.0.clone().map(Bytes::from),
|
||||
requires_full_scan: false,
|
||||
resolved_scope_observer: Some(observer),
|
||||
},
|
||||
),
|
||||
)
|
||||
.await
|
||||
.expect("entry cycle should finish within the fixture deadline")
|
||||
.expect("entry cycle should succeed");
|
||||
assert_eq!(result.status, ScannerCycleStatus::Complete);
|
||||
let scope = observed.await.expect("production resolver should report its decision");
|
||||
assert_eq!(
|
||||
scope.selected_buckets.as_deref(),
|
||||
selected.map(|name| HashSet::from([name.to_string()])).as_ref()
|
||||
);
|
||||
let usage = receiver.recv().await.expect("one candidate should be delivered");
|
||||
assert!(receiver.recv().await.is_none(), "there must be exactly one terminal candidate");
|
||||
assert!(usage.usage_snapshot_complete);
|
||||
assert!(!usage.usage_snapshot_partial);
|
||||
assert_eq!(usage.scanner_cycle, Some(cycle));
|
||||
assert_eq!(
|
||||
drive_identities(store).await,
|
||||
drives,
|
||||
"drive identities must not change during the oracle"
|
||||
);
|
||||
|
||||
let after = walk_counts(&drives);
|
||||
let mut actual = HashMap::new();
|
||||
for key in before.keys() {
|
||||
assert!(after.contains_key(key), "metrics eviction would invalidate this exact-delta oracle");
|
||||
}
|
||||
for ((bucket, drive, outcome), count) in after {
|
||||
let previous = before
|
||||
.get(&(bucket.clone(), drive.clone(), outcome.clone()))
|
||||
.copied()
|
||||
.unwrap_or(0);
|
||||
let delta = count.checked_sub(previous).expect("fixture counters must not reset");
|
||||
if delta > 0 {
|
||||
assert_eq!(outcome, "success", "no error or partial walker is expected");
|
||||
*actual.entry((drives[&drive].1, bucket)).or_insert(0_u64) += delta;
|
||||
}
|
||||
}
|
||||
assert_eq!(
|
||||
actual, expected_walks,
|
||||
"each listed source/bucket must have exactly the expected real walks"
|
||||
);
|
||||
assert_eq!(
|
||||
read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("root after scan"),
|
||||
root_before,
|
||||
"producing a candidate must not replace the coordinator-owned root baseline"
|
||||
);
|
||||
assert_eq!(dirty_usage_generation(), generation_before);
|
||||
assert!(
|
||||
dirty_usage_buckets_for_tests() == dirty_before,
|
||||
"candidate delivery must not ACK pending dirty buckets"
|
||||
);
|
||||
usage
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks() {
|
||||
let (_dir, store) = setup_two_pool_scanner_store().await;
|
||||
clear_dirty_usage_buckets_for_tests();
|
||||
let hot = format!("hot-{}", Uuid::new_v4().simple());
|
||||
let cold = format!("cold-{}", Uuid::new_v4().simple());
|
||||
create_bucket(&store, &hot).await;
|
||||
create_bucket(&store, &cold).await;
|
||||
record_dirty_usage_bucket(&hot);
|
||||
let baseline = run_entry(&store, 1, None, true).await;
|
||||
persist_baseline(&store, &baseline).await;
|
||||
|
||||
// A same-intent, same-cycle Current cache is a retry, not proof that a
|
||||
// later cycle may reuse unselected buckets without durable incarnation.
|
||||
run_entry(&store, 1, Some(&hot), false).await;
|
||||
let usage = run_entry(&store, 2, Some(&hot), true).await;
|
||||
assert_eq!(usage.buckets_usage[&hot].objects_count, 1);
|
||||
assert_eq!(usage.buckets_usage[&cold].objects_count, 1);
|
||||
assert_eq!(usage.objects_total_count, 2);
|
||||
clear_dirty_usage_buckets_for_tests();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scoped_entry_fallback_rejects_invalid_persisted_baseline_at_the_walker() {
|
||||
let (_dir, store) = setup_two_pool_scanner_store().await;
|
||||
clear_dirty_usage_buckets_for_tests();
|
||||
let hot = format!("hot-{}", Uuid::new_v4().simple());
|
||||
let cold = format!("cold-{}", Uuid::new_v4().simple());
|
||||
create_bucket(&store, &hot).await;
|
||||
create_bucket(&store, &cold).await;
|
||||
record_dirty_usage_bucket(&hot);
|
||||
// The first real scan is also the missing persisted-baseline case.
|
||||
let baseline = run_entry(&store, 1, None, true).await;
|
||||
for (index, kind) in [
|
||||
"malformed",
|
||||
"unconverged",
|
||||
"missing-set",
|
||||
"wrong-source",
|
||||
"mixed-plan",
|
||||
"wrong-epoch",
|
||||
]
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
{
|
||||
let mut candidate = baseline.clone();
|
||||
candidate.usage_snapshot_converged = Some(true);
|
||||
match kind {
|
||||
"unconverged" => candidate.usage_snapshot_converged = Some(false),
|
||||
"missing-set" => {
|
||||
candidate.usage_snapshot_set_states.pop();
|
||||
}
|
||||
"wrong-source" => candidate.usage_snapshot_set_states[0].set_index = 99,
|
||||
"mixed-plan" => candidate.usage_snapshot_set_states[1].scan_plan_digest = Some([0xA5; 32]),
|
||||
"wrong-epoch" => candidate.usage_snapshot_set_states[0].scanner_epoch = Some(10),
|
||||
"malformed" => {}
|
||||
_ => unreachable!(),
|
||||
}
|
||||
let bytes = if kind == "malformed" {
|
||||
b"{broken".to_vec()
|
||||
} else {
|
||||
serde_json::to_vec(&candidate).expect("candidate JSON")
|
||||
};
|
||||
crate::save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), bytes)
|
||||
.await
|
||||
.expect("negative baseline should persist");
|
||||
let usage = run_entry(&store, u64::try_from(index).expect("fixture cycle index should fit") + 2, None, true).await;
|
||||
assert_eq!(usage.objects_total_count, 2, "{kind}");
|
||||
assert_eq!(usage.buckets_usage[&cold].objects_count, 1, "{kind}");
|
||||
}
|
||||
clear_dirty_usage_buckets_for_tests();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scoped_entry_fallback_covers_overflow_and_new_bucket_inventory() {
|
||||
let (_dir, store) = setup_two_pool_scanner_store().await;
|
||||
clear_dirty_usage_buckets_for_tests();
|
||||
let hot = format!("hot-{}", Uuid::new_v4().simple());
|
||||
create_bucket(&store, &hot).await;
|
||||
record_dirty_usage_bucket(&hot);
|
||||
let baseline = run_entry(&store, 1, None, true).await;
|
||||
persist_baseline(&store, &baseline).await;
|
||||
for index in 0..=crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES {
|
||||
record_dirty_usage_bucket(&format!("overflow-{index}"));
|
||||
}
|
||||
assert!(dirty_usage_buckets_for_tests().len() > crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES);
|
||||
let usage = run_entry(&store, 2, None, true).await;
|
||||
assert_eq!(usage.objects_total_count, 1);
|
||||
|
||||
clear_dirty_usage_buckets_for_tests();
|
||||
record_dirty_usage_bucket(&hot);
|
||||
let new_bucket = format!("new-{}", Uuid::new_v4().simple());
|
||||
create_bucket(&store, &new_bucket).await;
|
||||
// Even a previously valid baseline cannot cover the changed inventory.
|
||||
let usage = run_entry(&store, 3, None, true).await;
|
||||
assert_eq!(usage.objects_total_count, 2);
|
||||
assert_eq!(usage.buckets_usage[&new_bucket].objects_count, 1);
|
||||
clear_dirty_usage_buckets_for_tests();
|
||||
}
|
||||
Reference in New Issue
Block a user