From 0d93fdb7c2117aa78fbc2bd01c2efb02ec6e59e5 Mon Sep 17 00:00:00 2001 From: Marian Date: Tue, 14 Jul 2026 18:14:47 +0000 Subject: [PATCH] feat(bucket): manage S3 lifecycle rules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an optional `lifecycle` list to BucketSpec, so a bucket's expiration rules become part of the same declarative object as its quota and versioning. Motivation: a pull-through cache bucket that no client ever prunes. The application writing it could not GC its own cache, and every external cleaner we tried needed a LIST — which is exactly what fails once the volume is full, so the cleanup broke precisely when it was needed. A server-side lifecycle rule is the only thing that still works under pressure, and it was the one bucket property the operator could not express, forcing an imperative PUT ?lifecycle job alongside otherwise declarative CRs. rc-core already exposes get/set/delete_bucket_lifecycle on ObjectStore; this only surfaces them. Semantics follow the existing quota/versioning fields: - unset -> unmanaged: the operator never even READS the bucket's lifecycle, so adopting a bucket that already has rules cannot wipe them - empty list -> "no rules": DELETEs the configuration (S3 rejects a PUT of an empty one) - non-empty -> replaces the configuration when it differs The CRD models a deliberate subset (expiration by age, abort incomplete multipart uploads). Transitions and non-current-version rules are omitted: they need storage tiers / versioning that RustFS does not offer, and a CRD field that silently does nothing is worse than no field. Drift is detected on the serialized form because rc-core's LifecycleRule has no PartialEq — comparing only the modelled fields would call a bucket in sync while an unmodelled rule still differed. Tests: 4 unit tests (apply / no-op when matching / empty list deletes / unset is untouched) and the integration test now round-trips a rule through a real RustFS container, asserting the rule comes back with its prefix, expiration and abort-multipart days, and is gone after delete. --- .../templates/buckets.yaml | 48 ++++++ charts/rustfs-operator/crds/crds.yaml | 48 ++++++ deploy/crds.yaml | 48 ++++++ src/crd.rs | 44 ++++++ src/provider.rs | 23 +++ src/reconcile/bucket.rs | 141 +++++++++++++++++- tests/e2e_k3s.rs | 3 + tests/integration_rustfs.rs | 45 ++++++ 8 files changed, 399 insertions(+), 1 deletion(-) diff --git a/charts/rustfs-operator-crds/templates/buckets.yaml b/charts/rustfs-operator-crds/templates/buckets.yaml index 27f6611..dd22636 100644 --- a/charts/rustfs-operator-crds/templates/buckets.yaml +++ b/charts/rustfs-operator-crds/templates/buckets.yaml @@ -72,6 +72,54 @@ spec: - Delete - Retain type: string + lifecycle: + description: |- + Lifecycle rules; unset means unmanaged (the operator never reads or writes + the bucket's lifecycle configuration). An explicitly EMPTY list is not the + same thing: it means "no rules", and removes any configuration present on + the bucket. + items: + description: |- + A single S3 lifecycle rule. + + Deliberately a *subset* of the S3 lifecycle spec: expiration by age and + aborting stale multipart uploads. Transitions and non-current-version rules + are omitted — they only make sense with storage tiers / versioning, which + RustFS does not offer, and a CRD field that silently does nothing is worse + than no field. + properties: + abortIncompleteMultipartUploadDays: + description: |- + Abort multipart uploads left incomplete for this many days. Worth setting + on any bucket that takes large writes: aborted parts are invisible to a + normal LIST and still consume the disk. + format: int32 + nullable: true + type: integer + expirationDays: + description: Expire objects this many days after creation. + format: int32 + nullable: true + type: integer + id: + description: Rule ID, unique within the bucket. + type: string + prefix: + description: Object-key prefix the rule applies to. Unset = the whole bucket. + nullable: true + type: string + status: + default: Enabled + description: Enabled (default) or Disabled. + enum: + - Enabled + - Disabled + type: string + required: + - id + type: object + nullable: true + type: array quotaBytes: description: Hard quota in bytes; unset means unmanaged. format: uint64 diff --git a/charts/rustfs-operator/crds/crds.yaml b/charts/rustfs-operator/crds/crds.yaml index e3b98f3..526a95e 100644 --- a/charts/rustfs-operator/crds/crds.yaml +++ b/charts/rustfs-operator/crds/crds.yaml @@ -55,6 +55,54 @@ spec: - Delete - Retain type: string + lifecycle: + description: |- + Lifecycle rules; unset means unmanaged (the operator never reads or writes + the bucket's lifecycle configuration). An explicitly EMPTY list is not the + same thing: it means "no rules", and removes any configuration present on + the bucket. + items: + description: |- + A single S3 lifecycle rule. + + Deliberately a *subset* of the S3 lifecycle spec: expiration by age and + aborting stale multipart uploads. Transitions and non-current-version rules + are omitted — they only make sense with storage tiers / versioning, which + RustFS does not offer, and a CRD field that silently does nothing is worse + than no field. + properties: + abortIncompleteMultipartUploadDays: + description: |- + Abort multipart uploads left incomplete for this many days. Worth setting + on any bucket that takes large writes: aborted parts are invisible to a + normal LIST and still consume the disk. + format: int32 + nullable: true + type: integer + expirationDays: + description: Expire objects this many days after creation. + format: int32 + nullable: true + type: integer + id: + description: Rule ID, unique within the bucket. + type: string + prefix: + description: Object-key prefix the rule applies to. Unset = the whole bucket. + nullable: true + type: string + status: + default: Enabled + description: Enabled (default) or Disabled. + enum: + - Enabled + - Disabled + type: string + required: + - id + type: object + nullable: true + type: array quotaBytes: description: Hard quota in bytes; unset means unmanaged. format: uint64 diff --git a/deploy/crds.yaml b/deploy/crds.yaml index e3b98f3..526a95e 100644 --- a/deploy/crds.yaml +++ b/deploy/crds.yaml @@ -55,6 +55,54 @@ spec: - Delete - Retain type: string + lifecycle: + description: |- + Lifecycle rules; unset means unmanaged (the operator never reads or writes + the bucket's lifecycle configuration). An explicitly EMPTY list is not the + same thing: it means "no rules", and removes any configuration present on + the bucket. + items: + description: |- + A single S3 lifecycle rule. + + Deliberately a *subset* of the S3 lifecycle spec: expiration by age and + aborting stale multipart uploads. Transitions and non-current-version rules + are omitted — they only make sense with storage tiers / versioning, which + RustFS does not offer, and a CRD field that silently does nothing is worse + than no field. + properties: + abortIncompleteMultipartUploadDays: + description: |- + Abort multipart uploads left incomplete for this many days. Worth setting + on any bucket that takes large writes: aborted parts are invisible to a + normal LIST and still consume the disk. + format: int32 + nullable: true + type: integer + expirationDays: + description: Expire objects this many days after creation. + format: int32 + nullable: true + type: integer + id: + description: Rule ID, unique within the bucket. + type: string + prefix: + description: Object-key prefix the rule applies to. Unset = the whole bucket. + nullable: true + type: string + status: + default: Enabled + description: Enabled (default) or Disabled. + enum: + - Enabled + - Disabled + type: string + required: + - id + type: object + nullable: true + type: array quotaBytes: description: Hard quota in bytes; unset means unmanaged. format: uint64 diff --git a/src/crd.rs b/src/crd.rs index cb90d48..3f03f9c 100644 --- a/src/crd.rs +++ b/src/crd.rs @@ -96,6 +96,44 @@ pub enum DeletionPolicy { Retain, } +/// Whether a lifecycle rule is active. +#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize, JsonSchema, PartialEq)] +pub enum LifecycleStatus { + /// The rule is evaluated by the scanner. + #[default] + Enabled, + /// The rule is kept but not evaluated. + Disabled, +} + +/// A single S3 lifecycle rule. +/// +/// Deliberately a *subset* of the S3 lifecycle spec: expiration by age and +/// aborting stale multipart uploads. Transitions and non-current-version rules +/// are omitted — they only make sense with storage tiers / versioning, which +/// RustFS does not offer, and a CRD field that silently does nothing is worse +/// than no field. +#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct LifecycleRuleSpec { + /// Rule ID, unique within the bucket. + pub id: String, + /// Enabled (default) or Disabled. + #[serde(default)] + pub status: LifecycleStatus, + /// Object-key prefix the rule applies to. Unset = the whole bucket. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub prefix: Option, + /// Expire objects this many days after creation. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub expiration_days: Option, + /// Abort multipart uploads left incomplete for this many days. Worth setting + /// on any bucket that takes large writes: aborted parts are invisible to a + /// normal LIST and still consume the disk. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub abort_incomplete_multipart_upload_days: Option, +} + /// Shared status for all RustFS resources. #[derive(Clone, Debug, Default, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "camelCase")] @@ -153,6 +191,12 @@ pub struct BucketSpec { /// Hard quota in bytes; unset means unmanaged. #[serde(default, skip_serializing_if = "Option::is_none")] pub quota_bytes: Option, + /// Lifecycle rules; unset means unmanaged (the operator never reads or writes + /// the bucket's lifecycle configuration). An explicitly EMPTY list is not the + /// same thing: it means "no rules", and removes any configuration present on + /// the bucket. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub lifecycle: Option>, #[serde(default)] pub deletion_policy: DeletionPolicy, } diff --git a/src/provider.rs b/src/provider.rs index 2cfad2e..b94f4d4 100644 --- a/src/provider.rs +++ b/src/provider.rs @@ -7,6 +7,7 @@ use async_trait::async_trait; use rc_core::admin::{ AdminApi, CreateServiceAccountRequest, Policy, PolicyEntity, ServiceAccount, User, UserStatus, }; +use rc_core::lifecycle::LifecycleRule; use rc_core::traits::ObjectStore; use rc_core::{Alias, Error as RcError}; use rc_s3::{AdminClient, S3Client}; @@ -56,6 +57,13 @@ pub trait RustFs: Send + Sync { async fn set_versioning(&self, bucket: &str, enabled: bool) -> Result<()>; async fn get_bucket_quota(&self, bucket: &str) -> Result>; async fn set_bucket_quota(&self, bucket: &str, quota: u64) -> Result<()>; + /// Current lifecycle rules. Empty when the bucket has no configuration. + async fn get_bucket_lifecycle(&self, bucket: &str) -> Result>; + /// Replace the bucket's lifecycle configuration. RustFS has no per-rule API — + /// the whole configuration is written at once. + async fn set_bucket_lifecycle(&self, bucket: &str, rules: Vec) -> Result<()>; + /// Remove the lifecycle configuration entirely. + async fn delete_bucket_lifecycle(&self, bucket: &str) -> Result<()>; // Users async fn get_user(&self, access_key: &str) -> Result>; @@ -204,6 +212,21 @@ impl RustFs for RustFsProvider { Ok(()) } + async fn get_bucket_lifecycle(&self, bucket: &str) -> Result> { + Self::cli(&format!("ilm rule list $ALIAS/{bucket}")); + Ok(self.s3.get_bucket_lifecycle(bucket).await?) + } + + async fn set_bucket_lifecycle(&self, bucket: &str, rules: Vec) -> Result<()> { + Self::cli(&format!("ilm rule add $ALIAS/{bucket} (x{})", rules.len())); + Ok(self.s3.set_bucket_lifecycle(bucket, rules).await?) + } + + async fn delete_bucket_lifecycle(&self, bucket: &str) -> Result<()> { + Self::cli(&format!("ilm rule rm --all --force $ALIAS/{bucket}")); + Ok(self.s3.delete_bucket_lifecycle(bucket).await?) + } + async fn get_user(&self, access_key: &str) -> Result> { Self::cli(&format!("admin user info $ALIAS {access_key}")); optional(self.admin.get_user(access_key).await) diff --git a/src/reconcile/bucket.rs b/src/reconcile/bucket.rs index 4b56af1..59b589b 100644 --- a/src/reconcile/bucket.rs +++ b/src/reconcile/bucket.rs @@ -6,9 +6,13 @@ use kube::runtime::controller::Action; use kube::runtime::finalizer::{Event, finalizer}; use kube::{Api, ResourceExt}; +use rc_core::lifecycle::{LifecycleExpiration, LifecycleRule, LifecycleRuleStatus}; + use super::{Context, FINALIZER, REQUEUE_OK, namespace_of, patch_status}; use crate::connection::provider_for; -use crate::crd::{Bucket, BucketSpec, DeletionPolicy, ResourceStatus}; +use crate::crd::{ + Bucket, BucketSpec, DeletionPolicy, LifecycleRuleSpec, LifecycleStatus, ResourceStatus, +}; use crate::error::{Error, Result}; use crate::provider::RustFs; @@ -28,9 +32,55 @@ pub async fn ensure_bucket(fs: &dyn RustFs, name: &str, spec: &BucketSpec) -> Re { fs.set_bucket_quota(name, quota).await?; } + if let Some(rules) = &spec.lifecycle { + let desired: Vec = rules.iter().map(to_rc_rule).collect(); + if !lifecycle_eq(&fs.get_bucket_lifecycle(name).await?, &desired) { + // An explicitly empty list means "no rules" — which is a DELETE, not a + // PUT of an empty configuration (S3 rejects the latter). + if desired.is_empty() { + fs.delete_bucket_lifecycle(name).await?; + } else { + fs.set_bucket_lifecycle(name, desired).await?; + } + } + } Ok(()) } +/// Map a CRD rule onto `rc-core`'s wire type. +fn to_rc_rule(spec: &LifecycleRuleSpec) -> LifecycleRule { + LifecycleRule { + id: spec.id.clone(), + status: match spec.status { + LifecycleStatus::Enabled => LifecycleRuleStatus::Enabled, + LifecycleStatus::Disabled => LifecycleRuleStatus::Disabled, + }, + prefix: spec.prefix.clone(), + tags: None, + expiration: spec.expiration_days.map(|days| LifecycleExpiration { + days: Some(days), + date: None, + }), + transition: None, + noncurrent_version_expiration: None, + noncurrent_version_transition: None, + expired_object_delete_marker: None, + abort_incomplete_multipart_upload_days: spec.abort_incomplete_multipart_upload_days, + } +} + +/// `rc-core`'s `LifecycleRule` has no `PartialEq`, and comparing only the fields the +/// CRD models would call a bucket "in sync" while a rule we do not model still differs. +/// Compare the serialized form instead — it covers every field on the wire. +fn lifecycle_eq(current: &[LifecycleRule], desired: &[LifecycleRule]) -> bool { + match (serde_json::to_value(current), serde_json::to_value(desired)) { + (Ok(a), Ok(b)) => a == b, + // Unserializable means we cannot prove they match — rewrite rather than + // silently leave a bucket unmanaged. + _ => false, + } +} + /// Remove the bucket if the deletion policy asks for it. pub async fn cleanup_bucket(fs: &dyn RustFs, name: &str, spec: &BucketSpec) -> Result<()> { match spec.deletion_policy { @@ -92,10 +142,28 @@ mod tests { bucket_name: None, versioning, quota_bytes, + lifecycle: None, deletion_policy: DeletionPolicy::default(), } } + fn spec_with_lifecycle(rules: Option>) -> BucketSpec { + BucketSpec { + lifecycle: rules, + ..spec(None, None) + } + } + + fn expire_after(id: &str, prefix: &str, days: i32) -> LifecycleRuleSpec { + LifecycleRuleSpec { + id: id.into(), + status: LifecycleStatus::Enabled, + prefix: Some(prefix.into()), + expiration_days: Some(days), + abort_incomplete_multipart_upload_days: None, + } + } + #[tokio::test] async fn creates_missing_bucket() { let mut fs = MockRustFs::new(); @@ -140,6 +208,77 @@ mod tests { .unwrap(); } + #[tokio::test] + async fn applies_lifecycle_when_bucket_has_none() { + let mut fs = MockRustFs::new(); + fs.expect_bucket_exists().returning(|_| Ok(true)); + fs.expect_get_bucket_lifecycle().returning(|_| Ok(vec![])); + fs.expect_set_bucket_lifecycle() + .withf(|b, rules| { + b == "cache" + && rules.len() == 1 + && rules[0].id == "expire-proxy-cache" + && rules[0].prefix.as_deref() == Some("proxy-cache/") + && rules[0].expiration.as_ref().and_then(|e| e.days) == Some(1) + }) + .times(1) + .returning(|_, _| Ok(())); + + let spec = spec_with_lifecycle(Some(vec![expire_after( + "expire-proxy-cache", + "proxy-cache/", + 1, + )])); + ensure_bucket(&fs, "cache", &spec).await.unwrap(); + } + + #[tokio::test] + async fn lifecycle_untouched_when_already_matching() { + let rule = expire_after("expire-proxy-cache", "proxy-cache/", 1); + let existing = to_rc_rule(&rule); + + let mut fs = MockRustFs::new(); + fs.expect_bucket_exists().returning(|_| Ok(true)); + fs.expect_get_bucket_lifecycle() + .returning(move |_| Ok(vec![existing.clone()])); + // No set/delete: a matching bucket must not be rewritten on every reconcile. + fs.expect_set_bucket_lifecycle().never(); + fs.expect_delete_bucket_lifecycle().never(); + + let spec = spec_with_lifecycle(Some(vec![rule])); + ensure_bucket(&fs, "cache", &spec).await.unwrap(); + } + + #[tokio::test] + async fn empty_lifecycle_list_deletes_the_configuration() { + let mut fs = MockRustFs::new(); + fs.expect_bucket_exists().returning(|_| Ok(true)); + fs.expect_get_bucket_lifecycle() + .returning(|_| Ok(vec![to_rc_rule(&expire_after("old", "tmp/", 7))])); + // An empty list is a DELETE, not a PUT of an empty configuration. + fs.expect_delete_bucket_lifecycle() + .times(1) + .returning(|_| Ok(())); + fs.expect_set_bucket_lifecycle().never(); + + let spec = spec_with_lifecycle(Some(vec![])); + ensure_bucket(&fs, "cache", &spec).await.unwrap(); + } + + #[tokio::test] + async fn unset_lifecycle_is_unmanaged() { + let mut fs = MockRustFs::new(); + fs.expect_bucket_exists().returning(|_| Ok(true)); + // Unset means the operator must not even LOOK at the bucket's lifecycle — + // otherwise adopting a bucket that already has rules would wipe them. + fs.expect_get_bucket_lifecycle().never(); + fs.expect_set_bucket_lifecycle().never(); + fs.expect_delete_bucket_lifecycle().never(); + + let spec = spec_with_lifecycle(None); + ensure_bucket(&fs, "cache", &spec).await.unwrap(); + } + #[tokio::test] async fn retain_policy_skips_remote_delete() { let fs = MockRustFs::new(); // any call panics diff --git a/tests/e2e_k3s.rs b/tests/e2e_k3s.rs index 24d0759..bfa3eb8 100644 --- a/tests/e2e_k3s.rs +++ b/tests/e2e_k3s.rs @@ -215,6 +215,7 @@ async fn operator_reconciles_crs_against_rustfs() { bucket_name: None, versioning: Some(true), quota_bytes: Some(10 * 1024 * 1024), + lifecycle: None, deletion_policy: DeletionPolicy::Delete, }, ), @@ -425,6 +426,7 @@ async fn operator_reconciles_crs_against_rustfs() { bucket_name: None, versioning: None, quota_bytes: None, + lifecycle: None, deletion_policy: DeletionPolicy::Delete, }, ), @@ -464,6 +466,7 @@ async fn operator_reconciles_crs_against_rustfs() { versioning: None, quota_bytes: None, // Retain: cleanup must not need the (denied) connection + lifecycle: None, deletion_policy: DeletionPolicy::Retain, }, ), diff --git a/tests/integration_rustfs.rs b/tests/integration_rustfs.rs index 3b91fd9..bad9b2d 100644 --- a/tests/integration_rustfs.rs +++ b/tests/integration_rustfs.rs @@ -6,6 +6,7 @@ mod common; +use rc_core::lifecycle::{LifecycleExpiration, LifecycleRule, LifecycleRuleStatus}; use rustfs_operator::provider::RustFs; use serde_json::json; @@ -31,6 +32,50 @@ async fn provider_manages_buckets_policies_and_users() { Some(10 * 1024 * 1024) ); + // --- bucket lifecycle --- + assert!( + fs.get_bucket_lifecycle("it-bucket") + .await + .unwrap() + .is_empty(), + "a fresh bucket must report no lifecycle rules, not an error" + ); + + let rule = LifecycleRule { + id: "expire-cache".into(), + status: LifecycleRuleStatus::Enabled, + prefix: Some("cache/".into()), + tags: None, + expiration: Some(LifecycleExpiration { + days: Some(1), + date: None, + }), + transition: None, + noncurrent_version_expiration: None, + noncurrent_version_transition: None, + expired_object_delete_marker: None, + abort_incomplete_multipart_upload_days: Some(1), + }; + fs.set_bucket_lifecycle("it-bucket", vec![rule]) + .await + .unwrap(); + + let got = fs.get_bucket_lifecycle("it-bucket").await.unwrap(); + assert_eq!(got.len(), 1); + assert_eq!(got[0].id, "expire-cache"); + assert_eq!(got[0].prefix.as_deref(), Some("cache/")); + assert_eq!(got[0].expiration.as_ref().and_then(|e| e.days), Some(1)); + assert_eq!(got[0].abort_incomplete_multipart_upload_days, Some(1)); + + fs.delete_bucket_lifecycle("it-bucket").await.unwrap(); + assert!( + fs.get_bucket_lifecycle("it-bucket") + .await + .unwrap() + .is_empty(), + "lifecycle configuration must be gone after delete" + ); + // --- policies --- let document = json!({ "Version": "2012-10-17",