mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-06 20:19:14 +00:00
RustFS
07833379b4
* test(e2e): add distributed 4x4 validation * test(e2e): prove operations overlap data movement --------- Co-authored-by: Zhengchao An <anzhengchao@gmail.com>
1236 lines
47 KiB
Rust
1236 lines
47 KiB
Rust
// 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.
|
||
|
||
//! Shared 4-node distributed e2e helpers.
|
||
//!
|
||
//! Two localhost-expressible layouts cover the suite:
|
||
//!
|
||
//! * **4×4 single pool** (`four_by_four`) — four processes, four drives each,
|
||
//! one `DistErasure` pool (16 explicit volume endpoints). This is the
|
||
//! default S3 / lock / versioning / chaos topology.
|
||
//! * **4×4 four pool** — start one durable single-node pool, then append three
|
||
//! single-node pools one at a time. Required for decommission/rebalance/expand,
|
||
//! which the server rejects on a single pool. Cold-starting multiple empty
|
||
//! pools races their bootstrap identities on localhost DistErasure.
|
||
//!
|
||
//! Genuine multi-node *striped* pools still need multi-host CI (backlog
|
||
//! #1313 / #1314). Site replication uses two 4-node 1-drive clusters so the
|
||
//! process count stays at eight rather than sixteen.
|
||
|
||
use crate::common::{
|
||
ClusterTopology, FAST_DATA_USAGE_SCANNER_ENV, RustFSTestClusterEnvironment, admin_request, build_test_s3_config,
|
||
local_http_client, replication_fast_env, signed_request,
|
||
};
|
||
use crate::replication_extension_test::LOOPBACK_REPLICATION_TARGET_ENV;
|
||
use aws_sdk_s3::Client;
|
||
use aws_sdk_s3::primitives::ByteStream;
|
||
use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
|
||
use http::{Method, StatusCode};
|
||
use sha2::{Digest, Sha256};
|
||
use std::collections::BTreeMap;
|
||
use std::path::{Path, PathBuf};
|
||
use std::time::Duration;
|
||
use tokio::time::{Instant, sleep};
|
||
use uuid::Uuid;
|
||
|
||
pub(crate) type TestResult<T = ()> = Result<T, Box<dyn std::error::Error + Send + Sync>>;
|
||
|
||
pub(crate) const NODE_COUNT: usize = 4;
|
||
pub(crate) const DRIVES_PER_NODE: usize = 4;
|
||
/// Retire the seed pool after test data is written there, proving that user
|
||
/// objects—not only internal metadata—move to the expansion pools.
|
||
pub(crate) const DECOMMISSION_POOL_ID: usize = 0;
|
||
const POOL_ROOTS_ENV: &str = "RUSTFS_E2E_POOL_ROOTS";
|
||
const POOL_META_V3_ENV: [(&str, &str); 2] = [
|
||
("RUSTFS_POOL_META_V3_WRITE", "true"),
|
||
("RUSTFS_POOL_META_V3_FLEET_CONFIRMED", "true"),
|
||
];
|
||
|
||
#[derive(Clone, Copy, Debug)]
|
||
pub(crate) enum DistLayout {
|
||
/// 4 nodes × 4 drives, one erasure pool spanning every endpoint.
|
||
FourByFour,
|
||
/// 4 nodes × 1 drive, one erasure pool (minimum 4-node 4-disk layout).
|
||
FourNodeFourDisk,
|
||
/// 1 node × 4 drives, used only as the durable seed for pool expansion.
|
||
SingleNodeFourDrive,
|
||
}
|
||
|
||
pub(crate) struct DistCluster {
|
||
pub cluster: RustFSTestClusterEnvironment,
|
||
pool_storage_roots: Option<Vec<PathBuf>>,
|
||
owned_pool_dirs: Vec<PathBuf>,
|
||
}
|
||
|
||
impl DistCluster {
|
||
pub async fn start(layout: DistLayout) -> TestResult<Self> {
|
||
Self::start_with_env(layout, &[]).await
|
||
}
|
||
|
||
pub async fn start_with_env(layout: DistLayout, extra_env: &[(&str, &str)]) -> TestResult<Self> {
|
||
let mut dist = Self::new_stopped_with_env(layout, extra_env).await?;
|
||
dist.cluster.start().await?;
|
||
Ok(dist)
|
||
}
|
||
|
||
/// Allocate ports and data dirs without spawning processes.
|
||
///
|
||
/// Upgrade tests configure capture logs, then start a pinned previous
|
||
/// binary against the same directories.
|
||
pub async fn new_stopped(layout: DistLayout) -> TestResult<Self> {
|
||
Self::new_stopped_with_env(layout, &[]).await
|
||
}
|
||
|
||
pub async fn new_stopped_with_env(layout: DistLayout, extra_env: &[(&str, &str)]) -> TestResult<Self> {
|
||
let topology = match layout {
|
||
DistLayout::FourByFour => ClusterTopology::single_pool_multidrive(NODE_COUNT, DRIVES_PER_NODE),
|
||
DistLayout::FourNodeFourDisk => ClusterTopology::single_pool(NODE_COUNT),
|
||
DistLayout::SingleNodeFourDrive => ClusterTopology::per_node_pools(DRIVES_PER_NODE, vec![vec![0]]),
|
||
};
|
||
let mut cluster = RustFSTestClusterEnvironment::with_topology(topology).await?;
|
||
let pool_storage_roots = match layout {
|
||
DistLayout::SingleNodeFourDrive => Some(configured_pool_storage_roots()?),
|
||
DistLayout::FourByFour | DistLayout::FourNodeFourDisk => None,
|
||
};
|
||
let mut owned_pool_dirs = Vec::new();
|
||
if let Some(roots) = pool_storage_roots.as_deref() {
|
||
owned_pool_dirs.push(relocate_pool_storage(&mut cluster, 0, roots)?);
|
||
}
|
||
cluster.set_env("NO_PROXY", "127.0.0.1,localhost");
|
||
cluster.set_env("HTTP_PROXY", "");
|
||
cluster.set_env("HTTPS_PROXY", "");
|
||
if matches!(layout, DistLayout::SingleNodeFourDrive) {
|
||
// This fresh, same-version fleet is safe to initialize at V3. The
|
||
// durable generation protocol is required for concurrent
|
||
// decommission progress updates and crash-recoverable movement.
|
||
for (key, value) in POOL_META_V3_ENV {
|
||
cluster.set_env(key, value);
|
||
}
|
||
}
|
||
for &(key, value) in extra_env {
|
||
cluster.set_env(key, value);
|
||
}
|
||
configure_node_logs(&mut cluster)?;
|
||
Ok(Self {
|
||
cluster,
|
||
pool_storage_roots,
|
||
owned_pool_dirs,
|
||
})
|
||
}
|
||
|
||
/// Start every node with a specific `rustfs` binary, keeping the allocated
|
||
/// data directories. Used to seed an old on-disk format before upgrading.
|
||
pub async fn start_from_binary(&mut self, binary: &Path) -> TestResult {
|
||
self.cluster.start_with_binary(binary).await?;
|
||
wait_for_ready(&self.cluster).await?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Stop every node and bring the same data directories up on the workspace
|
||
/// binary (direct upgrade).
|
||
pub async fn restart_with_current_binary(&mut self) -> TestResult {
|
||
self.cluster.stop();
|
||
self.cluster.start().await?;
|
||
wait_for_ready(&self.cluster).await?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Planned topology changes use graceful shutdown. Crash semantics are
|
||
/// exercised separately by the chaos cases.
|
||
pub async fn restart_current_binary_gracefully(&mut self) -> TestResult {
|
||
self.stop_all_gracefully().await?;
|
||
self.cluster.start().await?;
|
||
wait_for_ready(&self.cluster).await?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Replace one running node with the workspace binary (rolling upgrade).
|
||
pub async fn replace_node_with_current_binary(&mut self, node_idx: usize) -> TestResult {
|
||
self.cluster.stop_node(node_idx)?;
|
||
self.cluster.start_node(node_idx).await?;
|
||
wait_for_ready(&self.cluster).await?;
|
||
Ok(())
|
||
}
|
||
|
||
pub fn client_with_credentials(&self, node_idx: usize, access_key: &str, secret_key: &str) -> TestResult<Client> {
|
||
if node_idx >= self.cluster.nodes.len() {
|
||
return Err("node_idx is invalid".into());
|
||
}
|
||
Ok(Client::from_conf(build_test_s3_config(
|
||
&self.cluster.nodes[node_idx].url,
|
||
access_key,
|
||
secret_key,
|
||
None,
|
||
"cluster-iam-test",
|
||
)))
|
||
}
|
||
|
||
pub async fn append_pool_and_restart(&mut self) -> TestResult<usize> {
|
||
self.stop_all_gracefully().await?;
|
||
let node_idx = self.cluster.append_single_node_pool().await?;
|
||
let roots = self
|
||
.pool_storage_roots
|
||
.as_deref()
|
||
.ok_or("pool expansion requires configured isolated pool filesystems")?;
|
||
self.owned_pool_dirs
|
||
.push(relocate_pool_storage(&mut self.cluster, node_idx, roots)?);
|
||
configure_node_logs(&mut self.cluster)?;
|
||
self.cluster.start().await?;
|
||
wait_for_ready(&self.cluster).await?;
|
||
Ok(node_idx)
|
||
}
|
||
|
||
async fn stop_all_gracefully(&mut self) -> TestResult {
|
||
for node_idx in 0..self.cluster.nodes.len() {
|
||
self.cluster.stop_node_gracefully(node_idx).await?;
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// Expand the durable seed into four single-node pools through three
|
||
/// serialized additions. Callers can seed objects before this step so a
|
||
/// later pool-0 decommission proves that user data crosses pool boundaries.
|
||
pub async fn expand_to_four_pools(&mut self) -> TestResult {
|
||
for _ in 0..3 {
|
||
self.append_pool_and_restart().await?;
|
||
}
|
||
self.restart_current_binary_gracefully().await?;
|
||
Ok(())
|
||
}
|
||
|
||
pub async fn start_replication_pair() -> TestResult<(Self, Self)> {
|
||
let mut extra: Vec<(&str, &str)> = replication_fast_env();
|
||
extra.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
|
||
extra.extend_from_slice(FAST_DATA_USAGE_SCANNER_ENV);
|
||
let source = Self::start_with_env(DistLayout::FourNodeFourDisk, &extra).await?;
|
||
let target = Self::start_with_env(DistLayout::FourNodeFourDisk, &extra).await?;
|
||
Ok((source, target))
|
||
}
|
||
|
||
pub fn client(&self, node_idx: usize) -> TestResult<Client> {
|
||
self.cluster.create_s3_client(node_idx)
|
||
}
|
||
|
||
pub fn clients(&self) -> TestResult<Vec<Client>> {
|
||
self.cluster.create_all_clients()
|
||
}
|
||
|
||
pub async fn create_bucket(&self, bucket: &str) -> TestResult {
|
||
self.cluster.create_test_bucket(bucket).await
|
||
}
|
||
}
|
||
|
||
impl Drop for DistCluster {
|
||
fn drop(&mut self) {
|
||
self.cluster.stop();
|
||
for path in &self.owned_pool_dirs {
|
||
if let Err(error) = std::fs::remove_dir_all(path) {
|
||
eprintln!("failed to clean up isolated distributed E2E pool directory {}: {error}", path.display());
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
fn configured_pool_storage_roots() -> TestResult<Vec<PathBuf>> {
|
||
let raw = std::env::var_os(POOL_ROOTS_ENV).ok_or_else(|| {
|
||
format!(
|
||
"{POOL_ROOTS_ENV} must name {NODE_COUNT} isolated filesystems for pool expansion, decommission, and rebalance tests"
|
||
)
|
||
})?;
|
||
validate_pool_storage_roots(&raw)
|
||
}
|
||
|
||
fn validate_pool_storage_roots(raw: &std::ffi::OsStr) -> TestResult<Vec<PathBuf>> {
|
||
let roots: Vec<PathBuf> = std::env::split_paths(raw).collect();
|
||
if roots.len() != NODE_COUNT {
|
||
return Err(format!("{POOL_ROOTS_ENV} must contain exactly {NODE_COUNT} paths, got {}", roots.len()).into());
|
||
}
|
||
|
||
let mut canonical_roots = Vec::with_capacity(roots.len());
|
||
for root in roots {
|
||
if !root.is_absolute() {
|
||
return Err(format!("{POOL_ROOTS_ENV} path must be absolute: {}", root.display()).into());
|
||
}
|
||
let canonical = root
|
||
.canonicalize()
|
||
.map_err(|error| format!("{POOL_ROOTS_ENV} path {} is unavailable: {error}", root.display()))?;
|
||
if !canonical.is_dir() {
|
||
return Err(format!("{POOL_ROOTS_ENV} path is not a directory: {}", canonical.display()).into());
|
||
}
|
||
if canonical_roots.contains(&canonical) {
|
||
return Err(format!("{POOL_ROOTS_ENV} contains a duplicate path: {}", canonical.display()).into());
|
||
}
|
||
canonical_roots.push(canonical);
|
||
}
|
||
|
||
#[cfg(unix)]
|
||
{
|
||
use std::os::unix::fs::MetadataExt;
|
||
|
||
let devices: std::collections::BTreeSet<u64> = canonical_roots
|
||
.iter()
|
||
.map(|root| std::fs::metadata(root).map(|metadata| metadata.dev()))
|
||
.collect::<Result<_, _>>()?;
|
||
if devices.len() != canonical_roots.len() {
|
||
return Err(format!(
|
||
"{POOL_ROOTS_ENV} paths must be backed by distinct filesystems; found {} devices for {} paths",
|
||
devices.len(),
|
||
canonical_roots.len()
|
||
)
|
||
.into());
|
||
}
|
||
}
|
||
|
||
Ok(canonical_roots)
|
||
}
|
||
|
||
fn relocate_pool_storage(cluster: &mut RustFSTestClusterEnvironment, node_idx: usize, roots: &[PathBuf]) -> TestResult<PathBuf> {
|
||
let root = roots
|
||
.get(node_idx)
|
||
.ok_or_else(|| format!("no isolated pool filesystem configured for node {node_idx}"))?;
|
||
let cluster_name = Path::new(&cluster.temp_dir)
|
||
.file_name()
|
||
.ok_or("cluster temp directory omitted a basename")?;
|
||
let pool_run_root = root.join(cluster_name);
|
||
let node_root = pool_run_root.join(format!("node{node_idx}"));
|
||
let data_dirs: Vec<String> = (0..cluster.topology.drives_per_node)
|
||
.map(|drive| node_root.join(format!("drive{drive}")).to_string_lossy().into_owned())
|
||
.collect();
|
||
for data_dir in &data_dirs {
|
||
std::fs::create_dir_all(data_dir)?;
|
||
}
|
||
let node = cluster
|
||
.nodes
|
||
.get_mut(node_idx)
|
||
.ok_or_else(|| format!("cannot relocate missing cluster node {node_idx}"))?;
|
||
node.data_dir = data_dirs[0].clone();
|
||
node.data_dirs = data_dirs;
|
||
Ok(pool_run_root)
|
||
}
|
||
|
||
fn configure_node_logs(cluster: &mut RustFSTestClusterEnvironment) -> TestResult {
|
||
let Some(log_dir) = std::env::var_os("RUSTFS_E2E_LOG_DIR") else {
|
||
return Ok(());
|
||
};
|
||
std::fs::create_dir_all(&log_dir)?;
|
||
let cluster_id = Uuid::new_v4().simple().to_string();
|
||
for node_idx in 0..cluster.nodes.len() {
|
||
let path = Path::new(&log_dir).join(format!("cluster-{cluster_id}-node-{node_idx}.log"));
|
||
cluster.set_node_capture_log_path(node_idx, path.to_string_lossy().into_owned())?;
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
pub(crate) fn unique_bucket(prefix: &str) -> String {
|
||
let id = Uuid::new_v4().simple().to_string();
|
||
format!("{prefix}-{}", &id[..12])
|
||
}
|
||
|
||
pub(crate) fn sha256_hex(bytes: &[u8]) -> String {
|
||
let digest = Sha256::digest(bytes);
|
||
digest.iter().map(|byte| format!("{byte:02x}")).collect()
|
||
}
|
||
|
||
pub(crate) fn payload_for(key: &str, size: usize) -> Vec<u8> {
|
||
let seed = key.as_bytes();
|
||
(0..size)
|
||
.map(|idx| seed.get(idx % seed.len()).copied().unwrap_or(0) ^ (idx as u8))
|
||
.collect()
|
||
}
|
||
|
||
pub(crate) async fn put_object(client: &Client, bucket: &str, key: &str, body: Vec<u8>) -> TestResult {
|
||
client
|
||
.put_object()
|
||
.bucket(bucket)
|
||
.key(key)
|
||
.body(ByteStream::from(body))
|
||
.send()
|
||
.await?;
|
||
Ok(())
|
||
}
|
||
|
||
pub(crate) async fn get_object_bytes(client: &Client, bucket: &str, key: &str) -> TestResult<Vec<u8>> {
|
||
let output = client.get_object().bucket(bucket).key(key).send().await?;
|
||
Ok(output.body.collect().await?.into_bytes().to_vec())
|
||
}
|
||
|
||
pub(crate) async fn assert_object_bytes(client: &Client, bucket: &str, key: &str, expected: &[u8]) -> TestResult {
|
||
let got = get_object_bytes(client, bucket, key).await?;
|
||
if got.as_slice() != expected {
|
||
return Err(format!(
|
||
"object {bucket}/{key} bytes mismatch: expected {} bytes sha256={} got {} bytes sha256={}",
|
||
expected.len(),
|
||
sha256_hex(expected),
|
||
got.len(),
|
||
sha256_hex(&got)
|
||
)
|
||
.into());
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
pub(crate) async fn put_inventory(
|
||
client: &Client,
|
||
bucket: &str,
|
||
count: usize,
|
||
size: usize,
|
||
) -> TestResult<BTreeMap<String, Vec<u8>>> {
|
||
let mut inventory = BTreeMap::new();
|
||
for idx in 0..count {
|
||
let key = format!("obj-{idx:04}");
|
||
let body = payload_for(&key, size);
|
||
put_object(client, bucket, &key, body.clone()).await?;
|
||
inventory.insert(key, body);
|
||
}
|
||
Ok(inventory)
|
||
}
|
||
|
||
/// Retry only transport-level service availability failures while a data
|
||
/// movement operation changes the pool map. Generic InternalError responses
|
||
/// remain fatal because accepting them would hide server defects.
|
||
pub(crate) async fn put_inventory_retrying(
|
||
client: &Client,
|
||
bucket: &str,
|
||
count: usize,
|
||
size: usize,
|
||
timeout: Duration,
|
||
) -> TestResult<BTreeMap<String, Vec<u8>>> {
|
||
let mut inventory = BTreeMap::new();
|
||
for idx in 0..count {
|
||
let key = format!("obj-{idx:04}");
|
||
let body = payload_for(&key, size);
|
||
retrying_put(client, bucket, &key, body.clone(), timeout).await?;
|
||
inventory.insert(key, body);
|
||
}
|
||
Ok(inventory)
|
||
}
|
||
|
||
pub(crate) async fn assert_inventory(client: &Client, bucket: &str, inventory: &BTreeMap<String, Vec<u8>>) -> TestResult {
|
||
for (key, expected) in inventory {
|
||
assert_object_bytes(client, bucket, key, expected).await?;
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
pub(crate) async fn enable_versioning(client: &Client, bucket: &str) -> TestResult {
|
||
client
|
||
.put_bucket_versioning()
|
||
.bucket(bucket)
|
||
.versioning_configuration(
|
||
VersioningConfiguration::builder()
|
||
.status(BucketVersioningStatus::Enabled)
|
||
.build(),
|
||
)
|
||
.send()
|
||
.await?;
|
||
Ok(())
|
||
}
|
||
|
||
pub(crate) async fn wait_until<F, Fut>(timeout: Duration, mut probe: F, label: &str) -> TestResult
|
||
where
|
||
F: FnMut() -> Fut,
|
||
Fut: std::future::Future<Output = TestResult<bool>>,
|
||
{
|
||
let deadline = Instant::now() + timeout;
|
||
let mut delay = Duration::from_millis(50);
|
||
loop {
|
||
let last_error = match probe().await {
|
||
Ok(true) => return Ok(()),
|
||
Ok(false) => format!("{label} still false"),
|
||
Err(error) => error.to_string(),
|
||
};
|
||
if Instant::now() >= deadline {
|
||
return Err(format!("{label} did not become true within {timeout:?}: {last_error}").into());
|
||
}
|
||
sleep(delay).await;
|
||
delay = (delay * 2).min(Duration::from_secs(1));
|
||
}
|
||
}
|
||
|
||
pub(crate) async fn cluster_admin(
|
||
cluster: &RustFSTestClusterEnvironment,
|
||
method: Method,
|
||
path_and_query: &str,
|
||
body: Option<String>,
|
||
) -> TestResult<(StatusCode, String)> {
|
||
admin_request(
|
||
&cluster.nodes[0].url,
|
||
method,
|
||
path_and_query,
|
||
body,
|
||
&cluster.access_key,
|
||
&cluster.secret_key,
|
||
)
|
||
.await
|
||
}
|
||
|
||
pub(crate) async fn cluster_admin_ok(
|
||
cluster: &RustFSTestClusterEnvironment,
|
||
method: Method,
|
||
path_and_query: &str,
|
||
body: Option<String>,
|
||
) -> TestResult<String> {
|
||
let (status, response) = cluster_admin(cluster, method.clone(), path_and_query, body).await?;
|
||
if !status.is_success() {
|
||
return Err(format!("{method} {path_and_query} failed: {status} {response}").into());
|
||
}
|
||
Ok(response)
|
||
}
|
||
|
||
pub(crate) async fn wait_for_ready(cluster: &RustFSTestClusterEnvironment) -> TestResult {
|
||
let client = local_http_client();
|
||
for node in &cluster.nodes {
|
||
let url = format!("{}/health/ready", node.url);
|
||
wait_until(
|
||
Duration::from_secs(30),
|
||
|| {
|
||
let client = client.clone();
|
||
let url = url.clone();
|
||
async move {
|
||
match client.get(&url).send().await {
|
||
Ok(response) if response.status().is_success() => Ok(true),
|
||
_ => Ok(false),
|
||
}
|
||
}
|
||
},
|
||
&format!("node {} ready", node.address),
|
||
)
|
||
.await?;
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
pub(crate) async fn set_remote_target(
|
||
source: &RustFSTestClusterEnvironment,
|
||
source_bucket: &str,
|
||
target: &RustFSTestClusterEnvironment,
|
||
target_bucket: &str,
|
||
) -> TestResult<String> {
|
||
let body = serde_json::json!({
|
||
"endpoint": target.nodes[0].address,
|
||
"credentials": {
|
||
"accessKey": target.access_key,
|
||
"secretKey": target.secret_key
|
||
},
|
||
"targetbucket": target_bucket,
|
||
"secure": false,
|
||
"type": "replication"
|
||
});
|
||
let url = format!(
|
||
"{}/rustfs/admin/v3/set-remote-target?bucket={}",
|
||
source.nodes[0].url,
|
||
urlencoding::encode(source_bucket)
|
||
);
|
||
let response = signed_request(
|
||
Method::PUT,
|
||
&url,
|
||
&source.access_key,
|
||
&source.secret_key,
|
||
Some(body.to_string().into_bytes()),
|
||
Some("application/json"),
|
||
)
|
||
.await?;
|
||
if response.status() != StatusCode::OK {
|
||
let status = response.status();
|
||
let body = response.text().await.unwrap_or_default();
|
||
return Err(format!("set remote target failed: {status} {body}").into());
|
||
}
|
||
Ok(serde_json::from_slice(&response.bytes().await?)?)
|
||
}
|
||
|
||
pub(crate) async fn put_bucket_replication(source: &RustFSTestClusterEnvironment, bucket: &str, target_arn: &str) -> TestResult {
|
||
let body = format!(
|
||
r#"<ReplicationConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
|
||
<Role></Role>
|
||
<Rule>
|
||
<ID>rule-1</ID>
|
||
<Priority>1</Priority>
|
||
<Status>Enabled</Status>
|
||
<DeleteMarkerReplication>
|
||
<Status>Enabled</Status>
|
||
</DeleteMarkerReplication>
|
||
<ExistingObjectReplication>
|
||
<Status>Enabled</Status>
|
||
</ExistingObjectReplication>
|
||
<Destination>
|
||
<Bucket>{target_arn}</Bucket>
|
||
</Destination>
|
||
</Rule>
|
||
</ReplicationConfiguration>"#
|
||
);
|
||
let url = format!("{}/{bucket}?replication", source.nodes[0].url);
|
||
let response = signed_request(
|
||
Method::PUT,
|
||
&url,
|
||
&source.access_key,
|
||
&source.secret_key,
|
||
Some(body.into_bytes()),
|
||
Some("application/xml"),
|
||
)
|
||
.await?;
|
||
if !response.status().is_success() {
|
||
let status = response.status();
|
||
let body = response.text().await.unwrap_or_default();
|
||
return Err(format!("put bucket replication failed: {status} {body}").into());
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
pub(crate) async fn wait_for_replicated_bytes(
|
||
client: &Client,
|
||
bucket: &str,
|
||
key: &str,
|
||
expected: &[u8],
|
||
timeout: Duration,
|
||
) -> TestResult {
|
||
wait_until(
|
||
timeout,
|
||
|| async {
|
||
match get_object_bytes(client, bucket, key).await {
|
||
Ok(got) if got.as_slice() == expected => Ok(true),
|
||
Ok(_) => Ok(false),
|
||
Err(error) => {
|
||
let message = error.to_string();
|
||
if message.contains("NoSuchKey") || message.contains("NotFound") {
|
||
Ok(false)
|
||
} else {
|
||
Err(error)
|
||
}
|
||
}
|
||
}
|
||
},
|
||
&format!("replicated object {bucket}/{key}"),
|
||
)
|
||
.await
|
||
}
|
||
|
||
pub(crate) async fn set_bucket_quota(cluster: &RustFSTestClusterEnvironment, bucket: &str, quota_bytes: u64) -> TestResult {
|
||
wait_until(
|
||
Duration::from_secs(30),
|
||
|| async {
|
||
let (status, _) =
|
||
cluster_admin(cluster, Method::GET, &format!("/rustfs/admin/v3/quota-stats/{bucket}"), None).await?;
|
||
Ok(status.is_success() || status == StatusCode::NOT_FOUND)
|
||
},
|
||
"quota stats ready",
|
||
)
|
||
.await?;
|
||
let body = serde_json::json!({ "quota": quota_bytes, "quota_type": "HARD" }).to_string();
|
||
wait_until(
|
||
Duration::from_secs(30),
|
||
|| async {
|
||
let (status, response) =
|
||
cluster_admin(cluster, Method::PUT, &format!("/rustfs/admin/v3/quota/{bucket}"), Some(body.clone())).await?;
|
||
if status.is_success() {
|
||
return Ok(true);
|
||
}
|
||
if status == StatusCode::SERVICE_UNAVAILABLE {
|
||
return Ok(false);
|
||
}
|
||
Err(format!("failed to set quota for {bucket}: {status} {response}").into())
|
||
},
|
||
"set hard quota",
|
||
)
|
||
.await
|
||
}
|
||
|
||
/// Start decommission and fail closed unless the admin API acknowledges it.
|
||
pub(crate) async fn start_decommission(cluster: &RustFSTestClusterEnvironment, pool_id: usize) -> TestResult {
|
||
let path = format!("/rustfs/admin/v3/pools/decommission?pool={pool_id}&by-id=true");
|
||
let deadline = Instant::now() + Duration::from_secs(45);
|
||
loop {
|
||
let (status, response) = cluster_admin(cluster, Method::POST, &path, None).await?;
|
||
if status.is_success() {
|
||
return Ok(());
|
||
}
|
||
if status == StatusCode::INTERNAL_SERVER_ERROR
|
||
&& response.contains("requires a live fleet capability proof")
|
||
&& Instant::now() < deadline
|
||
{
|
||
sleep(Duration::from_secs(1)).await;
|
||
continue;
|
||
}
|
||
return Err(format!("POST {path} did not start decommission: {status} {response}").into());
|
||
}
|
||
}
|
||
|
||
pub(crate) async fn decommission_status_json(cluster: &RustFSTestClusterEnvironment) -> TestResult<serde_json::Value> {
|
||
let body = cluster_admin_ok(cluster, Method::GET, "/rustfs/admin/v3/decommission/status", None).await?;
|
||
Ok(serde_json::from_str(&body)?)
|
||
}
|
||
|
||
fn pool_entry(status: &serde_json::Value, pool_id: usize) -> Option<&serde_json::Value> {
|
||
if let Some(pools) = status.get("pools").and_then(serde_json::Value::as_array) {
|
||
return pools
|
||
.iter()
|
||
.find(|pool| pool.get("id").and_then(serde_json::Value::as_u64) == Some(pool_id as u64));
|
||
}
|
||
if status.get("id").and_then(serde_json::Value::as_u64) == Some(pool_id as u64) {
|
||
Some(status)
|
||
} else {
|
||
None
|
||
}
|
||
}
|
||
|
||
fn nonzero_u64(value: Option<&serde_json::Value>) -> bool {
|
||
value.and_then(serde_json::Value::as_u64).is_some_and(|count| count > 0)
|
||
}
|
||
|
||
fn decommission_failure(pool: &serde_json::Value) -> Option<String> {
|
||
let info = pool.get("decommissionInfo");
|
||
let flagged = |key: &str| info.and_then(|value| value.get(key)).and_then(serde_json::Value::as_bool) == Some(true);
|
||
let terminal_flag = flagged("failed")
|
||
|| flagged("canceled")
|
||
|| pool
|
||
.get("status")
|
||
.and_then(serde_json::Value::as_str)
|
||
.is_some_and(|status| status.eq_ignore_ascii_case("failed") || status.eq_ignore_ascii_case("canceled"));
|
||
let object_failures = nonzero_u64(info.and_then(|value| value.get("objectsDecommissionedFailed")));
|
||
let byte_failures = nonzero_u64(info.and_then(|value| value.get("bytesDecommissionedFailed")));
|
||
let unresolved = info
|
||
.and_then(|value| value.get("unresolvedEntries"))
|
||
.and_then(serde_json::Value::as_array)
|
||
.is_some_and(|entries| !entries.is_empty());
|
||
terminal_flag
|
||
.then(|| "decommission reported failed or canceled".to_string())
|
||
.or_else(|| object_failures.then(|| "decommission reported failed objects".to_string()))
|
||
.or_else(|| byte_failures.then(|| "decommission reported failed bytes".to_string()))
|
||
.or_else(|| unresolved.then(|| "decommission reported unresolved entries".to_string()))
|
||
}
|
||
|
||
pub(crate) fn decommission_active(status: &serde_json::Value, pool_id: usize) -> TestResult<bool> {
|
||
let pool = pool_entry(status, pool_id).ok_or_else(|| format!("pool {pool_id} missing from decommission status: {status}"))?;
|
||
if let Some(reason) = decommission_failure(pool) {
|
||
return Err(format!("{reason}: {pool}").into());
|
||
}
|
||
let info = pool
|
||
.get("decommissionInfo")
|
||
.ok_or_else(|| format!("pool {pool_id} has no decommissionInfo: {pool}"))?;
|
||
let status_text = pool.get("status").and_then(serde_json::Value::as_str).unwrap_or("");
|
||
let pool_status = pool.get("poolStatus").and_then(serde_json::Value::as_str).unwrap_or("");
|
||
let queued = info.get("queued").and_then(serde_json::Value::as_bool) == Some(true);
|
||
Ok(queued || status_text.eq_ignore_ascii_case("running") || pool_status.eq_ignore_ascii_case("decommissioning"))
|
||
}
|
||
|
||
pub(crate) fn decommission_running_with_progress(status: &serde_json::Value, pool_id: usize) -> TestResult<bool> {
|
||
let pool = pool_entry(status, pool_id).ok_or_else(|| format!("pool {pool_id} missing from decommission status: {status}"))?;
|
||
if let Some(reason) = decommission_failure(pool) {
|
||
return Err(format!("{reason}: {pool}").into());
|
||
}
|
||
let info = pool
|
||
.get("decommissionInfo")
|
||
.ok_or_else(|| format!("pool {pool_id} has no decommissionInfo: {pool}"))?;
|
||
let status_text = pool.get("status").and_then(serde_json::Value::as_str).unwrap_or("");
|
||
let pool_status = pool.get("poolStatus").and_then(serde_json::Value::as_str).unwrap_or("");
|
||
let running = status_text.eq_ignore_ascii_case("running") || pool_status.eq_ignore_ascii_case("decommissioning");
|
||
let progressed = nonzero_u64(info.get("objectsDecommissioned")) || nonzero_u64(info.get("bytesDecommissioned"));
|
||
Ok(running && progressed)
|
||
}
|
||
|
||
pub(crate) fn decommission_complete(status: &serde_json::Value, pool_id: usize) -> TestResult<bool> {
|
||
let pool = pool_entry(status, pool_id).ok_or_else(|| format!("pool {pool_id} missing from decommission status: {status}"))?;
|
||
if let Some(reason) = decommission_failure(pool) {
|
||
return Err(format!("{reason}: {pool}").into());
|
||
}
|
||
let info = pool
|
||
.get("decommissionInfo")
|
||
.ok_or_else(|| format!("pool {pool_id} has no decommissionInfo: {pool}"))?;
|
||
let complete = info.get("complete").and_then(serde_json::Value::as_bool) == Some(true);
|
||
let status_text = pool.get("status").and_then(serde_json::Value::as_str).unwrap_or("");
|
||
let pool_status = pool.get("poolStatus").and_then(serde_json::Value::as_str).unwrap_or("");
|
||
let terminal = status_text.eq_ignore_ascii_case("complete") && pool_status.eq_ignore_ascii_case("decommissioned");
|
||
let moved_data = nonzero_u64(info.get("objectsDecommissioned")) && nonzero_u64(info.get("bytesDecommissioned"));
|
||
Ok(complete && terminal && moved_data)
|
||
}
|
||
|
||
pub(crate) async fn wait_for_decommission_active(
|
||
cluster: &RustFSTestClusterEnvironment,
|
||
pool_id: usize,
|
||
timeout: Duration,
|
||
) -> TestResult {
|
||
wait_for_decommission_state(cluster, pool_id, timeout, "active", decommission_active).await
|
||
}
|
||
|
||
pub(crate) async fn wait_for_decommission_running_with_progress(
|
||
cluster: &RustFSTestClusterEnvironment,
|
||
pool_id: usize,
|
||
timeout: Duration,
|
||
) -> TestResult {
|
||
let deadline = Instant::now() + timeout;
|
||
loop {
|
||
let status = decommission_status_json(cluster).await?;
|
||
if decommission_running_with_progress(&status, pool_id)? {
|
||
return Ok(());
|
||
}
|
||
if Instant::now() >= deadline {
|
||
return Err(format!(
|
||
"decommission did not become active with non-zero progress within {timeout:?}; last status: {status}"
|
||
)
|
||
.into());
|
||
}
|
||
sleep(Duration::from_millis(100)).await;
|
||
}
|
||
}
|
||
|
||
pub(crate) async fn wait_for_decommission_complete(
|
||
cluster: &RustFSTestClusterEnvironment,
|
||
pool_id: usize,
|
||
timeout: Duration,
|
||
) -> TestResult {
|
||
wait_for_decommission_state(cluster, pool_id, timeout, "complete with non-zero progress", decommission_complete).await
|
||
}
|
||
|
||
async fn wait_for_decommission_state(
|
||
cluster: &RustFSTestClusterEnvironment,
|
||
pool_id: usize,
|
||
timeout: Duration,
|
||
expected: &str,
|
||
predicate: fn(&serde_json::Value, usize) -> TestResult<bool>,
|
||
) -> TestResult {
|
||
let deadline = Instant::now() + timeout;
|
||
loop {
|
||
let status = decommission_status_json(cluster).await?;
|
||
if predicate(&status, pool_id)? {
|
||
return Ok(());
|
||
}
|
||
if Instant::now() >= deadline {
|
||
return Err(format!("decommission did not become {expected} within {timeout:?}; last status: {status}").into());
|
||
}
|
||
sleep(Duration::from_secs(1)).await;
|
||
}
|
||
}
|
||
|
||
pub(crate) async fn start_rebalance(cluster: &RustFSTestClusterEnvironment) -> TestResult<String> {
|
||
let path = "/rustfs/admin/v3/rebalance/start";
|
||
let deadline = Instant::now() + Duration::from_secs(45);
|
||
let response = loop {
|
||
let (status, response) = cluster_admin(cluster, Method::POST, path, None).await?;
|
||
if status.is_success() {
|
||
break response;
|
||
}
|
||
if status == StatusCode::INTERNAL_SERVER_ERROR
|
||
&& response.contains("requires a live fleet capability proof")
|
||
&& Instant::now() < deadline
|
||
{
|
||
sleep(Duration::from_secs(1)).await;
|
||
continue;
|
||
}
|
||
return Err(format!("POST {path} did not start rebalance: {status} {response}").into());
|
||
};
|
||
let parsed: serde_json::Value = serde_json::from_str(&response)?;
|
||
parsed
|
||
.get("id")
|
||
.and_then(serde_json::Value::as_str)
|
||
.filter(|id| !id.is_empty())
|
||
.map(str::to_string)
|
||
.ok_or_else(|| format!("rebalance start response omitted id: {response}").into())
|
||
}
|
||
|
||
pub(crate) async fn rebalance_status_json(cluster: &RustFSTestClusterEnvironment) -> TestResult<serde_json::Value> {
|
||
let body = cluster_admin_ok(cluster, Method::GET, "/rustfs/admin/v3/rebalance/status", None).await?;
|
||
Ok(serde_json::from_str(&body)?)
|
||
}
|
||
|
||
fn validate_rebalance_status<'a>(status: &'a serde_json::Value, expected_id: &str) -> TestResult<&'a [serde_json::Value]> {
|
||
let id = status
|
||
.get("id")
|
||
.and_then(serde_json::Value::as_str)
|
||
.ok_or("rebalance status omitted id")?;
|
||
if id != expected_id {
|
||
return Err(format!("rebalance status id changed: expected {expected_id}, got {id}").into());
|
||
}
|
||
let pools = status
|
||
.get("pools")
|
||
.and_then(serde_json::Value::as_array)
|
||
.filter(|pools| !pools.is_empty())
|
||
.ok_or("rebalance status omitted non-empty pools")?;
|
||
for pool in pools {
|
||
if pool.get("status").and_then(serde_json::Value::as_str).is_some_and(|status| {
|
||
["failed", "stopped", "canceled", "cancelled"]
|
||
.iter()
|
||
.any(|terminal| status.eq_ignore_ascii_case(terminal))
|
||
}) {
|
||
return Err(format!("rebalance entered an unsuccessful terminal state: {status}").into());
|
||
}
|
||
if pool.get("stopping").and_then(serde_json::Value::as_bool) == Some(true) {
|
||
return Err(format!("rebalance entered stopping state: {status}").into());
|
||
}
|
||
if pool
|
||
.get("lastError")
|
||
.and_then(serde_json::Value::as_str)
|
||
.is_some_and(|error| !error.is_empty())
|
||
{
|
||
return Err(format!("rebalance reported lastError: {status}").into());
|
||
}
|
||
if nonzero_u64(pool.get("cleanupWarnings").and_then(|warnings| warnings.get("count"))) {
|
||
return Err(format!("rebalance reported cleanup warnings: {status}").into());
|
||
}
|
||
}
|
||
let propagation_failed = status.get("stopPropagation").is_some_and(|propagation| {
|
||
propagation
|
||
.get("failedPeers")
|
||
.and_then(serde_json::Value::as_array)
|
||
.is_some_and(|peers| !peers.is_empty())
|
||
|| propagation
|
||
.get("terminalReloadFailedPeers")
|
||
.and_then(serde_json::Value::as_array)
|
||
.is_some_and(|peers| !peers.is_empty())
|
||
|| propagation.get("pendingTerminalReload").and_then(serde_json::Value::as_bool) == Some(true)
|
||
});
|
||
if propagation_failed {
|
||
return Err(format!("rebalance stop propagation is incomplete: {status}").into());
|
||
}
|
||
Ok(pools)
|
||
}
|
||
|
||
pub(crate) fn rebalance_active(status: &serde_json::Value, expected_id: &str) -> TestResult<bool> {
|
||
Ok(validate_rebalance_status(status, expected_id)?.iter().any(|pool| {
|
||
pool.get("status")
|
||
.and_then(serde_json::Value::as_str)
|
||
.is_some_and(|value| value.eq_ignore_ascii_case("started"))
|
||
}))
|
||
}
|
||
|
||
pub(crate) fn rebalance_running_with_progress(status: &serde_json::Value, expected_id: &str) -> TestResult<bool> {
|
||
Ok(validate_rebalance_status(status, expected_id)?.iter().any(|pool| {
|
||
let started = pool
|
||
.get("status")
|
||
.and_then(serde_json::Value::as_str)
|
||
.is_some_and(|value| value.eq_ignore_ascii_case("started"));
|
||
let progress = pool.get("progress");
|
||
started
|
||
&& (nonzero_u64(progress.and_then(|value| value.get("objects")))
|
||
|| nonzero_u64(progress.and_then(|value| value.get("versions")))
|
||
|| nonzero_u64(progress.and_then(|value| value.get("bytes"))))
|
||
}))
|
||
}
|
||
|
||
pub(crate) fn rebalance_complete(status: &serde_json::Value, expected_id: &str) -> TestResult<bool> {
|
||
let pools = validate_rebalance_status(status, expected_id)?;
|
||
let completed: Vec<&serde_json::Value> = pools
|
||
.iter()
|
||
.filter(|pool| {
|
||
pool.get("status")
|
||
.and_then(serde_json::Value::as_str)
|
||
.is_some_and(|value| value.eq_ignore_ascii_case("completed"))
|
||
})
|
||
.collect();
|
||
let all_terminal = pools.iter().all(|pool| {
|
||
pool.get("status")
|
||
.and_then(serde_json::Value::as_str)
|
||
.is_some_and(|value| value.eq_ignore_ascii_case("completed") || value.eq_ignore_ascii_case("none"))
|
||
});
|
||
let any_progress = completed.iter().any(|pool| {
|
||
let progress = pool.get("progress");
|
||
nonzero_u64(progress.and_then(|value| value.get("objects")))
|
||
|| nonzero_u64(progress.and_then(|value| value.get("versions")))
|
||
|| nonzero_u64(progress.and_then(|value| value.get("bytes")))
|
||
});
|
||
Ok(all_terminal && !completed.is_empty() && any_progress)
|
||
}
|
||
|
||
pub(crate) async fn wait_for_rebalance_active(
|
||
cluster: &RustFSTestClusterEnvironment,
|
||
expected_id: &str,
|
||
timeout: Duration,
|
||
) -> TestResult {
|
||
let deadline = Instant::now() + timeout;
|
||
loop {
|
||
let status = rebalance_status_json(cluster).await?;
|
||
if rebalance_active(&status, expected_id)? {
|
||
return Ok(());
|
||
}
|
||
if Instant::now() >= deadline {
|
||
return Err(format!("rebalance did not become active within {timeout:?}; last status: {status}").into());
|
||
}
|
||
sleep(Duration::from_secs(1)).await;
|
||
}
|
||
}
|
||
|
||
pub(crate) async fn wait_for_rebalance_running_with_progress(
|
||
cluster: &RustFSTestClusterEnvironment,
|
||
expected_id: &str,
|
||
timeout: Duration,
|
||
) -> TestResult {
|
||
let deadline = Instant::now() + timeout;
|
||
loop {
|
||
let status = rebalance_status_json(cluster).await?;
|
||
if rebalance_running_with_progress(&status, expected_id)? {
|
||
return Ok(());
|
||
}
|
||
if Instant::now() >= deadline {
|
||
return Err(format!(
|
||
"rebalance did not become active with non-zero progress within {timeout:?}; last status: {status}"
|
||
)
|
||
.into());
|
||
}
|
||
sleep(Duration::from_millis(100)).await;
|
||
}
|
||
}
|
||
|
||
pub(crate) async fn wait_for_rebalance_complete(
|
||
cluster: &RustFSTestClusterEnvironment,
|
||
expected_id: &str,
|
||
timeout: Duration,
|
||
) -> TestResult {
|
||
let deadline = Instant::now() + timeout;
|
||
loop {
|
||
let status = rebalance_status_json(cluster).await?;
|
||
if rebalance_complete(&status, expected_id)? {
|
||
return Ok(());
|
||
}
|
||
if Instant::now() >= deadline {
|
||
return Err(
|
||
format!("rebalance did not complete with non-zero progress within {timeout:?}; last status: {status}").into(),
|
||
);
|
||
}
|
||
sleep(Duration::from_secs(1)).await;
|
||
}
|
||
}
|
||
|
||
pub(crate) async fn list_pools_json(cluster: &RustFSTestClusterEnvironment) -> TestResult<serde_json::Value> {
|
||
let body = cluster_admin_ok(cluster, Method::GET, "/rustfs/admin/v3/pools/list", None).await?;
|
||
Ok(serde_json::from_str(&body)?)
|
||
}
|
||
|
||
pub(crate) async fn retrying_put(client: &Client, bucket: &str, key: &str, body: Vec<u8>, timeout: Duration) -> TestResult {
|
||
wait_until(
|
||
timeout,
|
||
|| {
|
||
let client = client.clone();
|
||
let bucket = bucket.to_string();
|
||
let key = key.to_string();
|
||
let body = body.clone();
|
||
async move {
|
||
match put_object(&client, &bucket, &key, body).await {
|
||
Ok(()) => Ok(true),
|
||
Err(error) => {
|
||
let message = error.to_string();
|
||
if message.contains("SlowDown") || message.contains("ServiceUnavailable") || message.contains("503") {
|
||
Ok(false)
|
||
} else {
|
||
Err(error)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
},
|
||
&format!("put {bucket}/{key} during data movement"),
|
||
)
|
||
.await
|
||
}
|
||
|
||
pub(crate) async fn retrying_get_equals(
|
||
client: &Client,
|
||
bucket: &str,
|
||
key: &str,
|
||
expected: &[u8],
|
||
timeout: Duration,
|
||
) -> TestResult {
|
||
wait_until(
|
||
timeout,
|
||
|| async {
|
||
match get_object_bytes(client, bucket, key).await {
|
||
Ok(got) if got.as_slice() == expected => Ok(true),
|
||
Ok(_) => Ok(false),
|
||
Err(error) => {
|
||
let message = error.to_string();
|
||
if message.contains("NoSuchKey")
|
||
|| message.contains("SlowDown")
|
||
|| message.contains("ServiceUnavailable")
|
||
|| message.contains("503")
|
||
{
|
||
Ok(false)
|
||
} else {
|
||
Err(error)
|
||
}
|
||
}
|
||
}
|
||
},
|
||
&format!("get {bucket}/{key} during data movement"),
|
||
)
|
||
.await
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn append_single_node_pool_extends_ellipses_volumes() {
|
||
let mut env = RustFSTestClusterEnvironment::with_topology(ClusterTopology::per_node_pools(2, vec![vec![0], vec![1]]))
|
||
.await
|
||
.expect("two-pool seed topology");
|
||
assert_eq!(env.rustfs_volumes_arg().split(' ').count(), 2);
|
||
|
||
let added = env.append_single_node_pool().await.expect("append third pool");
|
||
assert_eq!(added, 2);
|
||
assert_eq!(env.nodes.len(), 3);
|
||
assert_eq!(env.nodes[2].pool_idx, 2);
|
||
assert_eq!(env.nodes[2].data_dirs.len(), 2);
|
||
let volumes = env.rustfs_volumes_arg();
|
||
assert_eq!(volumes.split(' ').count(), 3, "expected three pool arguments, got: {volumes}");
|
||
assert!(volumes.contains("/drive{0...1}"), "expanded layout must keep drive ellipses: {volumes}");
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn append_single_node_pool_rejects_striped_single_pool() {
|
||
let mut env = RustFSTestClusterEnvironment::new(4).await.expect("four-node single pool");
|
||
let err = env
|
||
.append_single_node_pool()
|
||
.await
|
||
.expect_err("a striped single pool cannot gain a localhost pool");
|
||
let message = err.to_string();
|
||
assert!(
|
||
message.contains("drives_per_node") || message.contains("one node per pool"),
|
||
"unexpected error: {message}"
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn pool_storage_roots_require_exactly_four_unique_paths() {
|
||
let temp_root = std::env::temp_dir();
|
||
let too_few = std::env::join_paths([temp_root.as_path()]).expect("join one path");
|
||
let error = validate_pool_storage_roots(&too_few).expect_err("one pool root must be rejected");
|
||
assert!(error.to_string().contains("exactly 4 paths"));
|
||
|
||
let duplicates = std::env::join_paths([
|
||
temp_root.as_path(),
|
||
temp_root.as_path(),
|
||
temp_root.as_path(),
|
||
temp_root.as_path(),
|
||
])
|
||
.expect("join duplicate paths");
|
||
let error = validate_pool_storage_roots(&duplicates).expect_err("duplicate pool roots must be rejected");
|
||
assert!(error.to_string().contains("duplicate path"));
|
||
}
|
||
|
||
#[cfg(unix)]
|
||
#[tokio::test]
|
||
async fn pool_storage_roots_reject_distinct_paths_on_the_same_device() {
|
||
let env = RustFSTestClusterEnvironment::new(NODE_COUNT)
|
||
.await
|
||
.expect("create same-device pool root fixture");
|
||
let roots: Vec<PathBuf> = env.nodes.iter().map(|node| PathBuf::from(&node.data_dir)).collect();
|
||
let joined = std::env::join_paths(&roots).expect("join same-device paths");
|
||
let error = validate_pool_storage_roots(&joined).expect_err("same-device pool roots must be rejected");
|
||
assert!(error.to_string().contains("distinct filesystems"), "unexpected error: {error}");
|
||
}
|
||
|
||
#[test]
|
||
fn decommission_complete_requires_terminal_status_and_clean_counters() {
|
||
let status = serde_json::json!({
|
||
"pools": [
|
||
{
|
||
"id": 0,
|
||
"status": "complete",
|
||
"poolStatus": "decommissioned",
|
||
"decommissionInfo": {
|
||
"complete": true,
|
||
"failed": false,
|
||
"canceled": false,
|
||
"objectsDecommissioned": 1,
|
||
"bytesDecommissioned": 1024
|
||
}
|
||
},
|
||
{ "id": 1, "status": "none", "poolStatus": "active" }
|
||
]
|
||
});
|
||
assert!(decommission_complete(&status, 0).expect("complete fixture should be accepted"));
|
||
assert!(decommission_complete(&status, 1).is_err());
|
||
|
||
for missing_counter in ["objectsDecommissioned", "bytesDecommissioned"] {
|
||
let mut one_sided = status.clone();
|
||
one_sided["pools"][0]["decommissionInfo"][missing_counter] = serde_json::json!(0);
|
||
assert!(
|
||
!decommission_complete(&one_sided, 0).expect("one-sided progress fixture should be readable"),
|
||
"completion must require both movement counters; zeroed {missing_counter}"
|
||
);
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn decommission_overlap_requires_running_state_and_progress() {
|
||
let mut status = serde_json::json!({
|
||
"pools": [{
|
||
"id": 0,
|
||
"status": "queued",
|
||
"poolStatus": "active",
|
||
"decommissionInfo": {
|
||
"queued": true,
|
||
"objectsDecommissioned": 1,
|
||
"bytesDecommissioned": 1024
|
||
}
|
||
}]
|
||
});
|
||
assert!(
|
||
!decommission_running_with_progress(&status, 0).expect("queued fixture should be readable"),
|
||
"queued work is not temporal overlap"
|
||
);
|
||
status["pools"][0]["status"] = serde_json::json!("running");
|
||
assert!(decommission_running_with_progress(&status, 0).expect("running fixture should be readable"));
|
||
status["pools"][0]["decommissionInfo"]["objectsDecommissioned"] = serde_json::json!(0);
|
||
status["pools"][0]["decommissionInfo"]["bytesDecommissioned"] = serde_json::json!(0);
|
||
assert!(
|
||
!decommission_running_with_progress(&status, 0).expect("zero-progress fixture should be readable"),
|
||
"running state alone does not prove movement started"
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn rebalance_active_treats_started_as_in_progress() {
|
||
let started = serde_json::json!({ "id": "run-1", "pools": [{ "id": 0, "status": "Started", "stopping": false }] });
|
||
let done = serde_json::json!({ "id": "run-1", "pools": [{ "id": 0, "status": "Completed", "stopping": false }] });
|
||
assert!(rebalance_active(&started, "run-1").unwrap());
|
||
assert!(!rebalance_active(&done, "run-1").unwrap());
|
||
}
|
||
|
||
#[test]
|
||
fn rebalance_overlap_requires_started_state_and_progress() {
|
||
let mut status = serde_json::json!({
|
||
"id": "run-1",
|
||
"pools": [{ "id": 0, "status": "Started", "stopping": false, "progress": { "objects": 0, "bytes": 0 } }]
|
||
});
|
||
assert!(
|
||
!rebalance_running_with_progress(&status, "run-1").expect("zero-progress fixture should be readable"),
|
||
"started state alone does not prove movement"
|
||
);
|
||
status["pools"][0]["progress"]["objects"] = serde_json::json!(1);
|
||
assert!(rebalance_running_with_progress(&status, "run-1").expect("progress fixture should be readable"));
|
||
status["pools"][0]["status"] = serde_json::json!("Completed");
|
||
assert!(
|
||
!rebalance_running_with_progress(&status, "run-1").expect("completed fixture should be readable"),
|
||
"completed movement is not temporal overlap"
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn rebalance_complete_accepts_non_participating_pools_but_requires_progress() {
|
||
let completed = serde_json::json!({
|
||
"id": "run-1",
|
||
"pools": [
|
||
{ "id": 0, "status": "Completed", "stopping": false, "progress": { "objects": 2, "bytes": 1024 } },
|
||
{ "id": 1, "status": "None", "stopping": false, "progress": null },
|
||
{ "id": 2, "status": "None", "stopping": false, "progress": null },
|
||
{ "id": 3, "status": "None", "stopping": false, "progress": null }
|
||
]
|
||
});
|
||
assert!(rebalance_complete(&completed, "run-1").unwrap());
|
||
|
||
let no_movement = serde_json::json!({
|
||
"id": "run-1",
|
||
"pools": [
|
||
{ "id": 0, "status": "None", "stopping": false, "progress": null },
|
||
{ "id": 1, "status": "None", "stopping": false, "progress": null }
|
||
]
|
||
});
|
||
assert!(!rebalance_complete(&no_movement, "run-1").unwrap());
|
||
}
|