diff --git a/.github/workflows/e2e-upgrade.yml b/.github/workflows/e2e-upgrade.yml index 27ce41f2d..ed420cd06 100644 --- a/.github/workflows/e2e-upgrade.yml +++ b/.github/workflows/e2e-upgrade.yml @@ -49,8 +49,20 @@ env: UPGRADE_SOURCE_SHA256: 7c789386bf85278f865b8e0d359bf4edb84d5aa408cc3fa54a18c25ca74cd6e7 jobs: - direct-upgrade: - name: Direct upgrade from rc.2 + upgrade: + name: ${{ matrix.name }} + strategy: + fail-fast: false + matrix: + include: + - name: Direct upgrade from rc.2 + cache_key: e2e-direct-upgrade + test: direct_upgrade_from_rc2_preserves_object_contracts + artifact: direct-upgrade + - name: Mixed-version rolling upgrade from rc.2 + cache_key: e2e-mixed-version-upgrade + test: rolling_upgrade_from_rc2_preserves_mixed_version_contracts + artifact: mixed-version-upgrade runs-on: ubuntu-latest timeout-minutes: 60 env: @@ -64,7 +76,7 @@ jobs: - name: Setup Rust environment uses: ./.github/actions/setup with: - cache-shared-key: e2e-direct-upgrade + cache-shared-key: ${{ matrix.cache_key }} cache-save-if: ${{ github.ref == 'refs/heads/main' }} install-build-packaging-tools: "false" @@ -89,17 +101,17 @@ jobs: cargo build --locked -p rustfs --bin rustfs : > target/debug/rustfs.features - - name: Run direct-upgrade compatibility test + - name: Run upgrade compatibility test run: | cargo test --locked -p e2e_test \ - upgrade_compatibility_test::direct_upgrade_from_rc2_preserves_object_contracts \ + "upgrade_compatibility_test::${{ matrix.test }}" \ -- --ignored --exact --nocapture - name: Upload server logs if: always() uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6 with: - name: direct-upgrade-server-logs-${{ github.run_number }} + name: ${{ matrix.artifact }}-server-logs-${{ github.run_number }} path: ${{ runner.temp }}/rustfs-upgrade-logs if-no-files-found: warn retention-days: 14 diff --git a/crates/e2e_test/README.md b/crates/e2e_test/README.md index 334936a68..634df77bb 100644 --- a/crates/e2e_test/README.md +++ b/crates/e2e_test/README.md @@ -169,7 +169,7 @@ the same profile for membership and execution with one nightly worker. | `s3s-e2e` black-box | `e2e-tests` + `e2e-tests-rio-v2` jobs | **Active** (external conformance tool) | | ILM / lifecycle (ignored) | `test-ilm-integration-serial` lane, `-j1` | **Active** (backlog#1148 ilm-1) | | KMS suite | `e2e-full` job, merge queue + main | **Active** | -| Direct upgrade from pinned previous release | `e2e-upgrade.yml`, storage-sensitive PRs + release tags + weekly | **Active** | +| Direct and mixed-version rolling upgrades from pinned previous release | `e2e-upgrade.yml`, storage-sensitive PRs + release tags + weekly | **Active** | | Cluster faults (`e2e-nightly` profile) | consolidated nightly workflow | **Active** (backlog#1149 ci-7) | | Protocols (FTPS/WebDAV/SFTP) | consolidated nightly workflow, serial | **Active** (backlog#1149 ci-7) | | Replication (fast subset) | `e2e-smoke` profile, `e2e-tests` job, every PR | **Active** (backlog#1147 repl-1) | diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index cb942830a..5bd11df59 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -1469,31 +1469,18 @@ impl RustFSTestClusterEnvironment { /// times out, or cluster service readiness times out. pub async fn start(&mut self) -> Result<(), Box> { let binary_path = rustfs_binary_path(); + self.start_with_binary(&binary_path).await + } + + /// Start every cluster node with a specific RustFS binary. + /// + /// Upgrade compatibility tests use this to initialize a cluster with a + /// pinned previous release before replacing nodes with the workspace build. + pub async fn start_with_binary(&mut self, binary_path: &Path) -> Result<(), Box> { 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 mut command = Command::new(&binary_path); - command - .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") - .env("RUST_LOG", "rustfs=info,rustfs_notify=debug"); - - for (key, value) in &self.extra_env { - command.env(key, value); - } - for (key, value) in &self.node_extra_env[i] { - command.env(key, value); - } - capture_command_logs(&mut command, self.node_capture_log_paths[i].as_deref())?; - - let process = command.current_dir(&node.data_dir).spawn()?; - - node.process = Some(process); + for node_idx in 0..self.nodes.len() { + self.spawn_node(node_idx, binary_path, &volumes_arg)?; } for (i, node) in self.nodes.iter().enumerate() { @@ -1509,20 +1496,46 @@ impl RustFSTestClusterEnvironment { /// Start one node process using the cluster's existing volume layout. pub async fn start_node(&mut self, node_idx: usize) -> Result<(), Box> { + let binary_path = rustfs_binary_path(); + self.start_node_from_binary(node_idx, &binary_path).await + } + + /// Start one stopped cluster node with a specific RustFS binary while + /// preserving the cluster's volume layout and that node's data directory. + pub async fn start_node_from_binary( + &mut self, + node_idx: usize, + binary_path: &Path, + ) -> Result<(), Box> { + let volumes_arg = self.build_volumes_arg(); + self.spawn_node(node_idx, binary_path, &volumes_arg)?; + + self.wait_for_node_ready(&self.nodes[node_idx].address, node_idx).await?; + self.wait_for_node_service_ready(node_idx).await?; + Ok(()) + } + + fn spawn_node( + &mut self, + node_idx: usize, + binary_path: &Path, + volumes_arg: &str, + ) -> Result<(), Box> { self.ensure_node_index(node_idx)?; if self.nodes[node_idx].process.is_some() { return Err(format!("cluster node {node_idx} is already running").into()); } + if !binary_path.is_file() { + return Err(format!("RustFS binary does not exist: {}", binary_path.display()).into()); + } - let binary_path = rustfs_binary_path(); - let volumes_arg = self.build_volumes_arg(); let log_path = self.node_capture_log_paths[node_idx].clone(); let node = &mut self.nodes[node_idx]; - info!("Starting cluster node {} on {}", node_idx, node.address); + info!("Starting cluster node {} on {} with {}", node_idx, node.address, binary_path.display()); - let mut command = Command::new(&binary_path); + let mut command = Command::new(binary_path); command - .env("RUSTFS_VOLUMES", &volumes_arg) + .env("RUSTFS_VOLUMES", volumes_arg) .env("RUSTFS_ADDRESS", &node.address) .env("RUSTFS_ACCESS_KEY", &self.access_key) .env("RUSTFS_SECRET_KEY", &self.secret_key) @@ -1539,9 +1552,6 @@ impl RustFSTestClusterEnvironment { let process = command.current_dir(&node.data_dir).spawn()?; node.process = Some(process); - - self.wait_for_node_ready(&self.nodes[node_idx].address, node_idx).await?; - self.wait_for_node_service_ready(node_idx).await?; Ok(()) } diff --git a/crates/e2e_test/src/upgrade_compatibility_test.rs b/crates/e2e_test/src/upgrade_compatibility_test.rs index 3485f9593..da3b716a3 100644 --- a/crates/e2e_test/src/upgrade_compatibility_test.rs +++ b/crates/e2e_test/src/upgrade_compatibility_test.rs @@ -12,14 +12,15 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::common::{RustFSTestEnvironment, init_logging}; +use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging, rustfs_binary_path}; use aws_sdk_s3::Client; use aws_sdk_s3::error::ProvideErrorMetadata; use aws_sdk_s3::primitives::ByteStream; use aws_sdk_s3::types::{ BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, ServerSideEncryption, VersioningConfiguration, }; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; +use tokio::task::JoinSet; type TestResult = Result<(), Box>; @@ -28,6 +29,10 @@ const SSE_MASTER_KEY_ENV: &str = "RUSTFS_SSE_S3_MASTER_KEY"; const SSE_MASTER_KEY: &str = "QkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkI="; const PLAIN_BUCKET: &str = "upgrade-plain-data"; const VERSIONED_BUCKET: &str = "upgrade-versioned-data"; +const MIXED_BUCKET: &str = "upgrade-mixed-version-data"; +const MIXED_NODE_COUNT: usize = 4; +const MULTIPART_WORKERS: usize = 16; +const MULTIPART_UPLOADS_PER_WORKER: usize = 16; fn source_binary() -> Result> { let path = std::env::var_os(SOURCE_BINARY_ENV) @@ -103,6 +108,99 @@ async fn write_multipart(client: &Client, bucket: &str, key: &str, parts: &[Vec< Ok(()) } +fn configure_cluster_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)?; + for node_idx in 0..cluster.nodes.len() { + let path = Path::new(&log_dir).join(format!("mixed-upgrade-node-{node_idx}.log")); + cluster.set_node_capture_log_path(node_idx, path.to_string_lossy().into_owned())?; + } + Ok(()) +} + +async fn write_multipart_load(clients: &[Client], phase: &str) -> Result, Box> { + let mut tasks = JoinSet::new(); + for worker in 0..MULTIPART_WORKERS { + let client = clients[worker % clients.len()].clone(); + let phase = phase.to_string(); + tasks.spawn(async move { + let mut keys = Vec::with_capacity(MULTIPART_UPLOADS_PER_WORKER); + for upload in 0..MULTIPART_UPLOADS_PER_WORKER { + let key = format!("{phase}/multipart/{worker:02}/{upload:02}"); + let part = vec![u8::try_from(worker)?; 64 * 1024]; + write_multipart(&client, MIXED_BUCKET, &key, &[part]).await?; + keys.push(key); + } + Ok::<_, Box>(keys) + }); + } + + let mut keys = Vec::with_capacity(MULTIPART_WORKERS * MULTIPART_UPLOADS_PER_WORKER); + while let Some(result) = tasks.join_next().await { + keys.extend(result??); + } + Ok(keys) +} + +async fn exercise_mixed_cluster( + cluster: &RustFSTestClusterEnvironment, + phase: &str, + current_node: usize, + previous_node: usize, +) -> TestResult { + let clients = cluster.create_all_clients()?; + let current_client = &clients[current_node]; + let previous_client = &clients[previous_node]; + + let current_key = format!("{phase}/written-by-current"); + let current_body = format!("{phase}: current RustFS build").into_bytes(); + current_client + .put_object() + .bucket(MIXED_BUCKET) + .key(¤t_key) + .body(ByteStream::from(current_body.clone())) + .send() + .await?; + assert_eq!(read_object(previous_client, MIXED_BUCKET, ¤t_key, None).await?.1, current_body); + + let previous_key = format!("{phase}/written-by-previous"); + let previous_body = format!("{phase}: previous RustFS release").into_bytes(); + previous_client + .put_object() + .bucket(MIXED_BUCKET) + .key(&previous_key) + .body(ByteStream::from(previous_body.clone())) + .send() + .await?; + assert_eq!(read_object(current_client, MIXED_BUCKET, &previous_key, None).await?.1, previous_body); + + let multipart_keys = write_multipart_load(&clients, phase).await?; + let expected_count = multipart_keys.len() + 2; + for client in [current_client, previous_client] { + let listed = client + .list_objects_v2() + .bucket(MIXED_BUCKET) + .prefix(format!("{phase}/")) + .send() + .await?; + assert_eq!( + listed.contents().len(), + expected_count, + "both RustFS versions must stream the complete mixed-version listing" + ); + } + + let last_multipart_key = format!("{phase}/multipart/{:02}/{:02}", MULTIPART_WORKERS - 1, MULTIPART_UPLOADS_PER_WORKER - 1); + assert_eq!( + read_object(previous_client, MIXED_BUCKET, &last_multipart_key, None).await?.1, + vec![u8::try_from(MULTIPART_WORKERS - 1)?; 64 * 1024] + ); + + Ok(()) +} + #[tokio::test] #[ignore = "requires a pinned previous RustFS release binary"] async fn direct_upgrade_from_rc2_preserves_object_contracts() -> TestResult { @@ -252,3 +350,47 @@ async fn direct_upgrade_from_rc2_preserves_object_contracts() -> TestResult { Ok(()) } + +#[tokio::test] +#[ignore = "requires a pinned previous RustFS release binary"] +async fn rolling_upgrade_from_rc2_preserves_mixed_version_contracts() -> TestResult { + init_logging(); + let previous_binary = source_binary()?; + let current_binary = rustfs_binary_path(); + let mut cluster = RustFSTestClusterEnvironment::new(MIXED_NODE_COUNT).await?; + cluster.set_env("RUST_LOG", "rustfs=warn,rustfs_notify=warn"); + configure_cluster_logs(&mut cluster)?; + cluster.start_with_binary(&previous_binary).await?; + cluster.create_test_bucket(MIXED_BUCKET).await?; + + cluster.stop_node(0)?; + cluster.start_node_from_binary(0, ¤t_binary).await?; + exercise_mixed_cluster(&cluster, "one-current-node", 0, 1).await?; + + for node_idx in [1, 2] { + cluster.stop_node(node_idx)?; + cluster.start_node_from_binary(node_idx, ¤t_binary).await?; + } + exercise_mixed_cluster(&cluster, "one-previous-node", 0, 3).await?; + + cluster.stop_node(3)?; + cluster.start_node_from_binary(3, ¤t_binary).await?; + + for client in cluster.create_all_clients()? { + for phase in ["one-current-node", "one-previous-node"] { + let listed = client + .list_objects_v2() + .bucket(MIXED_BUCKET) + .prefix(format!("{phase}/")) + .send() + .await?; + assert_eq!( + listed.contents().len(), + MULTIPART_WORKERS * MULTIPART_UPLOADS_PER_WORKER + 2, + "the homogeneous current cluster must preserve every object from {phase}" + ); + } + } + + Ok(()) +}