diff --git a/.config/nextest.toml b/.config/nextest.toml index bebd726db..090b52119 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -178,12 +178,20 @@ test-group = 'ecstore-serial-flaky' # profile; its GHSA-r5qv sibling is a unit test that already runs in the # test-and-lint `--all --exclude e2e_test` pass. See # docs/testing/security-regressions.md for the full CI-execution map. +# +# ILM tiering main path (backlog#1148 ilm-7): the `reliant::tiering::` clause +# admits the hermetic transition e2e. Like the fast replication pair checks it +# spawns a second independent single-node server (the cold RustFS tier), not a +# cluster, so it keeps the lane's parallel-safe / no-external-dependency +# properties. The RustFS warm backend has no loopback guard (that guard is +# replication-only), so it needs no opt-in env for its 127.0.0.1 tier target. [profile.e2e-smoke] default-filter = """ package(e2e_test) & ( test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_source_invalid_date|content_encoding|anonymous_access|bucket_policy_check|presigned_negative|negative_sigv4|admin_auth|notification_webhook|tls_hot_reload)_test::/) | test(/^replication_extension_test::(test_replication_check_succeeds_with_remote_target|test_replication_check_rejects_target_without_object_lock|test_set_remote_target_rejects_unversioned_source_bucket|test_replication_check_rejects_unversioned_source_bucket|test_replication_check_rejects_missing_replication_config|test_replication_check_rejects_invalid_bucket|test_set_remote_target_rejects_same_bucket_on_same_deployment|test_set_remote_target_rejects_unversioned_target_bucket|test_set_remote_target_update_requires_arn|test_set_remote_target_update_rejects_missing_target|test_set_remote_target_rejects_invalid_target_url|test_set_remote_target_rejects_self_signed_https_target_without_skip_tls_verify|test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_pem|test_list_remote_targets_rejects_empty_bucket|test_list_remote_targets_rejects_invalid_bucket|test_remove_remote_target_rejects_missing_target|test_remove_remote_target_rejects_missing_arn|test_remove_remote_target_rejects_invalid_bucket|test_remove_remote_target_rejects_target_used_by_replication|test_delete_bucket_replication_removes_remote_target)$/) | test(/^reliant::lifecycle::/) + | test(/^reliant::tiering::/) ) """ fail-fast = false diff --git a/crates/e2e_test/src/reliant/mod.rs b/crates/e2e_test/src/reliant/mod.rs index f5c131708..ddc42a855 100644 --- a/crates/e2e_test/src/reliant/mod.rs +++ b/crates/e2e_test/src/reliant/mod.rs @@ -22,3 +22,4 @@ mod lifecycle; mod lock; mod node_interact_test; mod sql; +mod tiering; diff --git a/crates/e2e_test/src/reliant/tiering.rs b/crates/e2e_test/src/reliant/tiering.rs new file mode 100644 index 000000000..20de447cb --- /dev/null +++ b/crates/e2e_test/src/reliant/tiering.rs @@ -0,0 +1,380 @@ +#![cfg(test)] +// 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. + +//! Hermetic ILM transition main-path end-to-end test (backlog#1148 ilm-7). +//! +//! Two embedded `rustfs` servers, each with independent credentials, port and +//! data directory: +//! +//! * `cold` — a second RustFS server wired as a [`TierType::RustFS`] remote tier. +//! * `hot` — the source server that transitions objects to `cold`. +//! +//! There are no containers, no external S3 backend and no `awscurl`: the +//! `AddTier` admin call is signed in-process with `rustfs_signer`, exactly like +//! the other admin-API e2e suites in this crate. The RustFS warm backend has no +//! loopback/SSRF restriction (that guard is replication-only), so `hot` can tier +//! to `cold` over `http://127.0.0.1:`. +//! +//! A single test drives the full transition main path and pins the chain +//! required by ilm-7: +//! 1. `AddTier(RustFS)` on `hot` targeting `cold` — the real connectivity / +//! in-use probe runs (no `force`), so this also proves the tier is reachable. +//! 2. A `Transition Days=0` rule installed before a multipart PUT transitions +//! the object immediately (the completion path enqueues it; the 1s scanner +//! cycle is only a backstop). +//! 3. `HEAD` reports `x-amz-storage-class == ` and no `x-amz-restore`. +//! 4. `GET` streams byte-identical data back through the warm backend, and the +//! content-type and user metadata survive the round trip (rustfs#2246). +//! 5. Range `GET` within a part and across the part boundary read the correct +//! bytes from the tier (backlog#807). +//! 6. The remote object is present in the cold-tier bucket after transition. +//! 7. `DeleteObject` on `hot` drives free-version cleanup: the cold-tier copy +//! eventually disappears and the hot object is gone (no local residue). + +use crate::common::{RustFSTestEnvironment, local_http_client}; +use aws_sdk_s3::Client; +use aws_sdk_s3::error::ProvideErrorMetadata; +use aws_sdk_s3::primitives::ByteStream; +use aws_sdk_s3::types::{ + BucketLifecycleConfiguration, CompletedMultipartUpload, CompletedPart, ExpirationStatus, LifecycleRule, LifecycleRuleFilter, + Transition, TransitionStorageClass, +}; +use http::Method; +use http::header::HOST; +use rustfs_signer::constants::UNSIGNED_PAYLOAD; +use rustfs_signer::sign_v4; +use s3s::Body; +use std::time::{Duration as StdDuration, Instant}; + +type TestResult = Result<(), Box>; + +const TIER_NAME: &str = "COLDTIER"; +const TIER_BUCKET: &str = "ilm7-cold-tier"; +const TIER_PREFIX: &str = "tiered"; +const SOURCE_BUCKET: &str = "ilm7-hot"; +const OBJECT_KEY: &str = "tier/report.bin"; +const CONTENT_TYPE: &str = "application/x-ilm7"; +const USER_META_KEY: &str = "ilm7-origin"; +const USER_META_VAL: &str = "hermetic-transition"; + +/// 5 MiB — the S3 minimum size for a non-final multipart part; the object's only +/// internal part boundary sits at this offset. +const PART0_SIZE: usize = 5 * 1024 * 1024; +/// 1 MiB tail so the completed object is genuinely multipart (two parts). +const PART1_SIZE: usize = 1024 * 1024; +const OBJECT_SIZE: usize = PART0_SIZE + PART1_SIZE; + +/// Deterministic, position-dependent payload: adjacent offsets differ, so a +/// misaligned range read is caught. +fn payload() -> Vec { + (0..OBJECT_SIZE).map(|i| (i % 251) as u8).collect() +} + +/// Sign and send an admin request in-process (no `awscurl`). +/// +/// Mirrors the shared admin-API e2e pattern: the SigV4 signature is computed +/// over `UNSIGNED_PAYLOAD`, so the JSON body rides on the wire without being +/// pre-hashed. Returns the response status and body text. +async fn signed_admin_request( + base_url: &str, + method: Method, + path: &str, + body: Option<&str>, + access_key: &str, + secret_key: &str, +) -> Result<(reqwest::StatusCode, String), Box> { + let url = format!("{base_url}{path}"); + let uri = url.parse::()?; + let authority = uri.authority().ok_or("request URL missing authority")?.to_string(); + let body_bytes = body.map(|b| b.as_bytes().to_vec()).unwrap_or_default(); + + let request = http::Request::builder() + .method(method.clone()) + .uri(uri) + .header(HOST, authority) + .header("x-amz-content-sha256", UNSIGNED_PAYLOAD); + let signed = sign_v4(request.body(Body::empty())?, 0, access_key, secret_key, "", "us-east-1"); + + let client = local_http_client(); + let mut request_builder = client.request(method, url.as_str()); + for (name, value) in signed.headers() { + request_builder = request_builder.header(name, value); + } + if !body_bytes.is_empty() { + request_builder = request_builder.body(body_bytes); + } + let response = request_builder.send().await?; + let status = response.status(); + let text = response.text().await?; + Ok((status, text)) +} + +/// Wire `hot` -> `cold` as a `TierType::RustFS` remote tier via `AddTier`. +/// +/// No `force`, so the server runs the real in-use / connectivity probe against +/// `cold` (which requires the tier bucket to already exist there). +async fn add_rustfs_tier(hot: &RustFSTestEnvironment, cold: &RustFSTestEnvironment) -> TestResult { + let body = serde_json::json!({ + "type": "rustfs", + "rustfs": { + "name": TIER_NAME, + "endpoint": cold.url.as_str(), + "accessKey": cold.access_key.as_str(), + "secretKey": cold.secret_key.as_str(), + "bucket": TIER_BUCKET, + "prefix": TIER_PREFIX, + "region": "us-east-1", + "storageClass": "" + } + }) + .to_string(); + + let (status, resp) = signed_admin_request( + &hot.url, + Method::PUT, + "/rustfs/admin/v3/tier", + Some(&body), + &hot.access_key, + &hot.secret_key, + ) + .await?; + if !status.is_success() { + return Err(format!("AddTier(RustFS) failed: status={status}, body={resp}").into()); + } + Ok(()) +} + +/// A current-version `Transition Days=0` rule scoped to the object's prefix. +fn transition_rule() -> Result> { + Ok(LifecycleRule::builder() + .id("ilm7-transition") + .filter(LifecycleRuleFilter::builder().prefix("tier/").build()) + .transitions( + Transition::builder() + .days(0) + .storage_class(TransitionStorageClass::from(TIER_NAME)) + .build(), + ) + .status(ExpirationStatus::Enabled) + .build()?) +} + +/// Upload `data` as a two-part multipart object with a content-type and one +/// user-metadata entry. +async fn put_multipart_object(client: &Client, bucket: &str, key: &str, data: &[u8]) -> TestResult { + let create = client + .create_multipart_upload() + .bucket(bucket) + .key(key) + .content_type(CONTENT_TYPE) + .metadata(USER_META_KEY, USER_META_VAL) + .send() + .await?; + let upload_id = create + .upload_id() + .ok_or("CreateMultipartUpload returned no upload id")? + .to_string(); + + let mut completed = Vec::new(); + for (idx, chunk) in [&data[..PART0_SIZE], &data[PART0_SIZE..]].into_iter().enumerate() { + let part_number = (idx + 1) as i32; + let uploaded = client + .upload_part() + .bucket(bucket) + .key(key) + .upload_id(&upload_id) + .part_number(part_number) + .body(ByteStream::from(chunk.to_vec())) + .send() + .await?; + completed.push( + CompletedPart::builder() + .part_number(part_number) + .e_tag(uploaded.e_tag().unwrap_or_default()) + .build(), + ); + } + + client + .complete_multipart_upload() + .bucket(bucket) + .key(key) + .upload_id(&upload_id) + .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed)).build()) + .send() + .await?; + Ok(()) +} + +/// Number of objects currently stored in the cold-tier bucket. +async fn cold_tier_object_count(cold_client: &Client) -> Result> { + let resp = cold_client.list_objects_v2().bucket(TIER_BUCKET).send().await?; + Ok(resp.contents().len()) +} + +/// Poll `HEAD` until the object's storage class is the tier name (transition +/// complete), or fail after `deadline`. +async fn wait_for_transition(client: &Client, bucket: &str, key: &str, deadline: StdDuration) -> TestResult { + let start = Instant::now(); + loop { + let head = client.head_object().bucket(bucket).key(key).send().await?; + if head.storage_class().map(|sc| sc.as_str()) == Some(TIER_NAME) { + return Ok(()); + } + if start.elapsed() >= deadline { + return Err(format!( + "object {bucket}/{key} was not transitioned to {TIER_NAME} within {}s (storage_class={:?})", + deadline.as_secs(), + head.storage_class() + ) + .into()); + } + tokio::time::sleep(StdDuration::from_millis(500)).await; + } +} + +/// Poll until the cold-tier bucket is empty (remote free-version cleanup done), +/// or fail after `deadline`. +async fn wait_for_cold_tier_empty(cold_client: &Client, deadline: StdDuration) -> TestResult { + let start = Instant::now(); + loop { + let count = cold_tier_object_count(cold_client).await?; + if count == 0 { + return Ok(()); + } + if start.elapsed() >= deadline { + return Err(format!( + "cold-tier bucket still holds {count} object(s) {}s after DeleteObject; \ + free-version remote cleanup did not converge", + deadline.as_secs() + ) + .into()); + } + tokio::time::sleep(StdDuration::from_millis(500)).await; + } +} + +/// GET `bytes=start-end` (inclusive) and assert it equals `data[start..=end]`. +async fn assert_range(client: &Client, start: usize, end: usize, data: &[u8]) -> TestResult { + let range = format!("bytes={start}-{end}"); + let resp = client + .get_object() + .bucket(SOURCE_BUCKET) + .key(OBJECT_KEY) + .range(&range) + .send() + .await?; + let got = resp.body.collect().await?.into_bytes(); + let expected = &data[start..=end]; + assert_eq!(got.len(), expected.len(), "range {range}: length mismatch"); + assert_eq!(got.as_ref(), expected, "range {range}: bytes mismatch reading from the tier"); + Ok(()) +} + +/// Full ilm-7 hermetic transition main path across two embedded RustFS servers. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_hermetic_transition_main_path() -> TestResult { + // Cold-tier server (independent credentials). It is a passive tier target, + // so it needs no lifecycle/scanner configuration. + let mut cold = RustFSTestEnvironment::new().await?; + cold.access_key = "coldtieradmin".to_string(); + cold.secret_key = "coldtiersecret".to_string(); + cold.start_rustfs_server_without_cleanup(vec![]).await?; + let cold_client = cold.create_s3_client(); + cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; + + // Hot/source server. A 1s scanner cycle is a backstop; transition is + // primarily driven immediately by the multipart completion path. + let mut hot = RustFSTestEnvironment::new().await?; + hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_CYCLE", "1")]) + .await?; + let hot_client = hot.create_s3_client(); + + // Wire the RustFS remote tier (real connectivity probe, no force). + add_rustfs_tier(&hot, &cold).await?; + + // Source bucket + Days=0 transition rule installed BEFORE the write so the + // completion path enqueues the transition immediately. + hot_client.create_bucket().bucket(SOURCE_BUCKET).send().await?; + let lifecycle = BucketLifecycleConfiguration::builder().rules(transition_rule()?).build()?; + hot_client + .put_bucket_lifecycle_configuration() + .bucket(SOURCE_BUCKET) + .lifecycle_configuration(lifecycle) + .send() + .await?; + + let data = payload(); + put_multipart_object(&hot_client, SOURCE_BUCKET, OBJECT_KEY, &data).await?; + + // 1) Transition completes: HEAD reports the tier name and no restore state. + wait_for_transition(&hot_client, SOURCE_BUCKET, OBJECT_KEY, StdDuration::from_secs(90)).await?; + let head = hot_client.head_object().bucket(SOURCE_BUCKET).key(OBJECT_KEY).send().await?; + assert_eq!( + head.storage_class().map(|sc| sc.as_str()), + Some(TIER_NAME), + "transitioned object must report the tier name as its storage class" + ); + assert!( + head.restore().is_none(), + "a freshly transitioned object must not advertise x-amz-restore, got {:?}", + head.restore() + ); + + // 2) The remote object now lives in the cold-tier bucket. + assert!( + cold_tier_object_count(&cold_client).await? >= 1, + "cold-tier bucket must hold the transitioned object" + ); + + // 3) GET streams identical bytes back through the warm backend; content-type + // and user metadata survive the transition round trip (rustfs#2246). + let get = hot_client.get_object().bucket(SOURCE_BUCKET).key(OBJECT_KEY).send().await?; + assert_eq!(get.content_type(), Some(CONTENT_TYPE), "content-type must survive transition"); + assert_eq!( + get.metadata().and_then(|m| m.get(USER_META_KEY)).map(String::as_str), + Some(USER_META_VAL), + "user metadata must survive transition" + ); + let body = get.body.collect().await?.into_bytes(); + assert_eq!(body.len(), data.len(), "full transitioned GET length mismatch"); + assert_eq!(body.as_ref(), data.as_slice(), "full transitioned GET must be byte-identical"); + + // 4) Range GET within a single part and across the part boundary + // (backlog#807): both must read the correct bytes from the tier. + assert_range(&hot_client, 1000, 1099, &data).await?; + assert_range(&hot_client, PART0_SIZE - 5, PART0_SIZE + 4, &data).await?; + + // 5) DeleteObject drives free-version cleanup. The local object is gone + // immediately; the remote copy is removed asynchronously. + hot_client + .delete_object() + .bucket(SOURCE_BUCKET) + .key(OBJECT_KEY) + .send() + .await?; + + let get_after = hot_client.get_object().bucket(SOURCE_BUCKET).key(OBJECT_KEY).send().await; + let err = get_after.expect_err("hot object must be gone immediately after delete"); + assert_eq!( + err.as_service_error().and_then(|e| e.code()), + Some("NoSuchKey"), + "hot GET after delete must be NoSuchKey, got {err:?}" + ); + + wait_for_cold_tier_empty(&cold_client, StdDuration::from_secs(90)).await?; + + Ok(()) +}