mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 08:18:18 +00:00
test(ilm): add hermetic RustFS-to-RustFS transition main-path e2e (#4818)
This commit is contained in:
Zhengchao An
@@ -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
|
||||
|
||||
@@ -22,3 +22,4 @@ mod lifecycle;
|
||||
mod lock;
|
||||
mod node_interact_test;
|
||||
mod sql;
|
||||
mod tiering;
|
||||
|
||||
@@ -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:<port>`.
|
||||
//!
|
||||
//! 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 == <tier name>` 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<dyn std::error::Error + Send + Sync>>;
|
||||
|
||||
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<u8> {
|
||||
(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<dyn std::error::Error + Send + Sync>> {
|
||||
let url = format!("{base_url}{path}");
|
||||
let uri = url.parse::<http::Uri>()?;
|
||||
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<LifecycleRule, Box<dyn std::error::Error + Send + Sync>> {
|
||||
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<usize, Box<dyn std::error::Error + Send + Sync>> {
|
||||
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(())
|
||||
}
|
||||
Reference in New Issue
Block a user