diff --git a/crates/e2e_test/src/reliant/tiering.rs b/crates/e2e_test/src/reliant/tiering.rs index a8106d231..97931389c 100644 --- a/crates/e2e_test/src/reliant/tiering.rs +++ b/crates/e2e_test/src/reliant/tiering.rs @@ -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>; @@ -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> { + transition_rule_for("ilm7-transition", "tier/", 0) +} + +fn transition_rule_for(id: &str, prefix: &str, days: i32) -> Result> { 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 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, + status_endpoint: Option, + report: ManualTransitionRunReport, +} + +#[derive(Debug, Deserialize)] +struct ManualTransitionRunReport { + bucket: String, + prefix: String, + tier: Option, + 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> { + 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> { 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(()) +}