gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(e2e_test): add automated cluster environment for conditional PUT race test (#1673)

Browse Source 
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: loverustfs <hello@rustfs.com>
This commit is contained in:
LeonWang0735
2026-02-06 15:01:59 +08:00
committed by GitHub
parent 6eb90e7df9
commit a2b88a79ec
3 changed files with 517 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/e2e_test/src/cluster_concurrency_test.rs
+241
View File
@@ -0,0 +1,241 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::common::RustFSTestClusterEnvironment;
use aws_sdk_s3::Client;
use aws_sdk_s3::error::SdkError;
use bytes::Bytes;
use serial_test::serial;
use std::sync::Arc;
use tokio::sync::Barrier;
use tracing::{info, warn};
const BUCKET: &str = "conditional-put-race-bucket";
async fn cleanup_object(client: &Client, key: &str) {
if let Err(e) = client.delete_object().bucket(BUCKET).key(key).send().await {
warn!("Failed to delete object '{}' from bucket '{}' during cleanup: {:?}", key, BUCKET, e);
}
}
async fn conditional_put(
client: &Client,
key: &str,
data: &[u8],
client_id: usize,
) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
let result = client
.put_object()
.bucket(BUCKET)
.key(key)
.body(Bytes::copy_from_slice(data).into())
.if_none_match("*")
.send()
.await;
match result {
Ok(resp) => {
info!(" Client {} SUCCEEDED - ETag: {}", client_id, resp.e_tag().unwrap_or("none"));
Ok(true)
}
Err(SdkError::ServiceError(e)) => {
let code = e.err().meta().code().unwrap_or("");
let message = e.err().meta().message().unwrap_or("");
if code == "PreconditionFailed" {
warn!(" Client {} got 412 PreconditionFailed", client_id);
Ok(false)
} else {
warn!(" Client {} got ServiceError: code={}, message={}", client_id, code, message);
Err(e.into_err().into())
}
}
Err(e) => {
warn!(" Client {} got non-ServiceError: {:?}", client_id, e);
Err(e.into())
}
}
}
async fn run_race_iteration(
clients: &[Client],
test_key: &str,
iteration: usize,
) -> Result<usize, Box<dyn std::error::Error + Send + Sync>> {
cleanup_object(&clients[0], test_key).await;
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
let head_result = clients[0].head_object().bucket(BUCKET).key(test_key).send().await;
if head_result.is_ok() {
warn!("Warning: Object still exists after cleanup, skipping iteration {}", iteration);
return Ok(0);
}
info!("\n=== Iteration {} ===", iteration);
info!("Launching {} concurrent conditional PUTs to different nodes...", clients.len());
let barrier = Arc::new(Barrier::new(clients.len()));
let test_key = test_key.to_string();
let mut handles = vec![];
for (i, client) in clients.iter().enumerate() {
let client = client.clone();
let barrier = barrier.clone();
let key = test_key.clone();
let data = format!("data from client {}", i).into_bytes();
let handle = tokio::spawn(async move {
barrier.wait().await;
conditional_put(&client, &key, &data, i).await
});
handles.push(handle);
}
let mut success_count = 0;
let mut had_error = false;
for handle in handles {
match handle.await {
Ok(Ok(true)) => success_count += 1,
Ok(Ok(false)) => {}
Ok(Err(e)) => {
had_error = true;
info!(" Error: {}", e);
}
Err(e) => {
had_error = true;
info!(" Task error: {}", e);
}
}
}
info!("Result: {} out of {} succeeded", success_count, clients.len());
if success_count > 1 {
info!(">>> RACE CONDITION DETECTED!");
} else if success_count == 1 {
info!(">>> Correct behavior: exactly 1 writer succeeded.");
} else if had_error {
return Err("all conditional PUTs failed (e.g. cluster/bucket not ready)".into());
} else {
info!(">>> Unexpected: no writers succeeded.");
}
Ok(success_count)
}
#[tokio::test]
#[serial]
async fn test_conditional_put_race_cluster() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
crate::common::init_logging();
info!("Starting conditional PUT race test with auto cluster");
let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
cluster.start().await?;
cluster.create_test_bucket(BUCKET).await?;
let clients = cluster.create_all_clients()?;
let iterations = 5;
let mut races_detected = 0;
let mut correct_count = 0;
let mut error_count = 0;
for i in 1..=iterations {
let test_key = format!("race-test-{}-{}", std::process::id(), i);
match run_race_iteration(&clients, &test_key, i).await {
Ok(success_count) => {
if success_count > 1 {
races_detected += 1;
} else if success_count == 1 {
correct_count += 1;
}
}
Err(e) => {
error_count += 1;
warn!("Iteration {} failed (not a race; e.g. cluster/network): {}", i, e)
}
}
cleanup_object(&clients[0], &test_key).await;
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
}
info!("\n====================================");
info!("SUMMARY");
info!("====================================");
info!("Total iterations: {}", iterations);
info!("Correct (1 winner): {}", correct_count);
info!("Race conditions: {}", races_detected);
info!("Errors (skipped): {}", error_count);
assert_eq!(races_detected, 0, "Race conditions detected: {}/{}", races_detected, iterations);
assert_eq!(
error_count, 0,
"{} iteration(s) failed due to errors (e.g. cluster not ready)",
error_count
);
Ok(())
}
#[tokio::test]
#[serial]
async fn test_conditional_put_basic_cluster() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
crate::common::init_logging();
info!("Starting basic conditional PUT test with auto cluster");
let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
cluster.start().await?;
cluster.create_test_bucket(BUCKET).await?;
let client = cluster.create_s3_client(0)?;
let test_key = "basic-conditional-put";
cleanup_object(&client, test_key).await;
let result = client
.put_object()
.bucket(BUCKET)
.key(test_key)
.body(Bytes::from("first write").into())
.if_none_match("*")
.send()
.await;
assert!(result.is_ok(), "First PUT with If-None-Match:* should succeed");
let result = client
.put_object()
.bucket(BUCKET)
.key(test_key)
.body(Bytes::from("second write").into())
.if_none_match("*")
.send()
.await;
assert!(result.is_err(), "Second PUT with If-None-Match:* should fail");
assert!(
matches!(result, Err(SdkError::ServiceError(_))),
"Expected ServiceError but got different error type"
);
if let Err(SdkError::ServiceError(e)) = result {
let code = e.err().meta().code().unwrap_or("");
assert_eq!(code, "PreconditionFailed");
}
cleanup_object(&client, test_key).await;
Ok(())
}
crates/e2e_test/src/common.rs
+273
View File
@@ -374,3 +374,276 @@ pub async fn awscurl_delete(
) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
execute_awscurl(url, "DELETE", None, access_key, secret_key).await
}
/// Represents a single RustFS server instance in a test cluster.
///
/// Each `ClusterNode` tracks the node's network address, base URL for
/// S3-compatible requests, on-disk data directory, and the underlying
/// child process handle when the node is running.
pub struct ClusterNode {
pub address: String,
pub url: String,
pub data_dir: String,
pub process: Option<Child>,
}
/// Test environment for managing a multi-node RustFS cluster.
///
/// `RustFSTestClusterEnvironment` is responsible for starting and stopping
/// a group of `ClusterNode`s, managing their temporary storage directory,
/// and providing the shared access and secret keys used by tests to
/// interact with the cluster.
pub struct RustFSTestClusterEnvironment {
pub nodes: Vec<ClusterNode>,
pub temp_dir: String,
pub access_key: String,
pub secret_key: String,
}
impl RustFSTestClusterEnvironment {
/// Create a new RustFS test cluster environment with the specified number of nodes.
///
/// Generates a unique temporary root directory for the cluster, allocates an available TCP port
/// for each node, creates an independent data directory for every node, and initializes basic
/// cluster node configurations (node processes are not started at this stage).
///
/// # Arguments
///
/// * `node_count` - The number of nodes to create in the cluster, must be a positive integer
/// (an empty cluster will cause errors in subsequent startup operations).
///
/// # Returns
///
/// * `Ok(Self)` - A new instance of `RustFSTestClusterEnvironment` with initialized node
/// configurations and temporary directory info on success.
/// * `Err(Box<dyn Error + Send + Sync>)` - An error if any step fails, such as temporary
/// directory creation failure or available port lookup failure.
pub async fn new(node_count: usize) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
if node_count == 0 {
return Err("Node count must be greater than zero".into());
}
let temp_dir = format!("/tmp/rustfs_cluster_test_{}", Uuid::new_v4());
fs::create_dir_all(&temp_dir).await?;
let mut nodes = Vec::with_capacity(node_count);
for i in 0..node_count {
let port = RustFSTestEnvironment::find_available_port().await?;
let address = format!("127.0.0.1:{}", port);
let url = format!("http://{}", address);
let data_dir = format!("{}/node{}", temp_dir, i);
fs::create_dir_all(&data_dir).await?;
nodes.push(ClusterNode {
address,
url,
data_dir,
process: None,
});
}
Ok(Self {
nodes,
temp_dir,
access_key: DEFAULT_ACCESS_KEY.to_string(),
secret_key: DEFAULT_SECRET_KEY.to_string(),
})
}
/// Build the volumes argument string for RustFS binary (internal helper method).
///
/// Concatenates the address and data directory of all cluster nodes into a single string
/// used as the `RUSTFS_VOLUMES` environment variable for RustFS node processes.
fn build_volumes_arg(&self) -> String {
self.nodes
.iter()
.map(|n| format!("http://{}{}", n.address, n.data_dir))
.collect::<Vec<_>>()
.join(" ")
}
/// Start all node processes in the RustFS cluster and wait for the cluster service to be ready.
///
/// Spawns a RustFS binary process for each node with necessary environment variable configurations,
/// first waits for each node's TCP port to be reachable, then verifies the cluster's S3-compatible
/// service availability via the S3 API.
///
/// # Returns
///
/// * `Ok(())` - All nodes start successfully and the cluster S3 service is ready for requests.
/// * `Err(Box<dyn Error + Send + Sync>)` - An error if process spawning fails, TCP port readiness
/// times out, or cluster service readiness times out.
pub async fn start(&mut self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let binary_path = rustfs_binary_path();
let volumes_arg = self.build_volumes_arg();
for (i, node) in self.nodes.iter_mut().enumerate() {
info!("Starting cluster node {} on {}", i, node.address);
let process = Command::new(&binary_path)
.env("RUSTFS_VOLUMES", &volumes_arg)
.env("RUSTFS_ADDRESS", &node.address)
.env("RUSTFS_ACCESS_KEY", &self.access_key)
.env("RUSTFS_SECRET_KEY", &self.secret_key)
.env("RUSTFS_CONSOLE_ENABLE", "false")
.current_dir(&node.data_dir)
.spawn()?;
node.process = Some(process);
}
for (i, node) in self.nodes.iter().enumerate() {
self.wait_for_node_ready(&node.address, i).await?;
}
self.wait_for_service_ready().await?;
Ok(())
}
/// Wait for a single cluster node's TCP port to become reachable (internal helper method).
///
/// Attempts to establish a TCP connection to the node's address, retries up to 60 times
/// with a 1-second interval between attempts. Fails if the port is unreachable after all retries.
async fn wait_for_node_ready(&self, address: &str, idx: usize) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
for attempt in 0..60 {
if TcpStream::connect(address).await.is_ok() {
info!("Node {} ({}) TCP ready after {} attempts", idx, address, attempt + 1);
return Ok(());
}
sleep(Duration::from_secs(1)).await;
}
Err(format!("Node {} failed to become ready", idx).into())
}
/// Wait for the entire cluster's S3-compatible service to be ready (internal helper method).
///
/// Verifies service availability by calling the S3 `list_buckets` API, retries up to 120 times
/// with a 1-second interval between attempts. Fails if the API call remains unsuccessful after all retries.
async fn wait_for_service_ready(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let client = self.create_s3_client(0)?;
for attempt in 0..120 {
match client.list_buckets().send().await {
Ok(_) => {
info!("Cluster service ready after {} attempts", attempt + 1);
return Ok(());
}
Err(_) => {
sleep(Duration::from_secs(1)).await;
}
}
}
Err("Cluster service failed to become ready".into())
}
/// Create an S3 client configured to communicate with a specific cluster node.
///
/// Configures the S3 client with the cluster's authentication credentials, a fixed `us-east-1` region,
/// the target node's endpoint URL, and enforces path-style access (required for RustFS S3 compatibility).
/// Performs a validity check on the node index before creating the client to avoid out-of-bounds errors.
///
/// # Arguments
///
/// * `node_idx` - The zero-based index of the target cluster node. Must be in the range `[0, total_nodes - 1]`.
///
/// # Returns
///
/// * `Ok(Client)` - A fully configured AWS S3 `Client` instance for the specified node on success.
/// * `Err(Box<dyn Error + Send + Sync>)` - An error if the node index is invalid, or if the S3 client configuration fails.
pub fn create_s3_client(&self, node_idx: usize) -> Result<Client, Box<dyn std::error::Error + Send + Sync>> {
if node_idx >= self.nodes.len() {
return Err("node_idx is invalid".into());
}
let credentials = Credentials::new(&self.access_key, &self.secret_key, None, None, "cluster-test");
let config = Config::builder()
.credentials_provider(credentials)
.region(Region::new("us-east-1"))
.endpoint_url(&self.nodes[node_idx].url)
.force_path_style(true)
.behavior_version_latest()
.build();
Ok(Client::from_conf(config))
}
/// Create S3 clients for all nodes in the RustFS cluster and collect them into a vector.
///
/// Iterates over all cluster node indices, calls `create_s3_client` for each index, and aggregates
/// the resulting clients into a pre-allocated vector. Terminates immediately and returns an error
/// if any single node's S3 client creation fails (fails fast behavior).
///
/// # Returns
///
/// * `Ok(Vec<Client>)` - A vector of configured S3 `Client` instances (one per cluster node) on full success.
/// * `Err(Box<dyn Error + Send + Sync>)` - An error with a descriptive message if any client creation fails,
/// including the underlying error from `create_s3_client`.
pub fn create_all_clients(&self) -> Result<Vec<Client>, Box<dyn std::error::Error + Send + Sync>> {
(0..self.nodes.len()).map(|i| self.create_s3_client(i)).try_fold(
Vec::with_capacity(self.nodes.len()),
|mut clients, result| match result {
Ok(client) => {
clients.push(client);
Ok(clients)
}
Err(e) => Err(format!("Failed to create S3 client for node: {}", e).into()),
},
)
}
/// Create a test S3 bucket in the RustFS cluster.
///
/// Uses the S3 client of the first cluster node to call the S3 `create_bucket` API and
/// create a bucket with the specified name (follows S3 bucket naming conventions).
///
/// # Arguments
///
/// * `bucket_name` - The name of the bucket to create, must comply with S3 bucket naming
/// rules (lowercase, no spaces, valid characters only).
///
/// # Returns
///
/// * `Ok(())` - The test bucket is created successfully via the S3 API.
/// * `Err(Box<dyn Error + Send + Sync>)` - An error if the S3 `create_bucket` API call fails,
/// such as invalid bucket name, insufficient permissions, or an unready cluster.
pub async fn create_test_bucket(&self, bucket_name: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let client = self.create_s3_client(0)?;
client.create_bucket().bucket(bucket_name).send().await?;
info!("Created test bucket: {}", bucket_name);
Ok(())
}
/// Stop all running node processes in the RustFS cluster.
///
/// Iterates over all cluster nodes, attempts to kill the spawned RustFS process (if running),
/// and waits for the process to exit. Logs an error if process termination or waiting fails,
/// but does not panic (fails gracefully).
///
/// This method is automatically called by the `Drop` trait when the cluster environment
/// is destroyed, and can also be called manually to stop the cluster early.
pub fn stop(&mut self) {
for (i, node) in self.nodes.iter_mut().enumerate() {
if let Some(mut process) = node.process.take() {
info!("Stopping cluster node {}", i);
if let Err(e) = process.kill() {
error!("Failed to kill cluster node {}: {}", i, e);
}
if let Err(e) = process.wait() {
error!("Failed to wait for cluster node {} to exit: {}", i, e);
}
}
}
}
}
impl Drop for RustFSTestClusterEnvironment {
/// Clean up the RustFS test cluster environment when the instance is dropped.
///
/// Automatically calls the `stop` method to terminate all running node processes, then
/// attempts to delete the cluster's temporary root directory and all its contents.
/// Logs a warning if directory deletion fails (does not affect program exit).
fn drop(&mut self) {
self.stop();
if let Err(e) = std::fs::remove_dir_all(&self.temp_dir) {
warn!("Failed to clean up cluster temp directory {}: {}", self.temp_dir, e);
}
}
}
crates/e2e_test/src/lib.rs
+3
View File
@@ -61,6 +61,9 @@ mod protocols;
#[cfg(test)]
mod object_lock;
#[cfg(test)]
mod cluster_concurrency_test;
// PutObject / MultipartUpload with checksum (Content-MD5, x-amz-checksum-*)
#[cfg(test)]
mod checksum_upload_test;