diff --git a/.docker/compose/docker-compose.cluster.local-build.yml b/.docker/compose/docker-compose.cluster.local-build.yml index 6e2208ce9..8f2f5db26 100644 --- a/.docker/compose/docker-compose.cluster.local-build.yml +++ b/.docker/compose/docker-compose.cluster.local-build.yml @@ -30,6 +30,8 @@ services: - RUSTFS_OBS_USE_STDOUT=${RUSTFS_OBS_USE_STDOUT:-false} - RUSTFS_OBS_LOG_STDOUT_ENABLED=${RUSTFS_OBS_LOG_STDOUT_ENABLED:-false} - RUSTFS_ISSUE3031_DIAG_ENABLE=${RUSTFS_ISSUE3031_DIAG_ENABLE:-false} + - RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT:-5} + - RUSTFS_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_LOCK_ACQUIRE_TIMEOUT:-5} - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} extra_hosts: - "host.docker.internal:host-gateway" @@ -60,6 +62,8 @@ services: - RUSTFS_OBS_USE_STDOUT=${RUSTFS_OBS_USE_STDOUT:-false} - RUSTFS_OBS_LOG_STDOUT_ENABLED=${RUSTFS_OBS_LOG_STDOUT_ENABLED:-false} - RUSTFS_ISSUE3031_DIAG_ENABLE=${RUSTFS_ISSUE3031_DIAG_ENABLE:-false} + - RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT:-5} + - RUSTFS_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_LOCK_ACQUIRE_TIMEOUT:-5} - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} extra_hosts: - "host.docker.internal:host-gateway" @@ -90,6 +94,8 @@ services: - RUSTFS_OBS_USE_STDOUT=${RUSTFS_OBS_USE_STDOUT:-false} - RUSTFS_OBS_LOG_STDOUT_ENABLED=${RUSTFS_OBS_LOG_STDOUT_ENABLED:-false} - RUSTFS_ISSUE3031_DIAG_ENABLE=${RUSTFS_ISSUE3031_DIAG_ENABLE:-false} + - RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT:-5} + - RUSTFS_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_LOCK_ACQUIRE_TIMEOUT:-5} - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} extra_hosts: - "host.docker.internal:host-gateway" @@ -120,6 +126,8 @@ services: - RUSTFS_OBS_USE_STDOUT=${RUSTFS_OBS_USE_STDOUT:-false} - RUSTFS_OBS_LOG_STDOUT_ENABLED=${RUSTFS_OBS_LOG_STDOUT_ENABLED:-false} - RUSTFS_ISSUE3031_DIAG_ENABLE=${RUSTFS_ISSUE3031_DIAG_ENABLE:-false} + - RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT:-5} + - RUSTFS_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_LOCK_ACQUIRE_TIMEOUT:-5} - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} extra_hosts: - "host.docker.internal:host-gateway" diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 14b4acd6b..58cb5c467 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -3187,6 +3187,76 @@ mod tests { assert!(!parts.user_defined.contains_key(RUSTFS_MULTIPART_OBJECT_KEY)); } + #[tokio::test] + #[serial] + async fn repeated_upload_part_overwrites_previous_part_state() { + let (_paths, ecstore) = setup_test_env().await; + let bucket = format!("multipart-overwrite-{}", Uuid::new_v4().simple()); + let object = "overwrite/object.txt"; + create_test_bucket(&ecstore, &bucket).await; + + let upload = ecstore + .new_multipart_upload(&bucket, object, &ObjectOptions::default()) + .await + .expect("multipart upload should be created"); + + let mut first = PutObjReader::from_vec(vec![1, 2, 3]); + let first_part = ecstore + .put_object_part(&bucket, object, &upload.upload_id, 1, &mut first, &ObjectOptions::default()) + .await + .expect("first multipart part should be uploaded"); + + let mut second = PutObjReader::from_vec(vec![4, 5, 6, 7]); + let second_part = ecstore + .put_object_part(&bucket, object, &upload.upload_id, 1, &mut second, &ObjectOptions::default()) + .await + .expect("second multipart part should overwrite the previous part"); + + assert_ne!( + first_part.etag, second_part.etag, + "the overwrite path should persist the latest part metadata rather than reusing stale state" + ); + + let parts = ecstore + .list_object_parts( + &bucket, + object, + &upload.upload_id, + None, + crate::set_disk::MAX_PARTS_COUNT, + &ObjectOptions::default(), + ) + .await + .expect("multipart parts should be readable after overwrite"); + + assert_eq!(parts.parts.len(), 1, "only the latest version of part 1 should remain visible"); + assert_eq!(parts.parts[0].part_num, 1); + assert_eq!(parts.parts[0].etag, second_part.etag); + assert_eq!(parts.parts[0].size, second_part.size); + assert_eq!(parts.parts[0].actual_size, second_part.actual_size); + + let completed = ecstore + .complete_multipart_upload( + &bucket, + object, + &upload.upload_id, + vec![crate::store_api::CompletePart { + part_num: 1, + etag: second_part.etag.clone(), + checksum_crc32: None, + checksum_crc32c: None, + checksum_sha1: None, + checksum_sha256: None, + checksum_crc64nvme: None, + }], + &ObjectOptions::default(), + ) + .await + .expect("complete multipart upload should succeed with the latest overwritten part"); + + assert_eq!(completed.size, second_part.size as i64); + } + #[tokio::test] #[serial] async fn cleanup_removes_empty_multipart_sha_dirs() { diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index a47dc6350..f018df5cb 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -175,6 +175,10 @@ impl RemoteDisk { }); } + fn mark_suspect_or_offline(&self, reason: &'static str) -> bool { + self.health.mark_failure(&self.endpoint, reason) + } + /// Enable health monitoring after disk creation. /// Used to defer health checks until after startup format loading completes, /// so that remote peers have time to come online. @@ -202,7 +206,10 @@ impl RemoteDisk { let mut interval = time::interval(get_drive_active_check_interval()); // Perform basic connectivity check - if Self::perform_connectivity_check(&addr).await.is_err() && health.mark_offline(&endpoint, "connectivity_probe_failed") { + let initial_probe_ok = Self::perform_connectivity_check(&addr).await.is_ok(); + if initial_probe_ok { + health.record_operation_success(&endpoint, "connectivity_probe_success"); + } else if health.mark_failure(&endpoint, "connectivity_probe_failed") { warn!("Remote disk health check failed for {}: marking as faulty", addr); // Start recovery monitoring @@ -245,7 +252,9 @@ impl RemoteDisk { } // Perform basic connectivity check - if Self::perform_connectivity_check(&addr).await.is_err() && health.mark_offline(&endpoint, "connectivity_probe_failed") { + if Self::perform_connectivity_check(&addr).await.is_ok() { + health.record_operation_success(&endpoint, "connectivity_probe_success"); + } else if health.mark_failure(&endpoint, "connectivity_probe_failed") { warn!("Remote disk health check failed for {}: marking as faulty", addr); // Start recovery monitoring @@ -286,7 +295,7 @@ impl RemoteDisk { return; } } else { - health.mark_offline(&endpoint, "connectivity_probe_failed"); + health.mark_failure(&endpoint, "connectivity_probe_failed"); } } } @@ -440,7 +449,11 @@ impl RemoteDisk { } async fn mark_faulty_and_evict(&self, reason: &'static str) { - if self.health.mark_offline(&self.endpoint, reason) { + let previous_state = self.runtime_state(); + let became_offline = self.mark_suspect_or_offline(reason); + let state = self.runtime_state(); + + if state != previous_state { self.spawn_recovery_monitor_if_needed(); counter!( "rustfs_drive_faulty_mark_total", @@ -448,10 +461,17 @@ impl RemoteDisk { "reason" => reason.to_string() ) .increment(1); - warn!( - "Remote disk marked faulty after timeout: endpoint={}, addr={}, reason={}", - self.endpoint, self.addr, reason - ); + if became_offline || state == RuntimeDriveHealthState::Offline { + warn!( + "Remote disk marked faulty after timeout: endpoint={}, addr={}, reason={}", + self.endpoint, self.addr, reason + ); + } else { + warn!( + "Remote disk marked suspect after timeout: endpoint={}, addr={}, reason={}, state={:?}", + self.endpoint, self.addr, reason, state + ); + } counter!( "rustfs_drive_connection_evict_total", "endpoint" => self.endpoint.to_string(), @@ -1951,20 +1971,39 @@ mod tests { disk_idx: 0, }; - let disk_option = DiskOption { - cleanup: false, - health_check: true, - }; + temp_env::async_with_vars( + [ + (rustfs_config::ENV_DRIVE_ACTIVE_CHECK_INTERVAL_SECS, Some("1")), + (rustfs_config::ENV_DRIVE_ACTIVE_CHECK_TIMEOUT_SECS, Some("1")), + ], + async { + let disk_option = DiskOption { + cleanup: false, + health_check: true, + }; - let remote_disk = RemoteDisk::new(&endpoint, &disk_option, Arc::new(TcpHttpInternodeDataTransport)) - .await - .unwrap(); - remote_disk.enable_health_check(); + let remote_disk = RemoteDisk::new(&endpoint, &disk_option, Arc::new(TcpHttpInternodeDataTransport)) + .await + .unwrap(); + remote_disk.enable_health_check(); - // wait for health check connect timeout - tokio::time::sleep(Duration::from_secs(6)).await; - - assert!(!remote_disk.is_online().await); + // Wait out the initial success-grace window so the active probe loop + // actually attempts a connectivity check. Under the new + // suspect-first semantics we only need to prove that the drive + // transitions away from a clean Online state at least once. + tokio::time::sleep(SKIP_IF_SUCCESS_BEFORE + Duration::from_secs(2)).await; + assert!( + remote_disk.offline_duration_secs().is_some(), + "missing listener should transition the drive through suspect/offline tracking" + ); + assert_ne!( + remote_disk.runtime_state(), + RuntimeDriveHealthState::Online, + "missing listener should not remain in a clean Online state after probing" + ); + }, + ) + .await; } #[tokio::test] @@ -2284,7 +2323,12 @@ mod tests { .expect_err("timeout should fail"); assert!(err.to_string().contains("timeout")); - assert!(!remote_disk.is_online().await, "remote disk should be marked faulty after timeout"); + assert!(remote_disk.is_online().await, "first timeout should keep the remote disk online"); + assert_eq!( + remote_disk.runtime_state(), + RuntimeDriveHealthState::Suspect, + "first timeout should move the remote disk into suspect state" + ); } #[tokio::test] @@ -2450,7 +2494,15 @@ mod tests { }, std::io::ErrorKind::TimedOut ); - assert!(!remote_disk.is_online().await, "timeout-like errors should mark remote disk faulty"); + assert!( + remote_disk.is_online().await, + "first timeout-like error should keep the remote disk online" + ); + assert_eq!( + remote_disk.runtime_state(), + RuntimeDriveHealthState::Suspect, + "first timeout-like error should move the remote disk into suspect state" + ); assert!( !GLOBAL_CONN_MAP.read().await.contains_key(&addr), "timeout-like errors should evict cached connection" @@ -2503,7 +2555,15 @@ mod tests { }, std::io::ErrorKind::ConnectionRefused ); - assert!(!remote_disk.is_online().await, "network-like errors should mark remote disk faulty"); + assert!( + remote_disk.is_online().await, + "first network-like error should keep the remote disk online" + ); + assert_eq!( + remote_disk.runtime_state(), + RuntimeDriveHealthState::Suspect, + "first network-like error should move the remote disk into suspect state" + ); assert!( !GLOBAL_CONN_MAP.read().await.contains_key(&addr), "network-like errors should evict cached connection" diff --git a/crates/ecstore/src/rpc/remote_locker.rs b/crates/ecstore/src/rpc/remote_locker.rs index 0926e8966..3d6f4f911 100644 --- a/crates/ecstore/src/rpc/remote_locker.rs +++ b/crates/ecstore/src/rpc/remote_locker.rs @@ -25,7 +25,7 @@ use tokio::time::timeout; use tonic::Request; use tonic::service::interceptor::InterceptedService; use tonic::transport::Channel; -use tracing::{info, warn}; +use tracing::{debug, info, warn}; /// Remote lock client implementation #[derive(Debug, Clone)] @@ -64,16 +64,44 @@ impl RemoteClient { .map_err(|err| LockError::internal(format!("can not get client, err: {err}"))) } - async fn evict_connection(&self, op: &'static str, reason: &str) { - warn!( - addr = %self.addr, - op, - reason, - "Evicting cached remote lock connection after RPC failure" - ); + fn is_scanner_leader_lock(resource_summary: &str) -> bool { + resource_summary == ".rustfs.sys/leader.lock@latest" + } + + async fn evict_connection(&self, op: &'static str, reason: &str, resource_summary: &str) { + if Self::is_scanner_leader_lock(resource_summary) { + debug!( + addr = %self.addr, + op, + reason, + resource_summary, + "Evicting cached remote lock connection for scanner leader-lock RPC failure" + ); + } else { + warn!( + addr = %self.addr, + op, + reason, + resource_summary, + "Evicting cached remote lock connection after RPC failure" + ); + } evict_failed_connection(&self.addr).await; } + fn summarize_resources(requests: &[LockRequest]) -> String { + const LIMIT: usize = 3; + let mut resources = requests + .iter() + .take(LIMIT) + .map(|request| request.resource.to_string()) + .collect::>(); + if requests.len() > LIMIT { + resources.push(format!("... (+{} more)", requests.len() - LIMIT)); + } + resources.join(", ") + } + fn rpc_timeout(timeout_duration: Duration) -> Duration { if timeout_duration.is_zero() { Duration::from_millis(1) @@ -86,6 +114,7 @@ impl RemoteClient { &self, op: &'static str, timeout_duration: Duration, + resource_summary: &str, future: F, ) -> std::result::Result where @@ -96,12 +125,50 @@ impl RemoteClient { Ok(Ok(response)) => Ok(response), Ok(Err(err)) => { let reason = err.to_string(); - self.evict_connection(op, &reason).await; + if Self::is_scanner_leader_lock(resource_summary) { + debug!( + addr = %self.addr, + op, + timeout_ms = lock_timeout.as_millis(), + resource_summary, + tonic_code = ?err.code(), + tonic_message = err.message(), + "Remote lock RPC returned tonic error for scanner leader lock" + ); + } else { + warn!( + addr = %self.addr, + op, + timeout_ms = lock_timeout.as_millis(), + resource_summary, + tonic_code = ?err.code(), + tonic_message = err.message(), + "Remote lock RPC returned tonic error" + ); + } + self.evict_connection(op, &reason, resource_summary).await; Err(LockError::internal(format!("{op} RPC failed: {reason}"))) } Err(_) => { let reason = format!("RPC timed out after {:?}", lock_timeout); - self.evict_connection(op, &reason).await; + if Self::is_scanner_leader_lock(resource_summary) { + debug!( + addr = %self.addr, + op, + timeout_ms = lock_timeout.as_millis(), + resource_summary, + "Remote lock RPC timed out for scanner leader lock" + ); + } else { + warn!( + addr = %self.addr, + op, + timeout_ms = lock_timeout.as_millis(), + resource_summary, + "Remote lock RPC timed out" + ); + } + self.evict_connection(op, &reason, resource_summary).await; Err(LockError::timeout(format!("remote lock RPC {op} on {}", self.addr), lock_timeout)) } } @@ -182,12 +249,16 @@ impl LockClient for RemoteClient { async fn acquire_lock(&self, request: &LockRequest) -> Result { info!("remote acquire_exclusive for {}", request.resource); let mut client = self.get_client().await?; + let resource_summary = request.resource.to_string(); let req = Request::new(GenerallyLockRequest { args: serde_json::to_string(&request) .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?, }); - let resp = match self.execute_rpc("lock", request.acquire_timeout, client.lock(req)).await { + let resp = match self + .execute_rpc("lock", request.acquire_timeout, &resource_summary, client.lock(req)) + .await + { Ok(resp) => resp.into_inner(), Err(err @ LockError::Timeout { .. }) => return Ok(Self::rpc_timeout_failure_response(request, &err)), Err(err) => return Ok(Self::rpc_failure_response(request, &err)), @@ -215,6 +286,7 @@ impl LockClient for RemoteClient { } let mut client = self.get_client().await?; + let resource_summary = Self::summarize_resources(requests); let req = Request::new(BatchGenerallyLockRequest { args: requests .iter() @@ -225,7 +297,7 @@ impl LockClient for RemoteClient { }); let resp = match self - .execute_rpc("lock_batch", Self::batch_rpc_timeout(requests), client.lock_batch(req)) + .execute_rpc("lock_batch", Self::batch_rpc_timeout(requests), &resource_summary, client.lock_batch(req)) .await { Ok(resp) => resp.into_inner(), diff --git a/crates/ecstore/src/set_disk/write.rs b/crates/ecstore/src/set_disk/write.rs index 9ce0fe277..db34f6042 100644 --- a/crates/ecstore/src/set_disk/write.rs +++ b/crates/ecstore/src/set_disk/write.rs @@ -262,6 +262,11 @@ impl SetDisks { let dst_bucket = Arc::new(dst_bucket.to_string()); let dst_object = Arc::new(dst_object.to_string()); + // Match MinIO's multipart overwrite semantics: clear any stale destination + // part payload and metadata before the new per-disk rename fan-out begins. + self.cleanup_multipart_path(&[dst_object.to_string(), format!("{dst_object}.meta")]) + .await; + let mut errs = Vec::with_capacity(disks.len()); let futures = disks.iter().map(|disk| { diff --git a/rustfs/src/admin/console.rs b/rustfs/src/admin/console.rs index e651f926e..0b7529c58 100644 --- a/rustfs/src/admin/console.rs +++ b/rustfs/src/admin/console.rs @@ -28,7 +28,6 @@ use http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri}; use mime_guess::from_path; use rust_embed::RustEmbed; use serde::Serialize; -use serde_json::json; use std::{ net::{IpAddr, SocketAddr}, sync::OnceLock, @@ -283,7 +282,7 @@ async fn version_handler() -> impl IntoResponse { .header("content-type", "application/json") .status(StatusCode::OK) .body(Body::from( - json!({ + serde_json::json!({ "version": cfg.release.version, "version_info": cfg.version_info(), "date": cfg.release.date, diff --git a/scripts/validate_issue_3031_docker.sh b/scripts/validate_issue_3031_docker.sh new file mode 100644 index 000000000..2a072f2d9 --- /dev/null +++ b/scripts/validate_issue_3031_docker.sh @@ -0,0 +1,466 @@ +#!/usr/bin/env bash +set -euo pipefail + +PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +CLUSTER_COMPOSE="${CLUSTER_COMPOSE:-${PROJECT_ROOT}/.docker/compose/docker-compose.cluster.local-build.yml}" +PROJECT_NAME="${PROJECT_NAME:-rustfs-issue3031}" +RUSTFS_IMAGE="${RUSTFS_IMAGE:-rustfs/rustfs:local-4node}" +BUILD_LOCAL_IMAGE="${BUILD_LOCAL_IMAGE:-true}" +FORCE_BUILD="${FORCE_BUILD:-false}" +KEEP_UP="${KEEP_UP:-false}" +PRECHECK_AUTO_CLEANUP="${PRECHECK_AUTO_CLEANUP:-true}" +WAIT_TIMEOUT_SECS="${WAIT_TIMEOUT_SECS:-180}" +S3_READY_TIMEOUT_SECS="${S3_READY_TIMEOUT_SECS:-120}" + +RUSTFS_ACCESS_KEY="${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin}" +RUSTFS_SECRET_KEY="${RUSTFS_SECRET_KEY:-rustfs-cluster-secret}" +RUSTFS_UNSAFE_BYPASS_DISK_CHECK="${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true}" +RUSTFS_ISSUE3031_DIAG_ENABLE="${RUSTFS_ISSUE3031_DIAG_ENABLE:-true}" +RUSTFS_OBS_LOG_STDOUT_ENABLED="${RUSTFS_OBS_LOG_STDOUT_ENABLED:-true}" +RUSTFS_OBS_USE_STDOUT="${RUSTFS_OBS_USE_STDOUT:-false}" +RUSTFS_OBS_LOGGER_LEVEL="${RUSTFS_OBS_LOGGER_LEVEL:-info}" +RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT="${RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT:-5}" +RUSTFS_LOCK_ACQUIRE_TIMEOUT="${RUSTFS_LOCK_ACQUIRE_TIMEOUT:-5}" + +ENDPOINT="${ENDPOINT:-http://127.0.0.1:9000}" +WARP_HOST="${WARP_HOST:-node1:9000}" +BUCKET="${BUCKET:-rustfs-multipart-repro}" +SMOKE_OBJECTS="${SMOKE_OBJECTS:-16}" +SMOKE_OBJECT_SIZE_MB="${SMOKE_OBJECT_SIZE_MB:-32}" +SMOKE_PARALLELISM="${SMOKE_PARALLELISM:-8}" + +WARP_DURATION="${WARP_DURATION:-5m}" +WARP_CONCURRENT="${WARP_CONCURRENT:-16}" +WARP_PARTS="${WARP_PARTS:-16}" +WARP_PART_SIZE="${WARP_PART_SIZE:-16MiB}" +WARP_PART_CONCURRENT="${WARP_PART_CONCURRENT:-4}" + +MC_IMAGE="${MC_IMAGE:-minio/mc:latest}" +WARP_IMAGE="${WARP_IMAGE:-minio/warp:latest}" +OUT_DIR="${OUT_DIR:-${PROJECT_ROOT}/target/issue3031/$(date +%Y%m%d-%H%M%S)}" + +usage() { + cat <<'USAGE' +Usage: + scripts/validate_issue_3031_docker.sh [options] + +Options: + --skip-build skip local Docker image build + --force-build rebuild even if the local image already exists + --keep-up keep cluster running after the script exits + --project-name docker compose project name + --out-dir output directory + --warp-duration warp duration (default: 5m) + --warp-concurrent warp --concurrent (default: 16) + --warp-parts warp --parts (default: 16) + --warp-part-size warp --part.size (default: 16MiB) + --warp-part-concurrent warp --part.concurrent (default: 4) + --smoke-objects plain write smoke object count (default: 16) + --smoke-object-size-mb plain write smoke object size MiB (default: 32) + --smoke-parallelism plain write smoke parallelism (default: 8) + -h, --help show help + +Environment: + CLUSTER_COMPOSE PROJECT_NAME RUSTFS_IMAGE BUILD_LOCAL_IMAGE FORCE_BUILD KEEP_UP + RUSTFS_ACCESS_KEY RUSTFS_SECRET_KEY RUSTFS_UNSAFE_BYPASS_DISK_CHECK + RUSTFS_ISSUE3031_DIAG_ENABLE RUSTFS_OBS_LOG_STDOUT_ENABLED + RUSTFS_OBS_USE_STDOUT RUSTFS_OBS_LOGGER_LEVEL + RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT + RUSTFS_LOCK_ACQUIRE_TIMEOUT + S3_READY_TIMEOUT_SECS + ENDPOINT BUCKET + WARP_HOST + SMOKE_OBJECTS SMOKE_OBJECT_SIZE_MB SMOKE_PARALLELISM + WARP_DURATION WARP_CONCURRENT WARP_PARTS WARP_PART_SIZE WARP_PART_CONCURRENT + MC_IMAGE WARP_IMAGE OUT_DIR +USAGE +} + +log_info() { + printf '[INFO] %s\n' "$*" +} + +log_warn() { + printf '[WARN] %s\n' "$*" +} + +log_error() { + printf '[ERROR] %s\n' "$*" >&2 +} + +require_cmd() { + if ! command -v "$1" >/dev/null 2>&1; then + log_error "command not found: $1" + exit 1 + fi +} + +compose() { + docker compose \ + --project-name "${PROJECT_NAME}" \ + -f "${CLUSTER_COMPOSE}" \ + "$@" +} + +cleanup() { + if [[ "${KEEP_UP}" == "true" ]]; then + log_info "KEEP_UP=true, leaving cluster running" + return + fi + + log_info "Stopping docker compose services" + compose down -v >/dev/null 2>&1 || true +} + +parse_args() { + while [[ $# -gt 0 ]]; do + case "$1" in + --skip-build) + BUILD_LOCAL_IMAGE=false + shift + ;; + --keep-up) + KEEP_UP=true + shift + ;; + --force-build) + FORCE_BUILD=true + shift + ;; + --project-name) + PROJECT_NAME="$2" + shift 2 + ;; + --out-dir) + OUT_DIR="$2" + shift 2 + ;; + --warp-duration) + WARP_DURATION="$2" + shift 2 + ;; + --warp-concurrent) + WARP_CONCURRENT="$2" + shift 2 + ;; + --warp-parts) + WARP_PARTS="$2" + shift 2 + ;; + --warp-part-size) + WARP_PART_SIZE="$2" + shift 2 + ;; + --warp-part-concurrent) + WARP_PART_CONCURRENT="$2" + shift 2 + ;; + --smoke-objects) + SMOKE_OBJECTS="$2" + shift 2 + ;; + --smoke-object-size-mb) + SMOKE_OBJECT_SIZE_MB="$2" + shift 2 + ;; + --smoke-parallelism) + SMOKE_PARALLELISM="$2" + shift 2 + ;; + -h|--help) + usage + exit 0 + ;; + *) + log_error "unknown argument: $1" + usage + exit 1 + ;; + esac + done +} + +wait_http_ok() { + local url="$1" + local start now + start="$(date +%s)" + + while true; do + if curl -fsS --connect-timeout 2 --max-time 3 "${url}" >/dev/null 2>&1; then + return 0 + fi + + now="$(date +%s)" + if (( now - start >= WAIT_TIMEOUT_SECS )); then + log_error "timed out waiting for ${url}" + return 1 + fi + + sleep 2 + done +} + +cleanup_existing_project_containers() { + local existing_ids + existing_ids="$(docker ps -aq --filter "label=com.docker.compose.project=${PROJECT_NAME}")" + + if [[ -z "${existing_ids}" ]]; then + return 0 + fi + + log_warn "Found existing containers for project ${PROJECT_NAME}." + docker ps -a --filter "label=com.docker.compose.project=${PROJECT_NAME}" --format ' - {{.Names}} ({{.Status}})' + + if [[ "${PRECHECK_AUTO_CLEANUP}" == "true" ]]; then + log_info "PRECHECK_AUTO_CLEANUP=true, removing existing project containers." + # shellcheck disable=SC2086 + docker rm -f ${existing_ids} >/dev/null + else + log_error "existing project containers detected and PRECHECK_AUTO_CLEANUP=false" + exit 1 + fi +} + +build_image_if_needed() { + if [[ "${BUILD_LOCAL_IMAGE}" != "true" ]]; then + log_info "Skipping local image build" + return + fi + + if [[ "${FORCE_BUILD}" != "true" ]] && docker image inspect "${RUSTFS_IMAGE}" >/dev/null 2>&1; then + log_info "Reusing existing local image ${RUSTFS_IMAGE}; pass --force-build to rebuild" + return + fi + + log_info "Building ${RUSTFS_IMAGE} from Dockerfile.source" + docker build -f "${PROJECT_ROOT}/Dockerfile.source" -t "${RUSTFS_IMAGE}" "${PROJECT_ROOT}" +} + +start_cluster() { + mkdir -p "${OUT_DIR}" + cleanup_existing_project_containers + + log_info "Starting 4-node cluster from ${CLUSTER_COMPOSE}" + compose up -d + + log_info "Waiting for cluster health endpoint" + wait_http_ok "${ENDPOINT}/health" + + log_info "Waiting for cluster readiness endpoint" + wait_http_ok "${ENDPOINT}/health/ready" + + log_info "Waiting for S3 API readiness via mc alias/list" + wait_s3_api_ready +} + +cluster_network() { + echo "${PROJECT_NAME}_rustfs-cluster-net" +} + +run_mc_in_network() { + docker run --rm --network "$(cluster_network)" "${MC_IMAGE}" "$@" +} + +run_mc_shell_in_network() { + docker run --rm --network "$(cluster_network)" --entrypoint /bin/sh "${MC_IMAGE}" -lc "$1" +} + +run_warp_in_network() { + docker run --rm --network "$(cluster_network)" -v "${OUT_DIR}:/out" "${WARP_IMAGE}" "$@" +} + +wait_s3_api_ready() { + local start now + start="$(date +%s)" + + while true; do + if run_mc_in_network alias set rustfs "http://node1:9000" "${RUSTFS_ACCESS_KEY}" "${RUSTFS_SECRET_KEY}" >/dev/null 2>&1; then + if run_mc_in_network ls rustfs >/dev/null 2>&1; then + return 0 + fi + fi + + now="$(date +%s)" + if (( now - start >= S3_READY_TIMEOUT_SECS )); then + log_error "timed out waiting for S3 API readiness via mc alias/list" + return 1 + fi + + sleep 2 + done +} + +collect_cluster_info() { + local node + + for node in node1 node2 node3 node4; do + log_info "Collecting rustfs info from ${node}" + compose exec -T "${node}" rustfs info --all --json > "${OUT_DIR}/${node}-info.json" + done +} + +run_plain_smoke() { + local prefix + prefix="plain-smoke-$(date +%s)" + + log_info "Preparing bucket ${BUCKET}" + run_mc_in_network alias set rustfs "http://node1:9000" "${RUSTFS_ACCESS_KEY}" "${RUSTFS_SECRET_KEY}" >/dev/null + run_mc_in_network ls "rustfs/${BUCKET}" >/dev/null 2>&1 || run_mc_in_network mb "rustfs/${BUCKET}" >/dev/null + + log_info "Running plain smoke write/read/delete: objects=${SMOKE_OBJECTS} size_mb=${SMOKE_OBJECT_SIZE_MB} parallelism=${SMOKE_PARALLELISM}" + run_mc_shell_in_network " + set -euo pipefail + i=1 + while [ \"\$i\" -le \"${SMOKE_OBJECTS}\" ]; do + end=\$((i + ${SMOKE_PARALLELISM} - 1)) + [ \"\$end\" -le \"${SMOKE_OBJECTS}\" ] || end='${SMOKE_OBJECTS}' + j=\"\$i\" + while [ \"\$j\" -le \"\$end\" ]; do + ( + dd if=/dev/urandom bs='${SMOKE_OBJECT_SIZE_MB}M' count=1 2>/dev/null | \ + mc pipe 'rustfs/${BUCKET}/${prefix}/obj-\${j}' >/dev/null + ) & + j=\$((j + 1)) + done + wait + i=\$((end + 1)) + done + mc rm --recursive --force 'rustfs/${BUCKET}/${prefix}' >/dev/null + " +} + +run_warp_multipart() { + log_info "Running warp multipart-put against ${WARP_HOST}" + run_warp_in_network multipart-put \ + --host "${WARP_HOST}" \ + --access-key "${RUSTFS_ACCESS_KEY}" \ + --secret-key "${RUSTFS_SECRET_KEY}" \ + --bucket "${BUCKET}-warp" \ + --lookup path \ + --duration "${WARP_DURATION}" \ + --concurrent "${WARP_CONCURRENT}" \ + --parts "${WARP_PARTS}" \ + --part.size "${WARP_PART_SIZE}" \ + --part.concurrent "${WARP_PART_CONCURRENT}" \ + --benchdata /out/warp.csv.zst \ + --analyze.v \ + --no-color \ + | tee "${OUT_DIR}/warp.log" +} + +count_matches() { + local pattern="$1" + shift + local total=0 + local file count + + for file in "$@"; do + count="$(grep -cE "${pattern}" "${file}" 2>/dev/null || true)" + total=$((total + ${count:-0})) + done + + echo "${total}" +} + +collect_cluster_logs() { + local node + + for node in node1 node2 node3 node4; do + compose logs --no-color "${node}" > "${OUT_DIR}/${node}.log" || true + done +} + +summarize_results() { + local warp_log summary + local warp_errors quorum_not_reached connection_refused storage_insufficient + local lock_timeout erasure_write_quorum readiness_probe_failed liveness_probe_failed + local diag_read_parts diag_complete_part diag_rename_part + local startup_range_oob startup_volume_not_found startup_remote_network startup_remote_faulty startup_remote_lock_rpc + + warp_log="${OUT_DIR}/warp.log" + summary="${OUT_DIR}/summary.txt" + + warp_errors="$(count_matches 'warp: ' "${warp_log}")" + quorum_not_reached="$(count_matches 'Quorum not reached' "${warp_log}")" + connection_refused="$(count_matches 'connection refused' "${warp_log}")" + storage_insufficient="$(count_matches 'Storage resources are insufficient for the write operation' "${warp_log}")" + lock_timeout="$(count_matches 'Lock acquisition timeout' "${warp_log}")" + erasure_write_quorum="$(count_matches 'erasure write quorum' "${warp_log}")" + readiness_probe_failed="$(count_matches 'readiness_probe_failed' "${warp_log}")" + liveness_probe_failed="$(count_matches 'liveness_probe_failed' "${warp_log}")" + + diag_read_parts="$(count_matches 'issue3031_read_parts_part_quorum' "${OUT_DIR}"/node*.log)" + diag_complete_part="$(count_matches 'issue3031_complete_part_error' "${OUT_DIR}"/node*.log)" + diag_rename_part="$(count_matches 'issue3031_rename_part_context' "${OUT_DIR}"/node*.log)" + + startup_range_oob="$(count_matches 'Range \\[0, 4\\) is out of bounds' "${OUT_DIR}"/node*.log)" + startup_volume_not_found="$(count_matches 'volume not found' "${OUT_DIR}"/node*.log)" + startup_remote_network="$(count_matches 'Remote disk operation returned a network-like error' "${OUT_DIR}"/node*.log)" + startup_remote_faulty="$(count_matches 'Remote disk marked faulty after timeout' "${OUT_DIR}"/node*.log)" + startup_remote_lock_rpc="$(count_matches 'Evicting cached remote lock connection after RPC failure' "${OUT_DIR}"/node*.log)" + + cat > "${summary}" <