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

test(ilm): add manual transition e2e coverage

Browse Source 
Co-Authored-By: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-07-24 14:09:59 +08:00
parent aae38da307
commit 87af7c0c3d
1 changed files with 270 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/e2e_test/src/reliant/tiering.rs
+270 -5
View File
@@ -48,15 +48,17 @@ 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,
BucketLifecycleConfiguration, BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, ExpirationStatus,
LifecycleRule, LifecycleRuleFilter, NoncurrentVersionTransition, Transition, TransitionStorageClass, VersioningConfiguration,
};
use http::Method;
use http::header::HOST;
use rustfs_signer::constants::UNSIGNED_PAYLOAD;
use rustfs_signer::sign_v4;
use s3s::Body;
use serde::Deserialize;
use std::time::{Duration as StdDuration, Instant};
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
@@ -64,10 +66,18 @@ const TIER_NAME: &str = "COLDTIER";
const TIER_BUCKET: &str = "ilm7-cold-tier";
const TIER_PREFIX: &str = "tiered";
const SOURCE_BUCKET: &str = "ilm7-hot";
const MANUAL_DUE_BUCKET: &str = "ilm7-manual-due";
const MANUAL_DRY_RUN_BUCKET: &str = "ilm7-manual-dry-run";
const MANUAL_NOT_DUE_BUCKET: &str = "ilm7-manual-not-due";
const OBJECT_KEY: &str = "tier/鲁A12345/report.bin";
const MANUAL_DUE_KEY: &str = "manual-due/report.bin";
const MANUAL_DRY_RUN_KEY: &str = "manual-dry-run/report.bin";
const MANUAL_NOT_DUE_KEY: &str = "manual-not-due/report.bin";
const CONTENT_TYPE: &str = "application/x-ilm7";
const USER_META_KEY: &str = "ilm7-origin";
const USER_META_VAL: &str = "hermetic-transition";
const HDR_SOURCE_REPLICATION_REQUEST: &str = "x-rustfs-source-replication-request";
const HDR_SOURCE_MTIME: &str = "x-rustfs-source-mtime";
/// 5 MiB — the S3 minimum size for a non-final multipart part; the object's only
/// internal part boundary sits at this offset.
@@ -158,12 +168,16 @@ async fn add_rustfs_tier(hot: &RustFSTestEnvironment, cold: &RustFSTestEnvironme
/// 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>> {
transition_rule_for("ilm7-transition", "tier/", 0)
}
fn transition_rule_for(id: &str, prefix: &str, days: i32) -> Result<LifecycleRule, Box<dyn std::error::Error + Send + Sync>> {
Ok(LifecycleRule::builder()
.id("ilm7-transition")
.filter(LifecycleRuleFilter::builder().prefix("tier/").build())
.id(id)
.filter(LifecycleRuleFilter::builder().prefix(prefix).build())
.transitions(
Transition::builder()
.days(0)
.days(days)
.storage_class(TransitionStorageClass::from(TIER_NAME))
.build(),
)
@@ -171,6 +185,55 @@ fn transition_rule() -> Result<LifecycleRule, Box<dyn std::error::Error + Send +
.build()?)
}
async fn put_lifecycle_transition_rule(client: &Client, bucket: &str, id: &str, prefix: &str, days: i32) -> TestResult {
let lifecycle = BucketLifecycleConfiguration::builder()
.rules(transition_rule_for(id, prefix, days)?)
.build()?;
client
.put_bucket_lifecycle_configuration()
.bucket(bucket)
.lifecycle_configuration(lifecycle)
.send()
.await?;
Ok(())
}
async fn put_lifecycle_noncurrent_transition_rule(client: &Client, bucket: &str, id: &str, prefix: &str) -> TestResult {
let rule = LifecycleRule::builder()
.id(id)
.filter(LifecycleRuleFilter::builder().prefix(prefix).build())
.noncurrent_version_transitions(
NoncurrentVersionTransition::builder()
.noncurrent_days(0)
.storage_class(TransitionStorageClass::from(TIER_NAME))
.build(),
)
.status(ExpirationStatus::Enabled)
.build()?;
let lifecycle = BucketLifecycleConfiguration::builder().rules(rule).build()?;
client
.put_bucket_lifecycle_configuration()
.bucket(bucket)
.lifecycle_configuration(lifecycle)
.send()
.await?;
Ok(())
}
async fn enable_bucket_versioning(client: &Client, bucket: &str) -> TestResult {
client
.put_bucket_versioning()
.bucket(bucket)
.versioning_configuration(
VersioningConfiguration::builder()
.status(BucketVersioningStatus::Enabled)
.build(),
)
.send()
.await?;
Ok(())
}
/// 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 {
@@ -218,6 +281,90 @@ async fn put_multipart_object(client: &Client, bucket: &str, key: &str, data: &[
Ok(())
}
async fn put_single_part_object(client: &Client, bucket: &str, key: &str, body: &'static [u8]) -> TestResult {
client
.put_object()
.bucket(bucket)
.key(key)
.body(ByteStream::from_static(body))
.send()
.await?;
Ok(())
}
async fn put_backdated_single_part_object(
client: &Client,
bucket: &str,
key: &str,
body: &'static [u8],
mtime: OffsetDateTime,
) -> TestResult {
let mtime_rfc3339 = mtime.format(&Rfc3339)?;
client
.put_object()
.bucket(bucket)
.key(key)
.body(ByteStream::from_static(body))
.customize()
.mutate_request(move |req| {
req.headers_mut().insert(HDR_SOURCE_REPLICATION_REQUEST, "true");
req.headers_mut().insert(HDR_SOURCE_MTIME, mtime_rfc3339.clone());
})
.send()
.await?;
Ok(())
}
#[derive(Debug, Deserialize)]
struct ManualTransitionRunResponse {
state: String,
mode: String,
job_id: Option<String>,
status_endpoint: Option<String>,
report: ManualTransitionRunReport,
}
#[derive(Debug, Deserialize)]
struct ManualTransitionRunReport {
bucket: String,
prefix: String,
tier: Option<String>,
dry_run: bool,
lifecycle_config_found: bool,
scanned: u64,
eligible: u64,
enqueued: u64,
dry_run_eligible: u64,
skipped_not_transition: u64,
skipped_tier: u64,
skipped_delete_marker: u64,
skipped_directory: u64,
skipped_replication: u64,
skipped_already_in_flight: u64,
skipped_queue_full: u64,
skipped_queue_closed: u64,
skipped_queue_timeout: u64,
truncated_by_limit: bool,
}
async fn manual_transition_run(
hot: &RustFSTestEnvironment,
bucket: &str,
prefix: &str,
dry_run: bool,
) -> Result<ManualTransitionRunResponse, Box<dyn std::error::Error + Send + Sync>> {
let bucket = urlencoding::encode(bucket);
let prefix = urlencoding::encode(prefix);
let tier = urlencoding::encode(TIER_NAME);
let path =
format!("/rustfs/admin/v3/ilm/transition/run?bucket={bucket}&prefix={prefix}&tier={tier}&dryRun={dry_run}&maxObjects=10");
let (status, body) = signed_admin_request(&hot.url, Method::POST, &path, None, &hot.access_key, &hot.secret_key).await?;
if !status.is_success() {
return Err(format!("manual transition run failed: status={status}, body={body}").into());
}
Ok(serde_json::from_str(&body)?)
}
/// 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?;
@@ -283,6 +430,27 @@ async fn assert_range(client: &Client, start: usize, end: usize, data: &[u8]) ->
Ok(())
}
async fn assert_not_transitioned(client: &Client, bucket: &str, key: &str) -> TestResult {
let head = client.head_object().bucket(bucket).key(key).send().await?;
assert!(
head.storage_class().is_none(),
"{bucket}/{key} must remain in the hot tier, got storage_class={:?}",
head.storage_class()
);
Ok(())
}
async fn assert_remains_not_transitioned(client: &Client, bucket: &str, key: &str, duration: StdDuration) -> TestResult {
let deadline = Instant::now() + duration;
loop {
assert_not_transitioned(client, bucket, key).await?;
if Instant::now() >= deadline {
return Ok(());
}
tokio::time::sleep(StdDuration::from_millis(250)).await;
}
}
/// 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 {
@@ -378,3 +546,100 @@ async fn test_hermetic_transition_main_path() -> TestResult {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_manual_transition_run_black_box_semantics() -> TestResult {
let mut cold = RustFSTestEnvironment::new().await?;
cold.access_key = "manualcoldtieradmin".to_string();
cold.secret_key = "manualcoldtiersecret".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?;
let mut hot = RustFSTestEnvironment::new().await?;
hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")])
.await?;
let hot_client = hot.create_s3_client();
add_rustfs_tier(&hot, &cold).await?;
let due_mtime = OffsetDateTime::now_utc() - time::Duration::hours(25);
hot_client.create_bucket().bucket(MANUAL_DUE_BUCKET).send().await?;
put_lifecycle_transition_rule(&hot_client, MANUAL_DUE_BUCKET, "manual-due", "manual-due/", 0).await?;
put_backdated_single_part_object(&hot_client, MANUAL_DUE_BUCKET, MANUAL_DUE_KEY, b"manual due object", due_mtime).await?;
let due = manual_transition_run(&hot, MANUAL_DUE_BUCKET, "manual-due/", false).await?;
assert_eq!(due.mode, "enqueue_only");
assert!(due.job_id.is_none());
assert!(due.status_endpoint.is_none());
assert_eq!(due.state, "completed");
assert_eq!(due.report.bucket, MANUAL_DUE_BUCKET);
assert_eq!(due.report.prefix, "manual-due/");
assert_eq!(due.report.tier.as_deref(), Some(TIER_NAME));
assert!(!due.report.dry_run);
assert!(due.report.lifecycle_config_found);
assert_eq!(due.report.scanned, 1, "due report: {:#?}", due.report);
assert_eq!(due.report.eligible, 1, "due report: {:#?}", due.report);
assert_eq!(
due.report.enqueued + due.report.skipped_already_in_flight,
1,
"due report: {:#?}",
due.report
);
assert_eq!(due.report.skipped_tier, 0);
assert_eq!(due.report.skipped_delete_marker, 0);
assert_eq!(due.report.skipped_directory, 0);
assert_eq!(due.report.skipped_replication, 0);
assert!(!due.report.truncated_by_limit);
wait_for_transition(&hot_client, MANUAL_DUE_BUCKET, MANUAL_DUE_KEY, StdDuration::from_secs(90)).await?;
let remote_count_after_due = cold_tier_object_count(&cold_client).await?;
hot_client.create_bucket().bucket(MANUAL_DRY_RUN_BUCKET).send().await?;
enable_bucket_versioning(&hot_client, MANUAL_DRY_RUN_BUCKET).await?;
put_lifecycle_noncurrent_transition_rule(&hot_client, MANUAL_DRY_RUN_BUCKET, "manual-dry-run", "manual-dry-run/").await?;
put_single_part_object(&hot_client, MANUAL_DRY_RUN_BUCKET, MANUAL_DRY_RUN_KEY, b"manual dry-run object v1").await?;
put_single_part_object(&hot_client, MANUAL_DRY_RUN_BUCKET, MANUAL_DRY_RUN_KEY, b"manual dry-run object v2").await?;
let before_dry_run_remote_count = cold_tier_object_count(&cold_client).await?;
assert_eq!(
before_dry_run_remote_count, remote_count_after_due,
"dry-run setup must not enqueue transition work before the manual dry-run"
);
let dry = manual_transition_run(&hot, MANUAL_DRY_RUN_BUCKET, "manual-dry-run/", true).await?;
assert_eq!(dry.state, "completed");
assert_eq!(dry.report.bucket, MANUAL_DRY_RUN_BUCKET);
assert_eq!(dry.report.prefix, "manual-dry-run/");
assert_eq!(dry.report.tier.as_deref(), Some(TIER_NAME));
assert!(dry.report.dry_run);
assert_eq!(dry.report.scanned, 2, "dry-run report: {:#?}", dry.report);
assert_eq!(dry.report.eligible, 1, "dry-run report: {:#?}", dry.report);
assert_eq!(dry.report.dry_run_eligible, 1, "dry-run report: {:#?}", dry.report);
assert_eq!(dry.report.enqueued, 0, "dry-run report: {:#?}", dry.report);
assert_eq!(dry.report.skipped_not_transition, 1, "dry-run report: {:#?}", dry.report);
assert_eq!(
cold_tier_object_count(&cold_client).await?,
before_dry_run_remote_count,
"dry-run must not create a remote tier object"
);
assert_not_transitioned(&hot_client, MANUAL_DRY_RUN_BUCKET, MANUAL_DRY_RUN_KEY).await?;
hot_client.create_bucket().bucket(MANUAL_NOT_DUE_BUCKET).send().await?;
put_lifecycle_transition_rule(&hot_client, MANUAL_NOT_DUE_BUCKET, "manual-not-due", "manual-not-due/", 1).await?;
put_single_part_object(&hot_client, MANUAL_NOT_DUE_BUCKET, MANUAL_NOT_DUE_KEY, b"manual not-yet-due object").await?;
let not_due = manual_transition_run(&hot, MANUAL_NOT_DUE_BUCKET, "manual-not-due/", false).await?;
assert_eq!(not_due.state, "completed");
assert_eq!(not_due.report.bucket, MANUAL_NOT_DUE_BUCKET);
assert_eq!(not_due.report.prefix, "manual-not-due/");
assert_eq!(not_due.report.tier.as_deref(), Some(TIER_NAME));
assert!(!not_due.report.dry_run);
assert_eq!(not_due.report.scanned, 1, "not-due report: {:#?}", not_due.report);
assert_eq!(not_due.report.eligible, 0, "not-due report: {:#?}", not_due.report);
assert_eq!(not_due.report.enqueued, 0, "not-due report: {:#?}", not_due.report);
assert_eq!(not_due.report.skipped_not_transition, 1, "not-due report: {:#?}", not_due.report);
assert_eq!(not_due.report.skipped_queue_full, 0);
assert_eq!(not_due.report.skipped_queue_closed, 0);
assert_eq!(not_due.report.skipped_queue_timeout, 0);
assert_remains_not_transitioned(&hot_client, MANUAL_NOT_DUE_BUCKET, MANUAL_NOT_DUE_KEY, StdDuration::from_secs(2)).await?;
Ok(())
}