gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-08 13:06:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(heal): write EC84 distributed restart oracle (#7481)

Browse Source 
Bind the distributed EC8+4 restart evidence lane to a scanner/heal oracle artifact so release validation can consume the real nextest run instead of accepting only a passing test.

Require the registry to assert 8+4 erasure geometry for the three-node, four-drive case.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-08 18:15:46 +08:00
committed by GitHub
parent 7b3dad6bae
commit 2ba7f95547
2 changed files with 184 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.config/scanner-heal-required-tests.json
+2
View File
@@ -43,6 +43,8 @@
"min_objects": 5,
"max_objects": 5,
"topology": {"nodes": 3, "drives_per_node": 4},
"erasure": {"data_blocks": 8, "parity_blocks": 4},
"erasure_set_drive_count": 12,
"scope": "3-node x 4-drive single-set EC8+4, graceful target restart, preformatted replacement drive, exact unversioned S3 bodies and physical target shards; not mixed-version, multi-pool or long-window ABBA."
},
"background-target-restart-ec8-4": {
crates/e2e_test/src/distributed/heal_test.rs
+182 -1
View File
@@ -12,12 +12,18 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::harness::{DistCluster, DistLayout, TestResult, assert_inventory, payload_for, put_object, unique_bucket, wait_until};
use super::harness::{
DistCluster, DistLayout, TestResult, assert_inventory, get_object_bytes, payload_for, put_object, sha256_hex, unique_bucket,
wait_until,
};
use crate::chaos::{VersionShardCensus, census_object_version_on_disk, signed_admin_post};
use crate::common::init_logging;
use aws_sdk_s3::Client;
use aws_sdk_s3::primitives::ByteStream;
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, HashSet};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::time::Duration;
@@ -25,6 +31,8 @@ const EC84_NODE_COUNT: usize = 3;
const EC84_DRIVES_PER_NODE: usize = 4;
const EC84_DATA_BLOCKS: usize = 8;
const EC84_PARITY_BLOCKS: usize = 4;
const EC84_TARGET_DRIVE_RESTART_CASE: &str = "ec84-target-drive-restart";
const EC84_TARGET_DRIVE_RESTART_ORACLE: &str = "ec84-target-drive-restart.json";
#[derive(Clone)]
struct ExpectedShard {
@@ -33,6 +41,79 @@ struct ExpectedShard {
baseline: VersionShardCensus,
}
struct ScannerHealEvidenceContext {
directory: PathBuf,
run: Value,
}
fn file_sha256(path: &Path) -> TestResult<String> {
let mut file = std::fs::File::open(path)?;
let mut digest = Sha256::new();
let mut buffer = [0_u8; 64 * 1024];
loop {
let read = file.read(&mut buffer)?;
if read == 0 {
break;
}
digest.update(&buffer[..read]);
}
Ok(digest.finalize().iter().map(|byte| format!("{byte:02x}")).collect())
}
fn compiled_test_identity() -> Value {
serde_json::json!({
"source_revision": env!("RUSTFS_E2E_BUILD_COMMIT"),
"dirty": env!("RUSTFS_E2E_BUILD_DIRTY") != "false",
"lock_blob": env!("RUSTFS_E2E_BUILD_LOCK"),
"features": env!("RUSTFS_E2E_BUILD_FEATURES"),
"target": env!("RUSTFS_E2E_BUILD_TARGET"),
"profile": env!("RUSTFS_E2E_BUILD_PROFILE"),
"rustflags_hex": env!("RUSTFS_E2E_BUILD_RUSTFLAGS_HEX"),
})
}
fn string_field<'a>(value: &'a Value, path: &str) -> TestResult<&'a str> {
let mut current = value;
for segment in path.split('.') {
current = current
.get(segment)
.ok_or_else(|| format!("scanner/heal run receipt missing {path}"))?;
}
current
.as_str()
.filter(|text| !text.is_empty())
.ok_or_else(|| format!("scanner/heal run receipt has invalid {path}").into())
}
fn scanner_heal_evidence_context() -> TestResult<Option<ScannerHealEvidenceContext>> {
let Some(directory) = std::env::var_os("RUSTFS_SCANNER_HEAL_RUN_DIR") else {
return Ok(None);
};
let directory = PathBuf::from(directory);
let receipt = directory.join("run.json");
if receipt.metadata()?.len() > 1024 * 1024 {
return Err("oversized scanner/heal execution receipt".into());
}
let run: Value = serde_json::from_slice(&std::fs::read(receipt)?)?;
let built = compiled_test_identity();
for key in ["source_revision", "dirty", "lock_blob", "features"] {
if built[key] != run["test_build"][key] {
return Err(format!("compiled test identity differs for {key}").into());
}
}
let binary_path = PathBuf::from(string_field(&run, "binary.path")?);
if file_sha256(&binary_path)? != string_field(&run, "binary.sha256")? {
return Err("server binary must match the run receipt".into());
}
if file_sha256(&std::env::current_exe()?)? != string_field(&run, "test_binary.sha256")? {
return Err("test executable must match the run receipt".into());
}
if directory.join(EC84_TARGET_DRIVE_RESTART_ORACLE).exists() {
return Err("scanner/heal oracle already exists; create a new execution receipt".into());
}
Ok(Some(ScannerHealEvidenceContext { directory, run }))
}
fn assert_ec84_geometry(census: &VersionShardCensus, key: &str) -> TestResult {
if census.data_blocks != Some(EC84_DATA_BLOCKS) || census.parity_blocks != Some(EC84_PARITY_BLOCKS) {
return Err(format!("object {key} did not use EC8+4 geometry: {census:?}").into());
@@ -49,6 +130,76 @@ fn assert_ec84_geometry(census: &VersionShardCensus, key: &str) -> TestResult {
Ok(())
}
async fn write_scanner_heal_evidence(
context: ScannerHealEvidenceContext,
dist: &DistCluster,
bucket: &str,
expected: &[ExpectedShard],
outage_key: &str,
outage_body: &[u8],
replaced_drive: &Path,
pid_before: u32,
pid_after: u32,
node_listings: Vec<Vec<String>>,
) -> TestResult {
let verifier = dist.client(0)?;
let mut objects = Vec::new();
for item in expected {
let actual = get_object_bytes(&verifier, bucket, &item.key).await?;
let physical = census_object_version_on_disk(replaced_drive, bucket, &item.key, None)?;
objects.push(serde_json::json!({
"key": item.key,
"version_id": null,
"expected_bytes": item.body.len(),
"actual_bytes": actual.len(),
"expected_sha256": sha256_hex(&item.body),
"actual_sha256": sha256_hex(&actual),
"expected_physical": item.baseline,
"physical": physical,
}));
}
let actual = get_object_bytes(&verifier, bucket, outage_key).await?;
let physical = census_object_version_on_disk(replaced_drive, bucket, outage_key, None)?;
objects.push(serde_json::json!({
"key": outage_key,
"version_id": null,
"expected_bytes": outage_body.len(),
"actual_bytes": actual.len(),
"expected_sha256": sha256_hex(outage_body),
"actual_sha256": sha256_hex(&actual),
"expected_physical": null,
"physical": physical,
}));
let evidence = serde_json::json!({
"schema": 1,
"case": EC84_TARGET_DRIVE_RESTART_CASE,
"evidence": "process-restart",
"run_id": string_field(&context.run, "run_id")?,
"source_revision": string_field(&context.run, "source_revision")?,
"test_build": compiled_test_identity(),
"binary_sha256": string_field(&context.run, "binary.sha256")?,
"test_binary_sha256": string_field(&context.run, "test_binary.sha256")?,
"topology": {"nodes": EC84_NODE_COUNT, "drives_per_node": EC84_DRIVES_PER_NODE},
"pid_before": pid_before,
"pid_after": pid_after,
"unclean_shutdown_marker": false,
"objects": objects,
"node_listings": node_listings,
});
let data = serde_json::to_vec(&evidence)?;
if data.len() > 1024 * 1024 {
return Err("scanner/heal oracle exceeds the 1 MiB artifact budget".into());
}
let mut output = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(context.directory.join(EC84_TARGET_DRIVE_RESTART_ORACLE))?;
output.write_all(&data)?;
output.sync_all()?;
Ok(())
}
fn assert_replaced_drive_empty(drive: &Path, bucket: &str, keys: &[String]) -> TestResult {
for key in keys {
let census = census_object_version_on_disk(drive, bucket, key, None)?;
@@ -87,6 +238,7 @@ async fn put_large_inventory(client: &Client, bucket: &str) -> TestResult<Vec<Ex
#[tokio::test]
async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_restart() -> TestResult {
init_logging();
let evidence_context = scanner_heal_evidence_context()?;
let mut dist = DistCluster::start_with_env(
DistLayout::ThreeByFourEc84,
&[
@@ -116,6 +268,11 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
let format_path = replaced_drive.join(".rustfs.sys").join("format.json");
let format_json = std::fs::read(&format_path)?;
let target_pid_before = dist.cluster.nodes[replaced_node]
.process
.as_ref()
.ok_or("target process is absent before graceful restart")?
.id();
dist.cluster.stop_node_gracefully(replaced_node).await?;
let retired_drive = PathBuf::from(format!("{}.retired", replaced_drive.display()));
std::fs::rename(&replaced_drive, &retired_drive)?;
@@ -138,6 +295,11 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
.await?;
dist.cluster.start_node(replaced_node).await?;
let target_pid_after = dist.cluster.nodes[replaced_node]
.process
.as_ref()
.ok_or("target process is absent after restart")?
.id();
let heal_body =
r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
let heal_url = format!("{}/rustfs/admin/v3/heal/{bucket}?forceStart=true", dist.cluster.nodes[0].url);
@@ -167,6 +329,7 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
.chain(std::iter::once((outage_key.to_string(), outage_body.clone())))
.collect::<BTreeMap<_, _>>();
let expected_keys = inventory.keys().cloned().collect::<HashSet<_>>();
let mut node_listings = Vec::new();
for node_index in 0..dist.cluster.nodes.len() {
let client = dist.client(node_index)?;
assert_inventory(&client, &bucket, &inventory).await?;
@@ -177,6 +340,24 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
.filter_map(|object| object.key().map(str::to_owned))
.collect::<HashSet<_>>();
assert_eq!(observed, expected_keys, "node {node_index} listing diverged after EC8+4 heal");
let mut observed = observed.into_iter().collect::<Vec<_>>();
observed.sort();
node_listings.push(observed);
}
if let Some(context) = evidence_context {
write_scanner_heal_evidence(
context,
&dist,
&bucket,
&expected,
outage_key,
&outage_body,
&replaced_drive,
target_pid_before,
target_pid_after,
node_listings,
)
.await?;
}
Ok(())