diff --git a/CHANGELOG.md b/CHANGELOG.md index 3331562..e9da64f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,7 +39,11 @@ All notable changes to RustFS Operator are documented in this file. The format i - Declarative Tenant bucket lifecycle rules with drift-aware ownership and explicit removal. - Bucket lifecycle `transition` and `noncurrentVersionTransition` actions that move objects to a remote tier registered in RustFS. +- Tenant `spec.tiers` for declarative S3, RustFS, and MinIO remote tiers with Secret-sourced + credentials, rotation on Secret updates, and conflict reporting for immutable tier settings. - Declarative Tenant bucket versioning and Object Lock default retention with drift-aware ownership. +- Declarative default bucket encryption (`SSE-S3` or `SSE-KMS`) with drift-aware ownership and + explicit removal. - Helm chart `cosiDriver.enabled` to deploy the RustFS COSI (`rustfs.objectstorage.k8s.io`) driver alongside the upstream provisioner sidecar, with dedicated ServiceAccount/RBAC. - Seven built-in RustFS Grafana dashboards, published by default as sidecar-discoverable diff --git a/console-web/app/(dashboard)/tenants/[namespace]/[name]/tenant-detail-client.tsx b/console-web/app/(dashboard)/tenants/[namespace]/[name]/tenant-detail-client.tsx index 0850ab5..3eca432 100644 --- a/console-web/app/(dashboard)/tenants/[namespace]/[name]/tenant-detail-client.tsx +++ b/console-web/app/(dashboard)/tenants/[namespace]/[name]/tenant-detail-client.tsx @@ -104,6 +104,7 @@ function provisioningGroups(tenant: TenantDetailsResponse) { return [ { type: "Policy", items: provisioning.policies ?? [] }, { type: "User", items: provisioning.users ?? [] }, + { type: "Tier", items: provisioning.tiers ?? [] }, { type: "Bucket", items: provisioning.buckets ?? [] }, ].filter((group) => group.items.length > 0) } diff --git a/console-web/types/api.ts b/console-web/types/api.ts index 78a860f..4866b9a 100644 --- a/console-web/types/api.ts +++ b/console-web/types/api.ts @@ -130,16 +130,37 @@ export interface ProvisioningBucket { objectLockConfiguration?: BucketObjectLockConfiguration anonymous?: "Private" | "Download" | "Upload" | "Public" policy?: PolicyDocumentSource + encryption?: BucketEncryptionSpec lifecycle?: BucketLifecycleSpec deletionPolicy?: ProvisioningDeletionPolicy } +export interface ProvisioningTier { + name: string + type: "S3" | "RustFS" | "MinIO" + endpoint: string + bucket: string + prefix?: string + region?: string + credsSecret: { + name: string + } + force?: boolean + deletionPolicy?: ProvisioningDeletionPolicy +} + export interface BucketObjectLockConfiguration { state?: "Present" | "Absent" mode?: "Governance" | "Compliance" days?: number } +export interface BucketEncryptionSpec { + state?: "Present" | "Absent" + mode?: "SSE-S3" | "SSE-KMS" + kmsKeyId?: string +} + export interface BucketLifecycleSpec { state: "Present" | "Absent" rules?: BucketLifecycleRule[] @@ -190,6 +211,8 @@ export interface ProvisioningBucketStatus extends ProvisioningItemStatus { versioning?: "Unversioned" | "Enabled" | "Suspended" | null objectLockConfigurationDesiredHash?: string | null objectLockConfigurationLastAppliedHash?: string | null + encryptionDesiredHash?: string | null + encryptionLastAppliedHash?: string | null } export interface ProvisioningStatus { @@ -198,6 +221,7 @@ export interface ProvisioningStatus { policies?: ProvisioningItemStatus[] users?: ProvisioningItemStatus[] buckets?: ProvisioningBucketStatus[] + tiers?: ProvisioningItemStatus[] } export interface CreatePoolRequest { @@ -225,6 +249,7 @@ export interface CreateTenantRequest { policies?: ProvisioningPolicy[] users?: ProvisioningUser[] buckets?: ProvisioningBucket[] + tiers?: ProvisioningTier[] security_context?: CreateSecurityContextRequest } @@ -238,6 +263,7 @@ export interface UpdateTenantRequest { policies?: ProvisioningPolicy[] users?: ProvisioningUser[] buckets?: ProvisioningBucket[] + tiers?: ProvisioningTier[] logging?: { logType: string volumeSize?: string diff --git a/crates/rustfs-admin/src/admin_ops.rs b/crates/rustfs-admin/src/admin_ops.rs index 1ac7079..0982d39 100644 --- a/crates/rustfs-admin/src/admin_ops.rs +++ b/crates/rustfs-admin/src/admin_ops.rs @@ -20,13 +20,13 @@ use std::collections::{BTreeMap, BTreeSet}; use super::helpers::{ body_mentions_not_found, build_canonical_query, extract_canned_policy_document, - is_absent_resource, parse_tier_names, + is_absent_resource, parse_tiers, uri_encode, }; use super::{ ADD_CANNED_POLICY_PATH, ADD_USER_PATH, ADMIN_SIGNING_SERVICE, INFO_CANNED_POLICY_PATH, - JSON_CONTENT_TYPE, LIST_CANNED_POLICIES_PATH, LIST_TIERS_PATH, REMOVE_USER_PATH, - RustfsAdminClient, RustfsClientError, RustfsServerInfo, RustfsServerInfoResponse, - RustfsUserInfo, SERVER_INFO_PATH, SET_POLICY_PATH, USER_INFO_PATH, + JSON_CONTENT_TYPE, LIST_CANNED_POLICIES_PATH, REMOVE_USER_PATH, RustfsAdminClient, + RustfsClientError, RustfsServerInfo, RustfsServerInfoResponse, RustfsTier, RustfsTierConfig, + RustfsUserInfo, SERVER_INFO_PATH, SET_POLICY_PATH, TIER_PATH, USER_INFO_PATH, }; use reqwest::StatusCode; use serde_json::Value; @@ -175,13 +175,94 @@ impl RustfsAdminClient { .collect() } - /// Names of the remote tiers registered in RustFS. Only names are kept; the response also - /// carries tier endpoints and credentials, which are dropped while parsing. + /// Names of the remote tiers registered in RustFS. pub async fn list_tier_names(&self) -> Result, RustfsClientError> { + Ok(self + .list_tiers() + .await? + .into_iter() + .map(|tier| tier.config.name) + .collect()) + } + + /// Remote tiers registered in RustFS. Secret keys in the response are dropped while parsing. + pub async fn list_tiers(&self) -> Result, RustfsClientError> { let body = self - .send_admin_request("GET", LIST_TIERS_PATH, "", "", None) + .send_admin_request("GET", TIER_PATH, "", "", None) .await?; - parse_tier_names(&body) + parse_tiers(&body) + } + + /// Register an S3-style remote tier. Without `force`, RustFS rejects a remote prefix that + /// already holds objects with HTTP 409. + pub async fn add_tier( + &self, + config: &RustfsTierConfig, + access_key: &str, + secret_key: &str, + force: bool, + ) -> Result<(), RustfsClientError> { + if config.name.trim().is_empty() || config.tier_type.trim().is_empty() { + return Err(RustfsClientError::InvalidTierName); + } + if access_key.trim().is_empty() { + return Err(RustfsClientError::InvalidCredentialValue { key: "accesskey" }); + } + if secret_key.is_empty() { + return Err(RustfsClientError::EmptyCredentialValue { key: "secretkey" }); + } + + let mut body = serde_json::Map::new(); + body.insert("type".to_string(), Value::from(config.tier_type.as_str())); + body.insert("name".to_string(), Value::from(config.name.as_str())); + body.insert( + config.tier_type.clone(), + serde_json::json!({ + "name": config.name, + "endpoint": config.endpoint, + "accessKey": access_key, + "secretKey": secret_key, + "bucket": config.bucket, + "prefix": config.prefix, + "region": config.region, + }), + ); + let body = Value::Object(body).to_string(); + let query = build_canonical_query(&[("force", if force { "true" } else { "false" })]); + + self.send_admin_request("PUT", TIER_PATH, &query, &body, Some(JSON_CONTENT_TYPE)) + .await + .map(|_| ()) + } + + /// Replace the credentials of a registered tier. RustFS only allows credentials to change + /// on an existing tier; endpoint, bucket and prefix are fixed at registration. + pub async fn edit_tier_credentials( + &self, + tier_name: &str, + access_key: &str, + secret_key: &str, + ) -> Result<(), RustfsClientError> { + if tier_name.trim().is_empty() { + return Err(RustfsClientError::InvalidTierName); + } + if access_key.trim().is_empty() { + return Err(RustfsClientError::InvalidCredentialValue { key: "accesskey" }); + } + if secret_key.is_empty() { + return Err(RustfsClientError::EmptyCredentialValue { key: "secretkey" }); + } + + let body = serde_json::json!({ + "access": access_key, + "secret": secret_key, + }) + .to_string(); + let path = format!("{TIER_PATH}/{}", uri_encode(tier_name)); + + self.send_admin_request("POST", &path, "", &body, Some(JSON_CONTENT_TYPE)) + .await + .map(|_| ()) } pub async fn server_info(&self) -> Result { diff --git a/crates/rustfs-admin/src/helpers.rs b/crates/rustfs-admin/src/helpers.rs index fb5b008..2e740fa 100644 --- a/crates/rustfs-admin/src/helpers.rs +++ b/crates/rustfs-admin/src/helpers.rs @@ -13,7 +13,6 @@ // limitations under the License. //! Internal helper duties: shared credential parsing, signature/hash utilities, and parsers. -use std::collections::BTreeSet; use hmac::{Hmac, Mac}; use reqwest::StatusCode; @@ -21,7 +20,7 @@ use serde_json::Value; use sha2::{Digest, Sha256}; use url::form_urlencoded; -use super::RustfsClientError; +use super::{RustfsClientError, RustfsTier, RustfsTierConfig}; /// Encode an `application/x-www-form-urlencoded` request body. pub(super) fn build_form_body(params: &[(&str, &str)]) -> String { @@ -54,7 +53,7 @@ pub(super) fn build_canonical_query(params: &[(&str, &str)]) -> String { .join("&") } -fn uri_encode(value: &str) -> String { +pub(super) fn uri_encode(value: &str) -> String { const HEX: &[u8; 16] = b"0123456789ABCDEF"; let mut encoded = String::with_capacity(value.len()); @@ -102,6 +101,7 @@ pub(super) fn body_mentions_not_found(body: &str) -> bool { || body.contains("nosuchpolicy") || body.contains("no such policy") || body.contains("objectlockconfigurationnotfound") + || body.contains("serversideencryptionconfigurationnotfound") || body.contains("nosuchbucketpolicy") || body.contains("no such bucket policy") || body.contains("not found") @@ -153,12 +153,13 @@ pub(super) fn extract_canned_policy_document(body: &str) -> Result", "": {"name": ..}}` -/// entries where the backend object is keyed by the tier type. -pub(super) fn parse_tier_names(body: &str) -> Result, RustfsClientError> { +/// entries where the backend object is keyed by the tier type. `secretKey` is dropped and +/// never leaves this function. +pub(super) fn parse_tiers(body: &str) -> Result, RustfsClientError> { let tiers = match serde_json::from_str::(body) .map_err(|_| RustfsClientError::ParseResponseFailed)? { - Value::Null => return Ok(BTreeSet::new()), + Value::Null => return Ok(Vec::new()), Value::Array(tiers) => tiers, _ => return Err(RustfsClientError::ParseResponseFailed), }; @@ -169,12 +170,31 @@ pub(super) fn parse_tier_names(body: &str) -> Result, RustfsCli .get("type") .and_then(Value::as_str) .ok_or(RustfsClientError::ParseResponseFailed)?; - tier.get(tier_type) - .and_then(|backend| backend.get("name")) - .and_then(Value::as_str) - .filter(|name| !name.is_empty()) - .map(ToOwned::to_owned) - .ok_or(RustfsClientError::ParseResponseFailed) + let backend = tier + .get(tier_type) + .ok_or(RustfsClientError::ParseResponseFailed)?; + let field = |key: &str| { + backend + .get(key) + .and_then(Value::as_str) + .unwrap_or_default() + .to_string() + }; + let name = field("name"); + if name.is_empty() { + return Err(RustfsClientError::ParseResponseFailed); + } + Ok(RustfsTier { + config: RustfsTierConfig { + tier_type: tier_type.to_string(), + name, + endpoint: field("endpoint"), + bucket: field("bucket"), + prefix: field("prefix"), + region: field("region"), + }, + access_key: field("accessKey"), + }) }) .collect() } diff --git a/crates/rustfs-admin/src/lib.rs b/crates/rustfs-admin/src/lib.rs index 09a673b..f067484 100644 --- a/crates/rustfs-admin/src/lib.rs +++ b/crates/rustfs-admin/src/lib.rs @@ -33,6 +33,7 @@ mod pool_ops; #[path = "s3_ops.rs"] mod s3_ops; pub use s3_ops::{ + BucketEncryptionAlgorithm, BucketEncryptionConfiguration, BucketLifecycleAbortIncompleteMultipartUpload, BucketLifecycleConfiguration, BucketLifecycleExpiration, BucketLifecycleNoncurrentVersionTransition, BucketLifecycleRule, BucketLifecycleRuleStatus, BucketLifecycleTransition, BucketObjectLockConfiguration, @@ -55,7 +56,7 @@ const LIST_CANNED_POLICIES_PATH: &str = "/rustfs/admin/v3/list-canned-policies"; const ADD_CANNED_POLICY_PATH: &str = "/rustfs/admin/v3/add-canned-policy"; const INFO_CANNED_POLICY_PATH: &str = "/rustfs/admin/v3/info-canned-policy"; const SERVER_INFO_PATH: &str = "/rustfs/admin/v3/info"; -const LIST_TIERS_PATH: &str = "/rustfs/admin/v3/tier"; +const TIER_PATH: &str = "/rustfs/admin/v3/tier"; const POOLS_LIST_PATH: &str = "/rustfs/admin/v3/pools/list"; const POOLS_STATUS_PATH: &str = "/rustfs/admin/v3/pools/status"; const POOLS_DECOMMISSION_PATH: &str = "/rustfs/admin/v3/pools/decommission"; @@ -120,6 +121,24 @@ pub struct RustfsUserInfo { pub policy_names: Vec, } +/// Non-secret settings of an S3-style remote tier (`s3`, `rustfs` and `minio` backends). +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct RustfsTierConfig { + pub tier_type: String, + pub name: String, + pub endpoint: String, + pub bucket: String, + pub prefix: String, + pub region: String, +} + +/// Remote tier listed by `GET /rustfs/admin/v3/tier`. The secret key is never kept. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RustfsTier { + pub config: RustfsTierConfig, + pub access_key: String, +} + #[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] pub struct RustfsPoolDecommissionInfo { #[serde(rename = "startTime")] @@ -217,6 +236,7 @@ pub enum RustfsClientError { }, TenantSecretLookupFailed, InvalidPolicyName, + InvalidTierName, InvalidPolicyDocument, TenantTlsRequired, TenantTlsNotReady, @@ -238,6 +258,7 @@ pub enum RustfsClientError { }, InvalidBucketVersioningResponse, InvalidObjectLockConfigurationResponse, + InvalidBucketEncryptionResponse, InvalidLifecycleConfigurationResponse, ParseResponseFailed, SigningFailed, @@ -257,6 +278,7 @@ impl std::fmt::Display for RustfsClientError { write!(f, "failed to load tenant credential secret") } Self::InvalidPolicyName => write!(f, "invalid policy name"), + Self::InvalidTierName => write!(f, "invalid tier name"), Self::InvalidPolicyDocument => write!(f, "failed to parse canned policy response"), Self::TenantTlsRequired => write!(f, "STS requires a TLS-enabled tenant"), Self::TenantTlsNotReady => write!(f, "tenant TLS status is not ready"), @@ -289,6 +311,9 @@ impl std::fmt::Display for RustfsClientError { Self::InvalidObjectLockConfigurationResponse => { write!(f, "invalid bucket Object Lock configuration response") } + Self::InvalidBucketEncryptionResponse => { + write!(f, "invalid bucket encryption configuration response") + } Self::ParseResponseFailed => write!(f, "failed to parse AssumeRole response"), Self::SigningFailed => write!(f, "failed to compute request signature"), } @@ -305,6 +330,11 @@ impl RustfsClientError { matches!(self, Self::UnexpectedStatus { status, .. } if *status == StatusCode::NOT_FOUND) } + /// Whether RustFS answered HTTP 409, e.g. a tier whose remote prefix already holds objects. + pub fn is_conflict(&self) -> bool { + matches!(self, Self::UnexpectedStatus { status, .. } if *status == StatusCode::CONFLICT) + } + /// Network, timeout, and retryable upstream failures. Permanent configuration and /// 4xx semantic errors are not transient. pub fn is_transient(&self) -> bool { @@ -326,6 +356,7 @@ impl RustfsClientError { | Self::EmptyCredentialValue { .. } | Self::InvalidCredentialValue { .. } | Self::InvalidPolicyName + | Self::InvalidTierName | Self::InvalidPolicyDocument | Self::TenantTlsRequired | Self::TenantTlsClientCertificateRequired @@ -335,6 +366,7 @@ impl RustfsClientError { | Self::RequestBuildFailed | Self::InvalidBucketVersioningResponse | Self::InvalidObjectLockConfigurationResponse + | Self::InvalidBucketEncryptionResponse | Self::InvalidLifecycleConfigurationResponse | Self::SigningFailed => false, } @@ -559,7 +591,16 @@ impl RustfsAdminClient { #[cfg(test)] mod tests { - use super::helpers::{build_canonical_query, parse_tier_names}; + use super::helpers::{build_canonical_query, parse_tiers}; + use super::{RustfsTier, RustfsTierConfig}; + use std::collections::BTreeSet; + + fn parse_tier_names(body: &str) -> Result, RustfsClientError> { + Ok(parse_tiers(body)? + .into_iter() + .map(|tier| tier.config.name) + .collect()) + } use super::{RustfsClientError, StatusCode}; #[test] @@ -579,6 +620,44 @@ mod tests { assert!(parse_tier_names("null").expect("null list").is_empty()); } + #[test] + fn tier_settings_are_read_without_the_secret_key() { + let body = r#"[ + {"type":"s3","s3":{"name":"COLD","endpoint":"https://gateway.example.com","accessKey":"AK","secretKey":"REDACTED","bucket":"cold","prefix":"rustfs/","region":"us-1","storageClass":""}}, + {"type":"minio","minio":{"name":"ARCHIVE","endpoint":"https://minio.example.com","accessKey":"AK2","secretKey":"REDACTED","bucket":"archive"}} + ]"#; + + let tiers = parse_tiers(body).expect("tier list parses"); + assert_eq!( + tiers, + vec![ + RustfsTier { + config: RustfsTierConfig { + tier_type: "s3".to_string(), + name: "COLD".to_string(), + endpoint: "https://gateway.example.com".to_string(), + bucket: "cold".to_string(), + prefix: "rustfs/".to_string(), + region: "us-1".to_string(), + }, + access_key: "AK".to_string(), + }, + RustfsTier { + config: RustfsTierConfig { + tier_type: "minio".to_string(), + name: "ARCHIVE".to_string(), + endpoint: "https://minio.example.com".to_string(), + bucket: "archive".to_string(), + prefix: String::new(), + region: String::new(), + }, + access_key: "AK2".to_string(), + }, + ] + ); + assert!(!format!("{tiers:?}").contains("REDACTED")); + } + #[test] fn tier_list_without_a_backend_name_is_rejected() { for body in [ diff --git a/crates/rustfs-admin/src/s3_ops.rs b/crates/rustfs-admin/src/s3_ops.rs index ff43c3e..0ad6ec8 100644 --- a/crates/rustfs-admin/src/s3_ops.rs +++ b/crates/rustfs-admin/src/s3_ops.rs @@ -14,6 +14,7 @@ //! S3 boundary: //! - bucket lifecycle methods (create/lookup features) +//! - default bucket encryption (`?encryption`) //! - request semantics for S3-style object storage operations. use super::helpers::{ @@ -140,6 +141,52 @@ impl BucketObjectLockConfiguration { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BucketEncryptionAlgorithm { + /// SSE-S3 (`AES256`). + Aes256, + /// SSE-KMS (`aws:kms`). + AwsKms, +} + +impl BucketEncryptionAlgorithm { + const fn as_s3_str(self) -> &'static str { + match self { + Self::Aes256 => "AES256", + Self::AwsKms => "aws:kms", + } + } +} + +/// Default bucket encryption (`ServerSideEncryptionConfiguration` with a single rule). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BucketEncryptionConfiguration { + pub algorithm: BucketEncryptionAlgorithm, + pub kms_master_key_id: Option, + pub bucket_key_enabled: bool, +} + +impl BucketEncryptionConfiguration { + pub fn to_xml(&self) -> String { + let mut xml = format!( + "" + ); + xml.push_str(self.algorithm.as_s3_str()); + xml.push_str(""); + if let Some(key_id) = &self.kms_master_key_id { + xml.push_str(""); + xml.push_str(&escape_xml(key_id)); + xml.push_str(""); + } + xml.push_str(""); + if self.bucket_key_enabled { + xml.push_str("true"); + } + xml.push_str(""); + xml + } +} + #[derive(Debug)] struct StrictXmlElement { name: String, @@ -257,6 +304,58 @@ fn parse_bucket_object_lock_xml( }) } +fn parse_bucket_encryption_xml( + xml: &str, +) -> Result { + let invalid = || RustfsClientError::InvalidBucketEncryptionResponse; + let root = parse_strict_configuration_xml(xml, "ServerSideEncryptionConfiguration") + .map_err(|()| invalid())?; + root.validate_children(&["Rule"]).map_err(|()| invalid())?; + let rule = root + .child("Rule") + .map_err(|()| invalid())? + .ok_or_else(invalid)?; + rule.validate_children(&["ApplyServerSideEncryptionByDefault", "BucketKeyEnabled"]) + .map_err(|()| invalid())?; + let by_default = rule + .child("ApplyServerSideEncryptionByDefault") + .map_err(|()| invalid())? + .ok_or_else(invalid)?; + by_default + .validate_children(&["SSEAlgorithm", "KMSMasterKeyID"]) + .map_err(|()| invalid())?; + let algorithm = match by_default + .child("SSEAlgorithm") + .map_err(|()| invalid())? + .ok_or_else(invalid)? + .leaf_text() + .map_err(|()| invalid())? + { + "AES256" => BucketEncryptionAlgorithm::Aes256, + "aws:kms" => BucketEncryptionAlgorithm::AwsKms, + _ => return Err(invalid()), + }; + let kms_master_key_id = match by_default.child("KMSMasterKeyID").map_err(|()| invalid())? { + Some(key_id) if key_id.text.trim().is_empty() && key_id.children.is_empty() => None, + Some(key_id) => Some(key_id.leaf_text().map_err(|()| invalid())?.to_string()), + None => None, + }; + let bucket_key_enabled = match rule.child("BucketKeyEnabled").map_err(|()| invalid())? { + Some(enabled) => match enabled.leaf_text().map_err(|()| invalid())? { + "true" => true, + "false" => false, + _ => return Err(invalid()), + }, + None => false, + }; + + Ok(BucketEncryptionConfiguration { + algorithm, + kms_master_key_id, + bucket_key_enabled, + }) +} + fn parse_positive_i32(value: &str) -> Option { value.parse::().ok().filter(|value| *value > 0) } @@ -1034,6 +1133,139 @@ impl RustfsAdminClient { .map(|configuration| configuration.is_some()) } + pub async fn get_bucket_encryption( + &self, + bucket: &str, + ) -> Result, RustfsClientError> { + if bucket.trim().is_empty() { + return Err(RustfsClientError::RequestBuildFailed); + } + + let path = format!("/{bucket}"); + let query = build_canonical_query(&[("encryption", "")]); + let signed = self.sign_request("GET", &path, &query, "", None, ADMIN_SIGNING_SERVICE)?; + let host = self.host()?; + let response = self + .http_client + .get(format!( + "{}{}?{query}", + self.base_url.trim_end_matches('/'), + path + )) + .header("x-amz-date", &signed.amz_date) + .header("x-amz-content-sha256", &signed.payload_hash) + .header("authorization", &signed.authorization) + .header("host", host) + .send() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + + if response.status().is_success() { + let body = read_bucket_configuration_response(response, || { + RustfsClientError::InvalidBucketEncryptionResponse + }) + .await?; + return parse_bucket_encryption_xml(&body).map(Some); + } + + // RustFS answers a missing configuration with 400 ServerSideEncryptionConfigurationNotFoundError. + let status = response.status(); + let (body, truncated) = RustfsClientError::limited_response_body(response).await; + if is_absent_resource(status, &body) { + return Ok(None); + } + Err(RustfsClientError::unexpected_status_with_limited_body( + status, &body, truncated, + )) + } + + pub async fn put_bucket_encryption( + &self, + bucket: &str, + configuration: &BucketEncryptionConfiguration, + ) -> Result<(), RustfsClientError> { + if bucket.trim().is_empty() { + return Err(RustfsClientError::RequestBuildFailed); + } + + let path = format!("/{bucket}"); + let query = build_canonical_query(&[("encryption", "")]); + let body = configuration.to_xml(); + let content_md5 = content_md5(&body); + let signed = self.sign_request_with_extra_headers( + "PUT", + &path, + &query, + &body, + ADMIN_SIGNING_SERVICE, + &[ + ("content-md5", content_md5.as_str()), + ("content-type", "application/xml"), + ], + )?; + let host = self.host()?; + let response = self + .http_client + .put(format!( + "{}{}?{query}", + self.base_url.trim_end_matches('/'), + path + )) + .header("x-amz-date", &signed.amz_date) + .header("x-amz-content-sha256", &signed.payload_hash) + .header("authorization", &signed.authorization) + .header("host", host) + .header("content-md5", content_md5) + .header("content-type", "application/xml") + .body(body) + .send() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + + if response.status().is_success() { + Ok(()) + } else { + Err(RustfsClientError::unexpected_response(response).await) + } + } + + pub async fn delete_bucket_encryption(&self, bucket: &str) -> Result<(), RustfsClientError> { + if bucket.trim().is_empty() { + return Err(RustfsClientError::RequestBuildFailed); + } + + let path = format!("/{bucket}"); + let query = build_canonical_query(&[("encryption", "")]); + let signed = self.sign_request("DELETE", &path, &query, "", None, ADMIN_SIGNING_SERVICE)?; + let host = self.host()?; + let response = self + .http_client + .delete(format!( + "{}{}?{query}", + self.base_url.trim_end_matches('/'), + path + )) + .header("x-amz-date", &signed.amz_date) + .header("x-amz-content-sha256", &signed.payload_hash) + .header("authorization", &signed.authorization) + .header("host", host) + .send() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + + if response.status().is_success() { + return Ok(()); + } + let status = response.status(); + let (body, truncated) = RustfsClientError::limited_response_body(response).await; + if is_absent_resource(status, &body) { + return Ok(()); + } + Err(RustfsClientError::unexpected_status_with_limited_body( + status, &body, truncated, + )) + } + pub async fn put_bucket_policy( &self, bucket: &str, @@ -1415,6 +1647,68 @@ mod tests { )); } + #[test] + fn bucket_encryption_xml_matches_rustfs_get_response_shape() { + // GetBucketEncryption body from RustFS 1.0.2-preview.2 after `PUT ?encryption` with AES256. + let rustfs_get = "AES256"; + let sse_s3 = BucketEncryptionConfiguration { + algorithm: BucketEncryptionAlgorithm::Aes256, + kms_master_key_id: None, + bucket_key_enabled: false, + }; + assert_eq!(parse_bucket_encryption_xml(rustfs_get).unwrap(), sse_s3); + assert_eq!( + sse_s3.to_xml(), + "AES256" + ); + assert_eq!( + parse_bucket_encryption_xml(&sse_s3.to_xml()).unwrap(), + sse_s3 + ); + + let sse_kms = BucketEncryptionConfiguration { + algorithm: BucketEncryptionAlgorithm::AwsKms, + kms_master_key_id: Some("app-key".to_string()), + bucket_key_enabled: true, + }; + assert!( + sse_kms + .to_xml() + .contains("app-key") + ); + assert_eq!( + parse_bucket_encryption_xml(&sse_kms.to_xml()).unwrap(), + sse_kms + ); + } + + #[test] + fn bucket_encryption_xml_rejects_unknown_ambiguous_or_missing_fields() { + for xml in [ + "", + "", + "AES128", + "AES256AES256", + "AES256", + "AES256yes", + "", + ] { + assert!( + matches!( + parse_bucket_encryption_xml(xml), + Err(RustfsClientError::InvalidBucketEncryptionResponse) + ), + "{xml} must be rejected" + ); + } + let empty_key = parse_bucket_encryption_xml( + "aws:kmsfalse", + ) + .unwrap(); + assert_eq!(empty_key.kms_master_key_id, None); + assert!(!empty_key.bucket_key_enabled); + } + #[test] fn lifecycle_xml_is_stable_and_escapes_values() { let configuration = BucketLifecycleConfiguration { diff --git a/deploy/rustfs-operator/README.md b/deploy/rustfs-operator/README.md index 908a986..68c90aa 100755 --- a/deploy/rustfs-operator/README.md +++ b/deploy/rustfs-operator/README.md @@ -298,7 +298,7 @@ field updates the StatefulSet Pod template and starts a rolling update. ### Tenant Provisioning -Tenants can declare RustFS canned policies, regular users, and buckets directly in `spec.policies`, `spec.users`, and `spec.buckets`. Provisioning starts only after the Tenant workload is ready, uses `spec.credsSecret` as the RustFS admin credential source, and reports progress under `status.provisioning`. +Tenants can declare RustFS canned policies, regular users, remote tiers, and buckets directly in `spec.policies`, `spec.users`, `spec.tiers`, and `spec.buckets`. Provisioning starts only after the Tenant workload is ready, uses `spec.credsSecret` as the RustFS admin credential source, and reports progress under `status.provisioning`. User provisioning requires a non-empty direct policy mapping: @@ -336,7 +336,7 @@ spec: days: 30 ``` -Policy ConfigMaps and user Secrets must live in the Tenant namespace. `users[].credsSecret.name` selects the credentials Secret; when omitted, the operator falls back to a Secret named after `users[].name` for compatibility with existing manifests. Bucket versioning can be enabled or suspended. Object Lock automatically requires enabled versioning and cannot be disabled after activation; `objectLockConfiguration` manages an optional default `Governance` or `Compliance` retention period in days. Omitting the configuration leaves an existing rule unmanaged, while `state: Absent` removes only an operator-owned default rule. Bucket lifecycle supports expiration, incomplete multipart upload cleanup, and transition of current or noncurrent versions to a remote tier registered in RustFS. Omission leaves lifecycle unmanaged; `state: Absent` deletes only a configuration whose live hash still matches the Operator's ownership record. The operator indexes references from Tenant specs, so creating or updating a referenced object enqueues every referencing Tenant without requiring or mutating labels or requiring write access to that object. Provisioned resources are retained when removed from the Tenant spec. +Policy ConfigMaps, user Secrets, and tier Secrets must live in the Tenant namespace. `users[].credsSecret.name` selects the credentials Secret; when omitted, the operator falls back to a Secret named after `users[].name` for compatibility with existing manifests. Bucket versioning can be enabled or suspended. Object Lock automatically requires enabled versioning and cannot be disabled after activation; `objectLockConfiguration` manages an optional default `Governance` or `Compliance` retention period in days. Omitting the configuration leaves an existing rule unmanaged, while `state: Absent` removes only an operator-owned default rule. `encryption` sets the bucket default encryption (`SSE-S3`, or `SSE-KMS` with an optional `kmsKeyId` when the Tenant enables KMS) with the same ownership rules. Bucket lifecycle supports expiration, incomplete multipart upload cleanup, and transition of current or noncurrent versions to a remote tier registered in RustFS. `spec.tiers` registers such tiers before buckets are reconciled; tier credentials come from `tiers[].credsSecret` and are rotated when the Secret changes, while a live tier with different immutable settings is reported as a conflict. Omission leaves lifecycle unmanaged; `state: Absent` deletes only a configuration whose live hash still matches the Operator's ownership record. The operator indexes references from Tenant specs, so creating or updating a referenced object enqueues every referencing Tenant without requiring or mutating labels or requiring write access to that object. Provisioned resources are retained when removed from the Tenant spec. ### RBAC Configuration diff --git a/deploy/rustfs-operator/crds/tenant-crd.yaml b/deploy/rustfs-operator/crds/tenant-crd.yaml index f7cbbf9..c4778d8 100644 --- a/deploy/rustfs-operator/crds/tenant-crd.yaml +++ b/deploy/rustfs-operator/crds/tenant-crd.yaml @@ -1235,6 +1235,43 @@ spec: enum: - Retain type: string + encryption: + description: |- + Default bucket encryption. Omission leaves the live configuration unmanaged; `Absent` + removes only a configuration previously owned by the operator. + nullable: true + properties: + kmsKeyId: + description: KMS key for `SSE-KMS`. Omission uses the RustFS default KMS key. + maxLength: 1024 + minLength: 1 + nullable: true + pattern: ^[A-Za-z0-9._:/+=@-]+$ + type: string + mode: + description: |- + `SSE-S3` stores S3 `AES256`; `SSE-KMS` stores `aws:kms` and requires + `spec.encryption.enabled` on the Tenant. + enum: + - SSE-S3 + - SSE-KMS + - null + nullable: true + type: string + state: + description: Omission is equivalent to `Present`. + enum: + - Present + - Absent + - null + nullable: true + type: string + type: object + x-kubernetes-validations: + - message: Present bucket encryption requires mode; Absent forbids mode and kmsKeyId + rule: '(!has(self.state) || self.state == ''Present'') ? has(self.mode) : !has(self.mode) && !has(self.kmsKeyId)' + - message: kmsKeyId requires mode SSE-KMS + rule: '!has(self.kmsKeyId) || (has(self.mode) && self.mode == ''SSE-KMS'')' lifecycle: description: |- Declarative S3 bucket lifecycle configuration. Omission leaves the live lifecycle @@ -3188,6 +3225,76 @@ spec: serviceAccountName: nullable: true type: string + tiers: + description: |- + Remote storage tiers that bucket lifecycle transitions can reference by name. + Tiers are reconciled before buckets. + items: + description: Remote storage tier registered in RustFS for bucket lifecycle transitions. + properties: + bucket: + description: Remote bucket that receives transitioned objects. + minLength: 1 + type: string + credsSecret: + description: |- + Secret with `accesskey` and `secretkey` for the remote bucket. Updating the Secret + rotates the tier credentials in RustFS. + properties: + name: + maxLength: 253 + minLength: 1 + type: string + required: + - name + type: object + deletionPolicy: + enum: + - Retain + type: string + endpoint: + description: Remote S3 endpoint URL, for example `https://gateway.storjshare.io`. + minLength: 1 + type: string + force: + description: |- + Register the tier even when the remote prefix already holds objects, for example a + prefix written by this Tenant before a restore. RustFS refuses such a prefix otherwise. + type: boolean + name: + description: Tier name, referenced as `storageClass` by bucket lifecycle transitions. + maxLength: 253 + minLength: 1 + type: string + x-kubernetes-validations: + - message: tier name must be uppercase + rule: self == self.upperAscii() + - message: tier name must not be a reserved storage class + rule: self != 'STANDARD' && self != 'REDUCED_REDUNDANCY' + prefix: + description: Object prefix inside the remote bucket. + type: string + region: + type: string + type: + description: Remote backend of a tier. All three use the S3 API with static credentials. + enum: + - S3 + - RustFS + - MinIO + type: string + required: + - bucket + - credsSecret + - endpoint + - name + - type + type: object + maxItems: 64 + type: array + x-kubernetes-list-map-keys: + - name + x-kubernetes-list-type: map tls: nullable: true properties: @@ -3926,6 +4033,12 @@ spec: desiredHash: nullable: true type: string + encryptionDesiredHash: + nullable: true + type: string + encryptionLastAppliedHash: + nullable: true + type: string lastAppliedAccessKeyHash: nullable: true type: string @@ -4052,6 +4165,56 @@ spec: - state type: object type: array + tiers: + items: + properties: + desiredHash: + nullable: true + type: string + lastAppliedAccessKeyHash: + nullable: true + type: string + lastAppliedGeneration: + format: int64 + nullable: true + type: integer + lastAppliedHash: + nullable: true + type: string + lastTransitionTime: + nullable: true + type: string + message: + nullable: true + type: string + name: + type: string + objectLock: + nullable: true + type: boolean + observedSecretName: + nullable: true + type: string + observedSecretResourceVersion: + nullable: true + type: string + policies: + items: + type: string + type: array + reason: + type: string + region: + nullable: true + type: string + state: + type: string + required: + - name + - reason + - state + type: object + type: array users: items: description: |- diff --git a/docs/operator-user-guide.md b/docs/operator-user-guide.md index f5350cf..9857509 100644 --- a/docs/operator-user-guide.md +++ b/docs/operator-user-guide.md @@ -952,35 +952,67 @@ spec: ### 7.8 Bootstrap Provisioning -The operator can create RustFS policies, users, and buckets after the Tenant workload is ready. Configure: +The operator can create RustFS policies, users, remote tiers, and buckets after the Tenant workload is ready. Configure: - `spec.credsSecret` for RustFS admin credentials. - `spec.policies` for canned policies sourced from ConfigMaps. - `spec.users` for regular users. Each user must have at least one direct policy mapping. +- `spec.tiers` for remote storage tiers that bucket lifecycle transitions move data to. - `spec.buckets` for buckets, versioning, Object Lock and default retention, canned anonymous access (`Private`, `Download`, `Upload`, `Public`), or a custom bucket policy ConfigMap. Non-private `anonymous` values and `policy` are mutually exclusive. Explicit `anonymous: Private` without `policy` removes a bucket policy previously applied by the Operator; it reports a conflict instead of deleting an unowned live policy. For compatibility, legacy `Private` plus `policy` configurations keep the custom policy. When both fields are omitted, the Operator does not change a live bucket policy and retains its ownership record for a later explicit `Private` transition. Bucket versioning and Object Lock are declarative but conservative. `versioning: true` enables versioning. `versioning: false` keeps a never-versioned bucket unchanged or requests S3 `Suspended` after versioning has been enabled; omitting the field leaves versioning unmanaged unless Object Lock requires it. `objectLock: true` enables Object Lock after versioning is enabled. Object Lock cannot be disabled, and the Operator refuses `objectLock: false` or versioning suspension when the live bucket has Object Lock enabled. `objectLockConfiguration` manages the default retention rule: omit `state` for `Present`, specify `mode: Governance` or `Compliance` with `days`, or use `state: Absent` without `mode` and `days` to remove only the default rule. Omitting the entire field leaves a live default-retention rule unmanaged. The Operator adopts an identical live rule, but does not replace or remove a different rule unless its hash matches the last configuration recorded in status. Default retention applies to new object versions; it does not retroactively lock existing versions. +Bucket default encryption is declared with `encryption`. `mode: SSE-S3` stores S3 `AES256`; `mode: SSE-KMS` stores `aws:kms` with an optional `kmsKeyId` (RustFS uses its default KMS key when omitted) and requires `spec.encryption.enabled` on the Tenant. Default encryption applies to objects written after it is set; it does not re-encrypt existing objects. `state: Absent` without `mode` and `kmsKeyId` removes the configuration. The Operator adopts a matching live configuration, but never replaces or removes a different one unless its hash matches the configuration last recorded in `status.provisioning.buckets`. Omitting `encryption` performs no encryption API calls and leaves the live configuration untouched. + Each bucket may also declare `lifecycle.state: Present` with one or more S3 lifecycle rules. The current schema supports expiration after a positive number of days, aborting incomplete multipart uploads after a positive number of days, `transition` of current versions after zero or more days, and `noncurrentVersionTransition` of noncurrent versions after zero or more days; every rule has a unique ID, an `Enabled` or `Disabled` status, and an explicit prefix filter (use `prefix: ""` for all objects). `lifecycle.state: Absent` requests deletion and cannot include rules. The Operator adopts an identical live configuration, but never replaces or deletes a different configuration without a matching ownership hash in `status.provisioning.buckets`. Omitting `lifecycle` performs no lifecycle API calls and leaves the live configuration untouched. -Transition actions move data to a remote tier. `storageClass` is the name of a tier already registered in RustFS (for example with `rc ilm tier add`); tier names are uppercase, and `STANDARD` and `REDUCED_REDUNDANCY` are reserved. The Operator does not register tiers. Before applying a lifecycle with transition actions it checks that every referenced tier exists; while one is missing, the bucket stays `Pending` with a message naming the tier and the Operator retries, so the tier can be registered after the rule is added: +Transition actions move data to a remote tier. `storageClass` is a tier name: a tier declared in `spec.tiers`, or one registered in RustFS by other means (for example with `rc ilm tier add`). Tier names are uppercase, and `STANDARD` and `REDUCED_REDUNDANCY` are reserved. Before applying a lifecycle with transition actions the Operator checks that every referenced tier exists; while one is missing, the bucket stays `Pending` with a message naming the tier and the Operator retries, so the tier can be registered after the rule is added. + +`spec.tiers` registers S3-compatible remote tiers (`type: S3`, `RustFS`, or `MinIO`). Tiers are reconciled before buckets, so a lifecycle rule can reference a tier declared in the same Tenant. Each tier sets the remote `endpoint` and `bucket`, an optional `prefix` and `region`, and `credsSecret`, a Secret in the Tenant namespace with `accesskey` and `secretkey`. RustFS does not return a tier's secret key, so the Operator records the Secret's `resourceVersion` in `status.provisioning.tiers` and updates the tier credentials when the Secret changes, when `credsSecret` points to another Secret, or when the access key listed by RustFS differs from the Secret. RustFS cannot change the type, endpoint, bucket, prefix, or region of a registered tier: a live tier with the same name and different settings is reported as `TierConflict` and left untouched, while one with matching settings is adopted. RustFS refuses to register a tier whose remote prefix already holds objects; set `force: true` only when that prefix belongs to this Tenant, for example after a restore. Tiers removed from the spec are retained in RustFS. ```yaml -lifecycle: - state: Present - rules: - - id: to-cold - status: Enabled - filter: - prefix: "" - transition: - days: 7 - storageClass: COLD +apiVersion: v1 +kind: Secret +metadata: + name: cold-tier + namespace: storage +type: Opaque +stringData: + accesskey: + secretkey: +--- +apiVersion: rustfs.com/v1alpha1 +kind: Tenant +metadata: + name: rustfs-a + namespace: storage +spec: + # credsSecret and pools as in the full example below + tiers: + - name: COLD + type: S3 + endpoint: https://s3.example.com + bucket: rustfs-cold + prefix: rustfs-a/ + credsSecret: + name: cold-tier + buckets: + - name: app-data + lifecycle: + state: Present + rules: + - id: to-cold + status: Enabled + filter: + prefix: "" + transition: + days: 7 + storageClass: COLD ``` Transient Kubernetes and RustFS admin/S3 failures (timeouts, 429, 5xx, connection errors, TLS not ready) leave provisioning items `Pending` and requeue instead of marking the Tenant `Failed`. Permanent 4xx configuration errors still fail and wait for a spec or object change. -ConfigMaps and user Secrets must live in the Tenant namespace. The Operator indexes references from Tenant specs, so creating or updating a referenced object enqueues every referencing Tenant without requiring or mutating labels or requiring write access to that object. +ConfigMaps, user Secrets, and tier Secrets must live in the Tenant namespace. The Operator indexes references from Tenant specs, so creating or updating a referenced object enqueues every referencing Tenant without requiring or mutating labels or requiring write access to that object. Policy documents are parsed by RustFS. Use S3 ARN resource patterns such as `arn:aws:s3:::bucket` and `arn:aws:s3:::bucket/*`; for all buckets, use `arn:aws:s3:::*`. A bare `Resource: "*"` is not accepted by RustFS policy parsing. @@ -1051,6 +1083,8 @@ spec: objectLockConfiguration: mode: Compliance days: 30 + encryption: + mode: SSE-S3 anonymous: Download lifecycle: state: Present diff --git a/docs/operator-user-guide.zh-CN.md b/docs/operator-user-guide.zh-CN.md index f24e35d..29eb49b 100644 --- a/docs/operator-user-guide.zh-CN.md +++ b/docs/operator-user-guide.zh-CN.md @@ -896,35 +896,67 @@ spec: ### 7.8 初始化 Provisioning -Operator 可以在 Tenant workload Ready 后自动创建 RustFS policy、user 和 bucket。需要配置: +Operator 可以在 Tenant workload Ready 后自动创建 RustFS policy、user、远程 tier 和 bucket。需要配置: - `spec.credsSecret`:RustFS 管理员凭据。 - `spec.policies`:从 ConfigMap 读取 policy document。 - `spec.users`:普通用户。每个 user 必须至少直接绑定一个 policy。 +- `spec.tiers`:bucket 生命周期迁移操作使用的远程存储层(tier)。 - `spec.buckets`:bucket、版本控制、Object Lock 及默认保留策略、匿名访问(`Private` / `Download` / `Upload` / `Public`),或来自 ConfigMap 的自定义 bucket policy。非 `Private` 的 `anonymous` 值与 `policy` 互斥。未设置 `policy` 时,显式设置 `anonymous: Private` 会删除此前由 Operator 应用的 bucket policy;若现有 policy 不归 Operator 管理,则报告冲突而不删除。为兼容旧对象,`Private` 与 `policy` 同时存在时继续以自定义 policy 为准。两者都省略时,Operator 不会改写已有 bucket policy,并会保留所有权记录供后续显式切换为 `Private` 时校验。 Bucket 版本控制和 Object Lock 采用声明式但保守的调和策略。`versioning: true` 启用版本控制。`versioning: false` 会让从未启用版本控制的 bucket 保持原状,或在版本控制已启用后请求 S3 `Suspended`;省略该字段时保持现状,除非 Object Lock 要求先启用版本控制。`objectLock: true` 会在版本控制启用后开启 Object Lock。Object Lock 不可关闭;当线上 bucket 已启用 Object Lock 时,Operator 会拒绝 `objectLock: false` 或暂停版本控制。`objectLockConfiguration` 管理默认保留规则:省略 `state` 等同于 `Present`,此时必须设置 `mode: Governance` 或 `Compliance` 以及 `days`;`state: Absent` 必须省略 `mode` 和 `days`,仅移除默认规则。完全省略 `objectLockConfiguration` 时,不管理已有默认保留规则。Operator 可以接管与声明完全相同的线上规则,但只有线上配置哈希仍与 status 中最后一次应用的配置一致时,才会替换或移除不同规则。默认保留策略只作用于之后创建的对象版本,不会追溯锁定已有版本。 +Bucket 默认加密通过 `encryption` 声明。`mode: SSE-S3` 写入 S3 `AES256`;`mode: SSE-KMS` 写入 `aws:kms`,可选 `kmsKeyId`(省略时 RustFS 使用默认 KMS key),并要求 Tenant 设置 `spec.encryption.enabled`。默认加密只作用于设置之后写入的对象,不会重新加密已有对象。`state: Absent`(不能包含 `mode` 和 `kmsKeyId`)表示移除该配置。Operator 可以接管与声明一致的线上配置,但只有线上配置哈希与 `status.provisioning.buckets` 中最后记录的配置一致时,才会替换或移除不同的配置。省略 `encryption` 时不会调用加密 API,也不会改动现有配置。 + 每个 bucket 还可以通过 `lifecycle.state: Present` 声明一个或多个 S3 生命周期规则。当前 schema 支持按正整数天数过期对象、按正整数天数清理未完成的分片上传、通过 `transition` 在零天或更多天后迁移当前版本,以及通过 `noncurrentVersionTransition` 在零天或更多天后迁移非当前版本;每条规则必须有唯一 ID、`Enabled` 或 `Disabled` 状态,并显式指定 prefix filter(`prefix: ""` 表示全部对象)。`lifecycle.state: Absent` 表示请求删除,且不能包含规则。Operator 可以接管与声明完全相同的现有配置,但没有 `status.provisioning.buckets` 中匹配的所有权哈希时,不会覆盖或删除不同的现有配置。省略 `lifecycle` 时不会调用 Lifecycle API,也不会改动现有配置。 -迁移类操作会把数据移动到远程层(tier)。`storageClass` 是 RustFS 中已注册的 tier 名称(例如通过 `rc ilm tier add` 注册);tier 名称必须为大写,`STANDARD` 和 `REDUCED_REDUNDANCY` 为保留名称。Operator 不负责注册 tier。在应用包含迁移操作的生命周期配置之前,Operator 会检查所有被引用的 tier 是否存在;如果有 tier 尚未注册,bucket 会保持 `Pending` 状态并在消息中列出该 tier,Operator 会持续重试,因此可以在添加规则之后再注册 tier: +迁移类操作会把数据移动到远程层(tier)。`storageClass` 是 tier 名称:可以是 `spec.tiers` 中声明的 tier,也可以是通过其他方式(例如 `rc ilm tier add`)在 RustFS 中注册的 tier。tier 名称必须为大写,`STANDARD` 和 `REDUCED_REDUNDANCY` 为保留名称。在应用包含迁移操作的生命周期配置之前,Operator 会检查所有被引用的 tier 是否存在;如果有 tier 尚未注册,bucket 会保持 `Pending` 状态并在消息中列出该 tier,Operator 会持续重试,因此可以在添加规则之后再注册 tier。 + +`spec.tiers` 用于注册兼容 S3 的远程 tier(`type: S3`、`RustFS` 或 `MinIO`)。tier 会先于 bucket 调和,因此生命周期规则可以引用同一 Tenant 中声明的 tier。每个 tier 需要设置远程 `endpoint` 和 `bucket`,可选的 `prefix` 和 `region`,以及 `credsSecret`:位于 Tenant namespace、包含 `accesskey` 和 `secretkey` 的 Secret。RustFS 不会返回 tier 的 secret key,因此 Operator 会在 `status.provisioning.tiers` 中记录 Secret 的 `resourceVersion`,并在 Secret 变更、`credsSecret` 指向另一个 Secret,或 RustFS 列出的 access key 与 Secret 不一致时更新 tier 凭据。RustFS 不能修改已注册 tier 的类型、endpoint、bucket、prefix 或 region:同名但配置不同的现有 tier 会报告为 `TierConflict` 且不会被改动;配置一致的现有 tier 会被接管。如果远程 prefix 中已有对象,RustFS 会拒绝注册该 tier;仅当该 prefix 属于此 Tenant 时(例如恢复之后)才设置 `force: true`。从 spec 中移除的 tier 会保留在 RustFS 中。 ```yaml -lifecycle: - state: Present - rules: - - id: to-cold - status: Enabled - filter: - prefix: "" - transition: - days: 7 - storageClass: COLD +apiVersion: v1 +kind: Secret +metadata: + name: cold-tier + namespace: storage +type: Opaque +stringData: + accesskey: + secretkey: +--- +apiVersion: rustfs.com/v1alpha1 +kind: Tenant +metadata: + name: rustfs-a + namespace: storage +spec: + # credsSecret 和 pools 参见下方完整示例 + tiers: + - name: COLD + type: S3 + endpoint: https://s3.example.com + bucket: rustfs-cold + prefix: rustfs-a/ + credsSecret: + name: cold-tier + buckets: + - name: app-data + lifecycle: + state: Present + rules: + - id: to-cold + status: Enabled + filter: + prefix: "" + transition: + days: 7 + storageClass: COLD ``` Kubernetes 或 RustFS 管理/S3 API 的瞬时失败(超时、429、5xx、连接错误、TLS 未就绪)会把 provisioning 条目保持为 `Pending` 并重新入队,而不会把 Tenant 标为 `Failed`。永久性 4xx 配置错误仍会失败并等待 spec 或对象变更。 -ConfigMap 和 user Secret 必须位于 Tenant namespace。Operator 会从 Tenant spec 建立反向引用索引,因此被引用资源的创建或更新会触发所有引用它的 Tenant reconcile;无需要求或修改资源标签,也不需要对这些资源拥有写权限。 +ConfigMap、user Secret 和 tier Secret 必须位于 Tenant namespace。Operator 会从 Tenant spec 建立反向引用索引,因此被引用资源的创建或更新会触发所有引用它的 Tenant reconcile;无需要求或修改资源标签,也不需要对这些资源拥有写权限。 Policy document 由 RustFS 解析。请使用 `arn:aws:s3:::bucket` 和 `arn:aws:s3:::bucket/*` 这类 S3 ARN resource 写法;如需匹配所有 bucket,请使用 `arn:aws:s3:::*`。RustFS policy parser 不接受裸 `Resource: "*"`。 @@ -995,6 +1027,8 @@ spec: objectLockConfiguration: mode: Compliance days: 30 + encryption: + mode: SSE-S3 anonymous: Download lifecycle: state: Present diff --git a/src/console/handlers/tenants.rs b/src/console/handlers/tenants.rs index 1f23588..8f7dcbc 100755 --- a/src/console/handlers/tenants.rs +++ b/src/console/handlers/tenants.rs @@ -327,6 +327,7 @@ fn tenant_from_create_request(req: CreateTenantRequest) -> Result { policies: req.policies.unwrap_or_default(), users: req.users.unwrap_or_default(), buckets: req.buckets.unwrap_or_default(), + tiers: req.tiers.unwrap_or_default(), security_context, ..Default::default() }, @@ -503,6 +504,11 @@ pub async fn update_tenant( updated_fields.push("buckets".to_string()); } + if let Some(tiers) = req.tiers { + tenant.spec.tiers = tiers; + updated_fields.push("tiers".to_string()); + } + if updated_fields.is_empty() { return Err(Error::BadRequest { message: "No fields to update".to_string(), @@ -997,6 +1003,7 @@ spec: policies: None, users: None, buckets: None, + tiers: None, security_context: None, } } diff --git a/src/console/models/tenant.rs b/src/console/models/tenant.rs index 81d6e68..f9c2389 100755 --- a/src/console/models/tenant.rs +++ b/src/console/models/tenant.rs @@ -13,7 +13,7 @@ // limitations under the License. use crate::types::v1alpha1::{ - provisioning::{ProvisioningBucket, ProvisioningPolicy, ProvisioningUser}, + provisioning::{ProvisioningBucket, ProvisioningPolicy, ProvisioningTier, ProvisioningUser}, status::{ ConditionStatus, ConditionType, CurrentState, Reason, Status, canonical_filter_state, canonical_state, certificate, next_actions_for_reason, primary_condition, provisioning, @@ -158,6 +158,7 @@ pub struct CreateTenantRequest { pub policies: Option>, pub users: Option>, pub buckets: Option>, + pub tiers: Option>, /// Optional legacy Pod security context form fields. pub security_context: Option, } @@ -212,6 +213,9 @@ pub struct UpdateTenantRequest { /// Replace bucket provisioning declarations. pub buckets: Option>, + + /// Replace remote tier provisioning declarations. + pub tiers: Option>, } /// Key/value environment variable diff --git a/src/console/openapi.rs b/src/console/openapi.rs index 23581f6..6574f68 100644 --- a/src/console/openapi.rs +++ b/src/console/openapi.rs @@ -55,13 +55,14 @@ use crate::console::models::topology::{ TopologyOverviewResponse, TopologyPod, TopologyPool, TopologyTenant, TopologyTenantSummary, }; use crate::types::v1alpha1::provisioning::{ - BucketAnonymousAccess, BucketLifecycleAbortIncompleteMultipartUpload, - BucketLifecycleExpiration, BucketLifecycleNoncurrentVersionTransition, BucketLifecycleRule, - BucketLifecycleRuleFilter, BucketLifecycleRuleStatus, BucketLifecycleSpec, - BucketLifecycleState, BucketLifecycleTransition, BucketObjectLockConfiguration, - BucketObjectLockConfigurationState, BucketObjectLockRetentionMode, ConfigMapKeyReference, - PolicyDocumentSource, ProvisioningBucket, ProvisioningDeletionPolicy, ProvisioningPolicy, - ProvisioningUser, UserCredentialsSecretRef, + BucketAnonymousAccess, BucketEncryptionMode, BucketEncryptionSpec, BucketEncryptionState, + BucketLifecycleAbortIncompleteMultipartUpload, BucketLifecycleExpiration, + BucketLifecycleNoncurrentVersionTransition, BucketLifecycleRule, BucketLifecycleRuleFilter, + BucketLifecycleRuleStatus, BucketLifecycleSpec, BucketLifecycleState, + BucketLifecycleTransition, BucketObjectLockConfiguration, BucketObjectLockConfigurationState, + BucketObjectLockRetentionMode, ConfigMapKeyReference, PolicyDocumentSource, ProvisioningBucket, + ProvisioningDeletionPolicy, ProvisioningPolicy, ProvisioningTier, ProvisioningTierType, + ProvisioningUser, TierCredentialsSecretRef, UserCredentialsSecretRef, }; use crate::types::v1alpha1::status::provisioning::{ BucketVersioningStatus, ProvisioningBucketStatus, ProvisioningItemState, @@ -130,6 +131,9 @@ use crate::types::v1alpha1::status::provisioning::{ ProvisioningUser, UserCredentialsSecretRef, ProvisioningBucket, + ProvisioningTier, + ProvisioningTierType, + TierCredentialsSecretRef, ProvisioningDeletionPolicy, PolicyDocumentSource, ConfigMapKeyReference, @@ -137,6 +141,9 @@ use crate::types::v1alpha1::status::provisioning::{ BucketObjectLockConfiguration, BucketObjectLockConfigurationState, BucketObjectLockRetentionMode, + BucketEncryptionSpec, + BucketEncryptionState, + BucketEncryptionMode, BucketLifecycleSpec, BucketLifecycleState, BucketLifecycleRule, @@ -798,7 +805,21 @@ mod tests { assert!(schemas.contains_key("BucketLifecycleTransition")); assert!(schemas.contains_key("BucketLifecycleNoncurrentVersionTransition")); assert!(schemas.contains_key("BucketObjectLockConfiguration")); + assert!(schemas.contains_key("BucketEncryptionSpec")); assert!(schemas.contains_key("UserCredentialsSecretRef")); + assert!(schemas.contains_key("ProvisioningTier")); + assert!(schemas.contains_key("ProvisioningTierType")); + assert!(schemas.contains_key("TierCredentialsSecretRef")); + assert_eq!( + spec.pointer("/components/schemas/UpdateTenantRequest/properties/tiers/items/$ref") + .and_then(Value::as_str), + Some("#/components/schemas/ProvisioningTier") + ); + assert_eq!( + spec.pointer("/components/schemas/ProvisioningStatus/properties/tiers/items/$ref") + .and_then(Value::as_str), + Some("#/components/schemas/ProvisioningItemStatus") + ); assert_eq!( spec.pointer("/components/schemas/TenantDetailsResponse/properties/provisioning/$ref") .and_then(Value::as_str), @@ -835,6 +856,13 @@ mod tests { .and_then(Value::as_str), Some("#/components/schemas/BucketObjectLockConfiguration") ); + assert_eq!( + spec.pointer( + "/components/schemas/ProvisioningBucket/properties/encryption/oneOf/1/$ref", + ) + .and_then(Value::as_str), + Some("#/components/schemas/BucketEncryptionSpec") + ); assert_eq!( spec.pointer("/components/schemas/ProvisioningStatus/properties/buckets/items/$ref") .and_then(Value::as_str), diff --git a/src/reconcile/provisioning.rs b/src/reconcile/provisioning.rs index 6380c48..068cd85 100644 --- a/src/reconcile/provisioning.rs +++ b/src/reconcile/provisioning.rs @@ -14,6 +14,8 @@ use crate::context::{self, Context}; use crate::sts::rustfs_client::{ + BucketEncryptionAlgorithm as AdminEncryptionAlgorithm, + BucketEncryptionConfiguration as AdminEncryptionConfiguration, BucketLifecycleAbortIncompleteMultipartUpload as AdminLifecycleAbort, BucketLifecycleConfiguration as AdminLifecycleConfiguration, BucketLifecycleExpiration as AdminLifecycleExpiration, @@ -25,15 +27,17 @@ use crate::sts::rustfs_client::{ BucketObjectLockDefaultRetention as AdminObjectLockDefaultRetention, BucketObjectLockRetentionMode as AdminObjectLockRetentionMode, BucketObjectLockRetentionPeriod as AdminObjectLockRetentionPeriod, BucketVersioningState, - BucketVersioningUpdate, CreateBucketResult, RustfsAdminClient, RustfsClientError, - canonicalize_bucket_lifecycle_xml, + BucketVersioningUpdate, CreateBucketResult, RustfsAdminClient, RustfsClientError, RustfsTier, + RustfsTierConfig, canonicalize_bucket_lifecycle_xml, }; use crate::types::v1alpha1::provisioning::{ - BucketAnonymousAccess, BucketLifecycleRuleStatus, BucketLifecycleSpec, BucketLifecycleState, + BucketAnonymousAccess, BucketEncryptionMode, BucketEncryptionSpec, BucketEncryptionState, + BucketLifecycleRuleStatus, BucketLifecycleSpec, BucketLifecycleState, BucketObjectLockConfigurationState, BucketObjectLockRetentionMode, - MAX_BUCKET_LIFECYCLE_RULE_ID_LENGTH, MAX_BUCKET_LIFECYCLE_RULES, - MAX_BUCKET_OBJECT_LOCK_RETENTION_DAYS, PolicyDocumentSource, ProvisioningBucket, - ProvisioningPolicy, ProvisioningUser, duplicate_user_credentials_secret_names, + MAX_BUCKET_ENCRYPTION_KMS_KEY_ID_LENGTH, MAX_BUCKET_LIFECYCLE_RULE_ID_LENGTH, + MAX_BUCKET_LIFECYCLE_RULES, MAX_BUCKET_OBJECT_LOCK_RETENTION_DAYS, PolicyDocumentSource, + ProvisioningBucket, ProvisioningPolicy, ProvisioningTier, ProvisioningUser, + duplicate_user_credentials_secret_names, }; use crate::types::v1alpha1::status::Reason; use crate::types::v1alpha1::status::provisioning::{ @@ -111,6 +115,14 @@ struct UserCredentials { resource_version: Option, } +#[derive(Clone)] +struct TierCredentials { + access_key: String, + secret_key: String, + secret_name: String, + resource_version: Option, +} + enum UserCredentialsCheck { DuplicateSecret, Checked { @@ -194,6 +206,10 @@ impl ProvisioningRun<'_> { self.previous_bucket_status(name).map(AsRef::as_ref) } + fn previous_tier(&self, name: &str) -> Option<&ProvisioningItemStatus> { + self.previous.tiers.iter().find(|item| item.name == name) + } + fn push_policy(&mut self, item: ProvisioningItemStatus) { self.log_item_transition("policy", self.previous_policy(&item.name), &item); if item.state == ProvisioningItemState::Failed.as_str() { @@ -225,6 +241,15 @@ impl ProvisioningRun<'_> { self.status.buckets.push(item); } + fn push_tier(&mut self, item: ProvisioningItemStatus) { + self.log_item_transition("tier", self.previous_tier(&item.name), &item); + if item.state == ProvisioningItemState::Failed.as_str() { + self.failures + .push((reason_from_str(&item.reason), item_message(&item))); + } + self.status.tiers.push(item); + } + fn log_item_transition( &self, item_type: &'static str, @@ -370,6 +395,7 @@ impl ProvisioningRun<'_> { .buckets .iter() .any(|item| is_pending(&item.state)) + || self.status.tiers.iter().any(|item| is_pending(&item.state)) } fn first_pending_message(&self) -> String { @@ -378,6 +404,13 @@ impl ProvisioningRun<'_> { .iter() .find(|item| item.state == ProvisioningItemState::Pending.as_str()) .map(item_message) + .or_else(|| { + self.status + .tiers + .iter() + .find(|item| item.state == ProvisioningItemState::Pending.as_str()) + .map(item_message) + }) .or_else(|| { self.status.buckets.iter().find_map(|item| { (item.state == ProvisioningItemState::Pending.as_str()) @@ -482,6 +515,19 @@ impl ProvisioningRun<'_> { } self.push_bucket(item); } + for tier in &self.tenant.spec.tiers { + let mut item = self.item( + self.previous_tier(&tier.name), + &tier.name, + state.clone(), + reason, + message, + ); + if let Some(previous) = self.previous_tier(&tier.name) { + copy_tier_ownership(&mut item, previous); + } + self.push_tier(item); + } } fn fail_all_active(&mut self, reason: Reason, message: &str) { @@ -509,6 +555,13 @@ impl ProvisioningRun<'_> { self.status.buckets.push(self.retained_bucket(previous)); } } + + let tiers = desired_names(self.tenant.spec.tiers.iter().map(|tier| &tier.name)); + for previous in &self.previous.tiers { + if !tiers.contains(&previous.name) { + self.status.tiers.push(self.retained_item(previous)); + } + } } fn prepare_status(&mut self, phase: ProvisioningPhase) { @@ -516,6 +569,7 @@ impl ProvisioningRun<'_> { self.status.policies.sort_by(|a, b| a.name.cmp(&b.name)); self.status.users.sort_by(|a, b| a.name.cmp(&b.name)); self.status.buckets.sort_by(|a, b| a.name.cmp(&b.name)); + self.status.tiers.sort_by(|a, b| a.name.cmp(&b.name)); if !self.status.is_empty() { self.status.observed_generation = self.tenant.metadata.generation; self.status.phase = Some(phase); @@ -652,6 +706,7 @@ pub(super) async fn reconcile_provisioning( }, }; } + reconcile_tiers(&mut run, &client).await; reconcile_buckets(&mut run, &client).await; run.finish() } @@ -2030,6 +2085,360 @@ fn validate_user_secret_key(secret_key: &str) -> Result<(), String> { Ok(()) } +async fn reconcile_tiers(run: &mut ProvisioningRun<'_>, client: &RustfsAdminClient) { + let tiers = run.tenant.spec.tiers.clone(); + if tiers.is_empty() { + return; + } + + let live = match client.list_tiers().await { + Ok(live) => live, + Err(error) => { + let error = SpecLoadError { + reason: Reason::TierApplyFailed, + message: format!("failed to list RustFS tiers: {error}"), + transient: error.is_transient(), + }; + for tier in &tiers { + let previous = run.previous_tier(&tier.name).cloned(); + let mut item = + run.item_from_spec_error(previous.as_ref(), &tier.name, error.clone()); + if let Some(previous) = previous.as_ref() { + copy_tier_ownership(&mut item, previous); + } + run.push_tier(item); + } + return; + } + }; + + for tier in &tiers { + let credentials = load_tier_secret(run, tier).await; + let live_tier = live.iter().find(|live| live.config.name == tier.name); + let item = reconcile_tier(run, client, tier, live_tier, credentials).await; + run.push_tier(item); + } +} + +async fn reconcile_tier( + run: &mut ProvisioningRun<'_>, + client: &RustfsAdminClient, + tier: &ProvisioningTier, + live: Option<&RustfsTier>, + credentials: Result, +) -> ProvisioningItemStatus { + let previous = run.previous_tier(&tier.name).cloned(); + let generation = run.tenant.metadata.generation; + if let Err(message) = validate_tier_name(&tier.name) { + let item = run.item( + previous.as_ref(), + &tier.name, + ProvisioningItemState::Failed, + Reason::TierApplyFailed, + message, + ); + return finalize_tier_item(item, previous.as_ref(), None, None, generation); + } + + let desired = tier_config(tier); + let desired_hash = tier_config_hash(&desired); + let credentials = match credentials { + Ok(credentials) => credentials, + Err(error) => { + let item = run.item_from_spec_error(previous.as_ref(), &tier.name, error); + return finalize_tier_item( + item, + previous.as_ref(), + Some(desired_hash), + None, + generation, + ); + } + }; + + let ready = |run: &ProvisioningRun<'_>, message: &str| { + run.item( + previous.as_ref(), + &tier.name, + ProvisioningItemState::Ready, + Reason::ProvisioningConfigured, + message, + ) + }; + let (item, applied) = match live { + None => match client + .add_tier( + &desired, + &credentials.access_key, + &credentials.secret_key, + tier.force, + ) + .await + { + Ok(()) => (ready(run, "RustFS tier was registered"), true), + Err(error) if error.is_conflict() => ( + run.item( + previous.as_ref(), + &tier.name, + ProvisioningItemState::Failed, + Reason::TierConflict, + format!( + "RustFS refused to register the tier because remote bucket '{}' prefix '{}' is in use; set force: true only if this prefix belongs to this Tenant: {error}", + desired.bucket, desired.prefix + ), + ), + false, + ), + Err(error) => ( + run.item_from_admin_error( + previous.as_ref(), + &tier.name, + Reason::TierApplyFailed, + error, + format!("failed to register RustFS tier '{}'", tier.name), + ), + false, + ), + }, + Some(live) => { + let differences = tier_config_differences(&live.config, &desired); + if !differences.is_empty() { + ( + run.item( + previous.as_ref(), + &tier.name, + ProvisioningItemState::Failed, + Reason::TierConflict, + format!( + "live RustFS tier differs from spec in {}; RustFS cannot change these fields of a registered tier", + differences.join(", ") + ), + ), + false, + ) + } else if tier_credentials_need_apply(previous.as_ref(), &credentials, &live.access_key) + { + match client + .edit_tier_credentials( + &tier.name, + &credentials.access_key, + &credentials.secret_key, + ) + .await + { + Ok(()) => { + let message = if previous + .as_ref() + .and_then(|item| item.last_applied_hash.as_ref()) + .is_none() + { + "Existing RustFS tier matches spec and was adopted with the Secret credentials" + } else { + "RustFS tier credentials were updated from the Secret" + }; + (ready(run, message), true) + } + Err(error) => ( + run.item_from_admin_error( + previous.as_ref(), + &tier.name, + Reason::TierApplyFailed, + error, + format!("failed to update credentials of RustFS tier '{}'", tier.name), + ), + false, + ), + } + } else { + (ready(run, "RustFS tier matches spec"), false) + } + } + }; + + finalize_tier_item( + item, + previous.as_ref(), + Some(desired_hash), + applied.then_some(&credentials), + generation, + ) +} + +fn tier_config(tier: &ProvisioningTier) -> RustfsTierConfig { + RustfsTierConfig { + tier_type: tier.tier_type.rustfs_type().to_string(), + name: tier.name.clone(), + endpoint: tier.endpoint.clone(), + bucket: tier.bucket.clone(), + prefix: tier.prefix.clone(), + region: tier.region.clone(), + } +} + +fn tier_config_hash(config: &RustfsTierConfig) -> String { + hash_document( + &serde_json::json!([ + config.tier_type, + config.endpoint, + config.bucket, + config.prefix, + config.region, + ]) + .to_string(), + ) +} + +fn tier_config_differences( + live: &RustfsTierConfig, + desired: &RustfsTierConfig, +) -> Vec<&'static str> { + [ + ("type", &live.tier_type, &desired.tier_type), + ("endpoint", &live.endpoint, &desired.endpoint), + ("bucket", &live.bucket, &desired.bucket), + ("prefix", &live.prefix, &desired.prefix), + ("region", &live.region, &desired.region), + ] + .into_iter() + .filter_map(|(field, live, desired)| (live != desired).then_some(field)) + .collect() +} + +/// Credentials cannot be read back from RustFS, so the Secret version applied last is the +/// ownership record; the listed access key still reveals out-of-band changes. +fn tier_credentials_need_apply( + previous: Option<&ProvisioningItemStatus>, + credentials: &TierCredentials, + live_access_key: &str, +) -> bool { + let live_access_key_listed = !live_access_key.is_empty() && live_access_key != "REDACTED"; + if live_access_key_listed && live_access_key != credentials.access_key { + return true; + } + let (Some(previous), Some(resource_version)) = + (previous, credentials.resource_version.as_deref()) + else { + return true; + }; + previous.observed_secret_resource_version.as_deref() != Some(resource_version) + || previous.observed_secret_name.as_deref() != Some(credentials.secret_name.as_str()) + || previous.last_applied_access_key_hash.as_deref() + != Some(access_key_hash(&credentials.access_key).as_str()) +} + +fn finalize_tier_item( + mut item: ProvisioningItemStatus, + previous: Option<&ProvisioningItemStatus>, + desired_hash: Option, + applied_credentials: Option<&TierCredentials>, + generation: Option, +) -> ProvisioningItemStatus { + if let Some(previous) = previous { + copy_tier_ownership(&mut item, previous); + } + if let Some(credentials) = applied_credentials { + item.observed_secret_resource_version = credentials.resource_version.clone(); + item.observed_secret_name = Some(credentials.secret_name.clone()); + item.last_applied_access_key_hash = Some(access_key_hash(&credentials.access_key)); + } + if item.state == ProvisioningItemState::Ready.as_str() { + let changed = applied_credentials.is_some() + || item.last_applied_hash != desired_hash + || item.last_applied_generation.is_none(); + item.last_applied_hash = desired_hash.clone(); + if changed { + item.last_applied_generation = generation; + } + } + item.desired_hash = desired_hash; + item +} + +fn copy_tier_ownership(item: &mut ProvisioningItemStatus, previous: &ProvisioningItemStatus) { + item.desired_hash = previous.desired_hash.clone(); + item.last_applied_hash = previous.last_applied_hash.clone(); + item.last_applied_generation = previous.last_applied_generation; + item.observed_secret_resource_version = previous.observed_secret_resource_version.clone(); + item.observed_secret_name = previous.observed_secret_name.clone(); + item.last_applied_access_key_hash = previous.last_applied_access_key_hash.clone(); +} + +/// Mirrors the RustFS tier name checks; the name is also sent as an S3 XML storage class. +fn validate_tier_name(name: &str) -> Result<(), String> { + if name.is_empty() { + return Err("tier name must not be empty".to_string()); + } + if !is_valid_xml_text(name) { + return Err(format!( + "tier name '{name}' contains characters that cannot be used as an S3 storage class" + )); + } + if name != name.to_uppercase() { + return Err(format!("tier name '{name}' must be uppercase")); + } + if name.eq_ignore_ascii_case("STANDARD") || name.eq_ignore_ascii_case("REDUCED_REDUNDANCY") { + return Err(format!( + "tier name '{name}' is a reserved storage class and cannot name a tier" + )); + } + Ok(()) +} + +async fn load_tier_secret( + run: &ProvisioningRun<'_>, + tier: &ProvisioningTier, +) -> Result { + let secret_name = tier.creds_secret.name.as_str(); + let invalid = |message: String| SpecLoadError { + reason: Reason::TierSecretInvalid, + message, + transient: false, + }; + let secret: Secret = run + .ctx + .get(secret_name, run.namespace) + .await + .map_err(|error| { + if context::is_kube_not_found(&error) { + invalid(format!("tier Secret '{secret_name}' was not found")) + } else { + SpecLoadError { + reason: Reason::TierSecretInvalid, + message: format!("failed to read tier Secret '{secret_name}': {error}"), + transient: context::is_transient_kube_error(&error), + } + } + })?; + let data = secret + .data + .as_ref() + .ok_or_else(|| invalid(format!("tier Secret '{secret_name}' has no data")))?; + + Ok(TierCredentials { + access_key: read_tier_secret_value(data, "accesskey", secret_name).map_err(invalid)?, + secret_key: read_tier_secret_value(data, "secretkey", secret_name).map_err(invalid)?, + secret_name: secret_name.to_string(), + resource_version: secret.metadata.resource_version, + }) +} + +fn read_tier_secret_value( + data: &BTreeMap, + key: &'static str, + secret_name: &str, +) -> Result { + let raw = data + .get(key) + .ok_or_else(|| format!("tier Secret '{secret_name}' is missing '{key}'"))?; + let value = String::from_utf8(raw.0.clone()) + .map_err(|_| format!("tier Secret '{secret_name}' key '{key}' must be valid UTF-8"))?; + let value = value.trim(); + if value.is_empty() { + return Err(format!("tier Secret '{secret_name}' key '{key}' is empty")); + } + Ok(value.to_string()) +} + async fn reconcile_buckets(run: &mut ProvisioningRun<'_>, client: &RustfsAdminClient) { let buckets = run.tenant.spec.buckets.clone(); for bucket in &buckets { @@ -2050,6 +2459,10 @@ async fn reconcile_bucket( .object_lock_configuration .as_ref() .and_then(object_lock_configuration_desired_hash); + status.encryption_desired_hash = bucket + .encryption + .as_ref() + .and_then(bucket_encryption_desired_hash); if status.state != ProvisioningItemState::Ready.as_str() { status.lifecycle_desired_hash = bucket.lifecycle.as_ref().and_then(lifecycle_desired_hash); return status; @@ -2075,6 +2488,12 @@ async fn reconcile_bucket( return status; } + status = reconcile_bucket_encryption(run, client, bucket, previous.as_ref(), status).await; + if status.state != ProvisioningItemState::Ready.as_str() { + status.lifecycle_desired_hash = bucket.lifecycle.as_ref().and_then(lifecycle_desired_hash); + return status; + } + let mut completed = reconcile_bucket_lifecycle(run, client, bucket, previous.as_ref(), status.item.clone()) .await; @@ -2586,6 +3005,11 @@ fn validate_bucket_configuration(bucket: &ProvisioningBucket) -> Result<(), (Rea { return Err((Reason::BucketLifecycleApplyFailed, message)); } + if let Some(spec) = bucket.encryption.as_ref() + && let Err(message) = bucket_encryption_configuration(spec) + { + return Err((Reason::BucketEncryptionApplyFailed, message)); + } Ok(()) } @@ -2682,67 +3106,323 @@ fn copy_bucket_control_status( source.object_lock_configuration_desired_hash.clone(); target.object_lock_configuration_last_applied_hash = source.object_lock_configuration_last_applied_hash.clone(); + target.encryption_desired_hash = source.encryption_desired_hash.clone(); + target.encryption_last_applied_hash = source.encryption_last_applied_hash.clone(); } -async fn reconcile_bucket_lifecycle( +async fn reconcile_bucket_encryption( run: &mut ProvisioningRun<'_>, client: &RustfsAdminClient, bucket: &ProvisioningBucket, previous: Option<&ProvisioningBucketStatus>, - base_item: ProvisioningItemStatus, + mut status: ProvisioningBucketStatus, ) -> ProvisioningBucketStatus { - let Some(spec) = bucket.lifecycle.as_ref() else { - let mut status = bucket_status_with_preserved_lifecycle(base_item, previous); - status.lifecycle_desired_hash = None; + let Some(spec) = bucket.encryption.as_ref() else { return status; }; - - let desired = match lifecycle_configuration(spec) { - Ok(configuration) => configuration, + let desired = match bucket_encryption_configuration(spec) { + Ok(desired) => desired, Err(message) => { - let item = replace_bucket_item_state( + status.item = replace_bucket_item_state( run, previous, - &base_item, + &status.item, ProvisioningItemState::Failed, - Reason::BucketLifecycleApplyFailed, + Reason::BucketEncryptionApplyFailed, message, ); - let mut status = bucket_status_with_preserved_lifecycle(item, previous); - status.lifecycle_desired_hash = None; return status; } }; - let desired_document = desired.as_ref().map(AdminLifecycleConfiguration::to_xml); - let desired_hash = match desired_document.as_deref() { - Some(document) => match canonicalize_bucket_lifecycle_xml(document) { - Ok(document) => Some(hash_document(&document)), - Err(error) => { - let item = replace_bucket_item_state( - run, - previous, - &base_item, - ProvisioningItemState::Failed, - Reason::BucketLifecycleApplyFailed, - format!("failed to encode RustFS bucket lifecycle configuration: {error}"), - ); - return bucket_status_with_preserved_lifecycle(item, previous); - } - }, - None => None, - }; - - let live_document = match client - .get_bucket_lifecycle_configuration(&bucket.name) - .await + if desired + .as_ref() + .is_some_and(|desired| desired.algorithm == AdminEncryptionAlgorithm::AwsKms) + && !run + .tenant + .spec + .encryption + .as_ref() + .is_some_and(|encryption| encryption.enabled) { - Ok(document) => document, + status.item = replace_bucket_item_state( + run, + previous, + &status.item, + ProvisioningItemState::Failed, + Reason::BucketEncryptionApplyFailed, + "SSE-KMS bucket encryption requires spec.encryption.enabled on the Tenant", + ); + return status; + } + + let live = match client.get_bucket_encryption(&bucket.name).await { + Ok(live) => live, Err(error) => { - let item = bucket_item_from_admin_error( + status.item = bucket_item_from_admin_error( run, previous, - &base_item, - Reason::BucketLifecycleApplyFailed, + &status.item, + Reason::BucketEncryptionApplyFailed, + error, + "failed to read RustFS bucket encryption", + ); + return status; + } + }; + let live_hash = live + .as_ref() + .map(|configuration| hash_document(&configuration.to_xml())); + let last_applied_hash = previous.and_then(|item| item.encryption_last_applied_hash.as_deref()); + + let Some(desired) = desired else { + status.encryption_desired_hash = None; + let Some(live_hash) = live_hash else { + status.encryption_last_applied_hash = None; + status.item.message = Some(append_bucket_message( + &status.item, + "RustFS bucket encryption is absent", + )); + return status; + }; + if last_applied_hash != Some(live_hash.as_str()) { + status.item = replace_bucket_item_state( + run, + previous, + &status.item, + ProvisioningItemState::Failed, + Reason::BucketEncryptionConflict, + bucket_encryption_conflict_message(last_applied_hash, "delete"), + ); + return status; + } + if let Err(error) = client.delete_bucket_encryption(&bucket.name).await { + status.item = bucket_item_from_admin_error( + run, + previous, + &status.item, + Reason::BucketEncryptionApplyFailed, + error, + "failed to remove RustFS bucket encryption", + ); + return status; + } + status.encryption_last_applied_hash = None; + status.item.message = Some(append_bucket_message( + &status.item, + "RustFS bucket encryption was removed", + )); + return status; + }; + + if let Some(live) = live.as_ref() + && bucket_encryption_matches(&desired, live) + { + status.encryption_last_applied_hash = live_hash; + status.item.message = Some(append_bucket_message( + &status.item, + "RustFS bucket encryption matches spec", + )); + return status; + } + if let Some(live_hash) = live_hash.as_deref() + && last_applied_hash != Some(live_hash) + { + status.item = replace_bucket_item_state( + run, + previous, + &status.item, + ProvisioningItemState::Failed, + Reason::BucketEncryptionConflict, + bucket_encryption_conflict_message(last_applied_hash, "replace"), + ); + return status; + } + + if let Err(error) = client.put_bucket_encryption(&bucket.name, &desired).await { + status.item = bucket_item_from_admin_error( + run, + previous, + &status.item, + Reason::BucketEncryptionApplyFailed, + error, + "failed to apply RustFS bucket encryption", + ); + return status; + } + let verified = match client.get_bucket_encryption(&bucket.name).await { + Ok(Some(verified)) if bucket_encryption_matches(&desired, &verified) => verified, + Ok(_) => { + status.item = replace_bucket_item_state( + run, + previous, + &status.item, + ProvisioningItemState::Failed, + Reason::BucketEncryptionApplyFailed, + "RustFS bucket encryption did not match spec after update", + ); + return status; + } + Err(error) => { + status.item = bucket_item_from_admin_error( + run, + previous, + &status.item, + Reason::BucketEncryptionApplyFailed, + error, + "failed to verify RustFS bucket encryption after update", + ); + return status; + } + }; + // Ownership is the verified live document: RustFS fills in the default KMS key for SSE-KMS. + status.encryption_last_applied_hash = Some(hash_document(&verified.to_xml())); + status.item.message = Some(append_bucket_message( + &status.item, + "RustFS bucket encryption was applied", + )); + status +} + +fn bucket_encryption_conflict_message(last_applied_hash: Option<&str>, action: &str) -> String { + if last_applied_hash.is_some() { + format!( + "Live RustFS bucket encryption changed since the operator last applied it; refusing to {action} it" + ) + } else { + format!( + "Live RustFS bucket encryption is not owned by this status; refusing to {action} it" + ) + } +} + +/// `Ok(None)` is a valid `Absent` spec. +fn bucket_encryption_configuration( + spec: &BucketEncryptionSpec, +) -> Result, String> { + match spec.state() { + BucketEncryptionState::Absent => { + if spec.mode.is_some() || spec.kms_key_id.is_some() { + return Err( + "Absent bucket encryption must not contain mode or kmsKeyId".to_string() + ); + } + Ok(None) + } + BucketEncryptionState::Present => { + let algorithm = match spec.mode { + Some(BucketEncryptionMode::SseS3) => AdminEncryptionAlgorithm::Aes256, + Some(BucketEncryptionMode::SseKms) => AdminEncryptionAlgorithm::AwsKms, + None => return Err("Present bucket encryption must contain a mode".to_string()), + }; + if let Some(key_id) = spec.kms_key_id.as_deref() { + if algorithm != AdminEncryptionAlgorithm::AwsKms { + return Err("bucket encryption kmsKeyId requires mode SSE-KMS".to_string()); + } + if !is_valid_kms_key_id(key_id) { + return Err(format!( + "bucket encryption kmsKeyId must contain between 1 and {MAX_BUCKET_ENCRYPTION_KMS_KEY_ID_LENGTH} characters from [A-Za-z0-9._:/+=@-]" + )); + } + } + Ok(Some(AdminEncryptionConfiguration { + algorithm, + kms_master_key_id: spec.kms_key_id.clone(), + bucket_key_enabled: false, + })) + } + } +} + +fn is_valid_kms_key_id(key_id: &str) -> bool { + !key_id.is_empty() + && key_id.len() <= MAX_BUCKET_ENCRYPTION_KMS_KEY_ID_LENGTH as usize + && key_id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || b"._:/+=@-".contains(&byte)) +} + +fn bucket_encryption_desired_hash(spec: &BucketEncryptionSpec) -> Option { + bucket_encryption_configuration(spec) + .ok() + .flatten() + .map(|configuration| hash_document(&configuration.to_xml())) +} + +/// SSE-KMS without `kmsKeyId` accepts whatever default key RustFS recorded. +fn bucket_encryption_matches( + desired: &AdminEncryptionConfiguration, + live: &AdminEncryptionConfiguration, +) -> bool { + desired.algorithm == live.algorithm + && desired.bucket_key_enabled == live.bucket_key_enabled + && match desired.kms_master_key_id.as_deref() { + Some(key_id) => live.kms_master_key_id.as_deref() == Some(key_id), + None => { + desired.algorithm == AdminEncryptionAlgorithm::AwsKms + || live.kms_master_key_id.is_none() + } + } +} + +async fn reconcile_bucket_lifecycle( + run: &mut ProvisioningRun<'_>, + client: &RustfsAdminClient, + bucket: &ProvisioningBucket, + previous: Option<&ProvisioningBucketStatus>, + base_item: ProvisioningItemStatus, +) -> ProvisioningBucketStatus { + let Some(spec) = bucket.lifecycle.as_ref() else { + let mut status = bucket_status_with_preserved_lifecycle(base_item, previous); + status.lifecycle_desired_hash = None; + return status; + }; + + let desired = match lifecycle_configuration(spec) { + Ok(configuration) => configuration, + Err(message) => { + let item = replace_bucket_item_state( + run, + previous, + &base_item, + ProvisioningItemState::Failed, + Reason::BucketLifecycleApplyFailed, + message, + ); + let mut status = bucket_status_with_preserved_lifecycle(item, previous); + status.lifecycle_desired_hash = None; + return status; + } + }; + let desired_document = desired.as_ref().map(AdminLifecycleConfiguration::to_xml); + let desired_hash = match desired_document.as_deref() { + Some(document) => match canonicalize_bucket_lifecycle_xml(document) { + Ok(document) => Some(hash_document(&document)), + Err(error) => { + let item = replace_bucket_item_state( + run, + previous, + &base_item, + ProvisioningItemState::Failed, + Reason::BucketLifecycleApplyFailed, + format!("failed to encode RustFS bucket lifecycle configuration: {error}"), + ); + return bucket_status_with_preserved_lifecycle(item, previous); + } + }, + None => None, + }; + + let live_document = match client + .get_bucket_lifecycle_configuration(&bucket.name) + .await + { + Ok(document) => document, + Err(error) => { + let item = bucket_item_from_admin_error( + run, + previous, + &base_item, + Reason::BucketLifecycleApplyFailed, error, "failed to read RustFS bucket lifecycle configuration", ); @@ -3553,6 +4233,7 @@ fn has_active_spec(tenant: &Tenant) -> bool { !tenant.spec.policies.is_empty() || !tenant.spec.users.is_empty() || !tenant.spec.buckets.is_empty() + || !tenant.spec.tiers.is_empty() } fn desired_names<'a>(names: impl Iterator) -> BTreeSet { @@ -3743,8 +4424,13 @@ fn reason_from_str(reason: &str) -> Reason { "BucketObjectLockConfigurationConflict" => Reason::BucketObjectLockConfigurationConflict, "BucketPolicyApplyFailed" => Reason::BucketPolicyApplyFailed, "BucketPolicyConflict" => Reason::BucketPolicyConflict, + "BucketEncryptionApplyFailed" => Reason::BucketEncryptionApplyFailed, + "BucketEncryptionConflict" => Reason::BucketEncryptionConflict, "BucketLifecycleApplyFailed" => Reason::BucketLifecycleApplyFailed, "BucketLifecycleConflict" => Reason::BucketLifecycleConflict, + "TierSecretInvalid" => Reason::TierSecretInvalid, + "TierApplyFailed" => Reason::TierApplyFailed, + "TierConflict" => Reason::TierConflict, _ => Reason::ProvisioningFailed, } } @@ -3802,8 +4488,10 @@ mod tests { existing: bool, ignore_versioning_put: bool, ignore_object_lock_put: bool, + ignore_encryption_put: bool, versioning: Arc>, object_lock: Arc>>, + encryption: Arc>>, requests: Arc>>, put_requests: Arc>>, create_object_lock_header: Arc>>, @@ -3823,8 +4511,10 @@ mod tests { existing: false, ignore_versioning_put: false, ignore_object_lock_put: false, + ignore_encryption_put: false, versioning: Arc::new(Mutex::new(BucketVersioningState::Unversioned)), object_lock: Arc::new(Mutex::new(None)), + encryption: Arc::new(Mutex::new(None)), requests: Arc::new(Mutex::new(Vec::new())), put_requests: Arc::new(Mutex::new(Vec::new())), create_object_lock_header: Arc::new(Mutex::new(None)), @@ -3981,6 +4671,49 @@ mod tests { } StatusCode::OK.into_response() } + (axum::http::Method::GET, "encryption=") => { + match capture.encryption.lock().await.clone() { + // RustFS omits the namespace on GetBucketEncryption. + Some(configuration) => ( + StatusCode::OK, + configuration + .to_xml() + .replace(" xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"", ""), + ) + .into_response(), + // RustFS answers a missing configuration with 400, not 404. + None => ( + StatusCode::BAD_REQUEST, + "ServerSideEncryptionConfigurationNotFoundErrorThe server-side encryption configuration was not found.", + ) + .into_response(), + } + } + (axum::http::Method::PUT, "encryption=") => { + let body = capture_bucket_control_put(&capture, &query, request).await; + let kms = body.contains("aws:kms"); + let kms_master_key_id = body + .split_once("") + .and_then(|(_, tail)| tail.split_once("")) + .map(|(key, _)| key.to_string()) + .or_else(|| kms.then(|| "rustfs-default-key".to_string())); + if !capture.ignore_encryption_put { + *capture.encryption.lock().await = Some(AdminEncryptionConfiguration { + algorithm: if kms { + AdminEncryptionAlgorithm::AwsKms + } else { + AdminEncryptionAlgorithm::Aes256 + }, + kms_master_key_id, + bucket_key_enabled: false, + }); + } + StatusCode::OK.into_response() + } + (axum::http::Method::DELETE, "encryption=") => { + *capture.encryption.lock().await = None; + StatusCode::NO_CONTENT.into_response() + } _ => StatusCode::METHOD_NOT_ALLOWED.into_response(), } } @@ -6235,51 +6968,453 @@ mod tests { server.abort(); } + fn sse_s3() -> AdminEncryptionConfiguration { + AdminEncryptionConfiguration { + algorithm: AdminEncryptionAlgorithm::Aes256, + kms_master_key_id: None, + bucket_key_enabled: false, + } + } + + fn encryption_bucket(encryption: serde_json::Value) -> ProvisioningBucket { + serde_json::from_value(serde_json::json!({ + "name": "secure-bucket", + "encryption": encryption + })) + .expect("encryption bucket should deserialize") + } + + fn previous_encryption_owner(hash: String) -> ProvisioningBucketStatus { + let mut previous = ProvisioningBucketStatus::new(ProvisioningItemStatus::new( + "secure-bucket", + ProvisioningItemState::Ready, + Reason::ProvisioningConfigured.as_str(), + )); + previous.encryption_last_applied_hash = Some(hash); + previous + } + + async fn requests_for(capture: &BucketControlCapture, request: &str) -> usize { + capture + .requests + .lock() + .await + .iter() + .filter(|logged| logged.as_str() == request) + .count() + } + #[tokio::test] - async fn matching_live_object_lock_configuration_is_adopted_without_a_write() { - let live = AdminObjectLockConfiguration { - default_retention: Some(AdminObjectLockDefaultRetention { - mode: AdminObjectLockRetentionMode::Compliance, - period: AdminObjectLockRetentionPeriod::Days(30), - }), - }; - let capture = BucketControlCapture { - existing: true, - versioning: Arc::new(Mutex::new(BucketVersioningState::Enabled)), - object_lock: Arc::new(Mutex::new(Some(live))), - ..Default::default() - }; + async fn sse_s3_bucket_encryption_is_applied_to_a_new_bucket_and_owned() { + let capture = BucketControlCapture::default(); let (client, server) = bucket_control_test_client(capture.clone()).await; let ctx = empty_kube_context(); let user = provisioning_user("app-user", "app-user-secret", "readwrite"); let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); let mut run = empty_run(&ctx, &tenant); - let item = reconcile_bucket(&mut run, &client, &object_lock_bucket()).await; + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-S3"})), + ) + .await; assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); - assert_eq!( - item.object_lock_configuration_desired_hash, - item.object_lock_configuration_last_applied_hash - ); assert!( - !capture - .requests - .lock() - .await - .iter() - .any(|request| request == "PUT object-lock=") + item.message + .as_deref() + .is_some_and(|message| message.contains("RustFS bucket encryption was applied")) ); + assert_eq!(capture.encryption.lock().await.clone(), Some(sse_s3())); + let owned = Some(hash_document(&sse_s3().to_xml())); + assert_eq!(item.encryption_desired_hash, owned); + assert_eq!(item.encryption_last_applied_hash, owned); + let put_requests = capture.put_requests.lock().await; + let put = put_requests + .iter() + .find(|request| request.query == "encryption=") + .expect("encryption PUT is captured"); + assert_eq!(put.body, sse_s3().to_xml()); + assert!(put.content_md5.is_some()); + assert!(put.authorization.as_deref().is_some_and(|authorization| { + authorization.contains( + "SignedHeaders=content-md5;content-type;host;x-amz-content-sha256;x-amz-date", + ) + })); server.abort(); } #[tokio::test] - async fn omitted_object_lock_configuration_leaves_live_retention_unmanaged() { - let live = AdminObjectLockConfiguration { - default_retention: Some(AdminObjectLockDefaultRetention { - mode: AdminObjectLockRetentionMode::Governance, - period: AdminObjectLockRetentionPeriod::Days(7), - }), + async fn matching_live_bucket_encryption_is_adopted_without_a_write() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(sse_s3()))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-S3"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!( + item.encryption_last_applied_hash, + Some(hash_document(&sse_s3().to_xml())) + ); + assert_eq!(requests_for(&capture, "PUT encryption=").await, 0); + server.abort(); + } + + #[tokio::test] + async fn omitted_bucket_encryption_makes_no_encryption_calls() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(sse_s3()))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + let bucket: ProvisioningBucket = + serde_json::from_value(serde_json::json!({"name": "secure-bucket"})) + .expect("bucket should deserialize"); + + let item = reconcile_bucket(&mut run, &client, &bucket).await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert!( + !capture + .requests + .lock() + .await + .iter() + .any(|request| request.ends_with("encryption=")) + ); + assert_eq!(capture.encryption.lock().await.clone(), Some(sse_s3())); + server.abort(); + } + + #[tokio::test] + async fn unowned_live_bucket_encryption_is_not_replaced_or_removed() { + let foreign = AdminEncryptionConfiguration { + algorithm: AdminEncryptionAlgorithm::AwsKms, + kms_master_key_id: Some("foreign-key".to_string()), + bucket_key_enabled: false, + }; + for spec in [ + serde_json::json!({"mode": "SSE-S3"}), + serde_json::json!({"state": "Absent"}), + ] { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(foreign.clone()))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket(&mut run, &client, &encryption_bucket(spec.clone())).await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str(), "{spec}"); + assert_eq!(item.reason, Reason::BucketEncryptionConflict.as_str()); + assert!( + item.message + .as_deref() + .is_some_and(|message| message.contains("not owned by this status")) + ); + assert_eq!(requests_for(&capture, "PUT encryption=").await, 0); + assert_eq!(requests_for(&capture, "DELETE encryption=").await, 0); + assert_eq!( + capture.encryption.lock().await.clone(), + Some(foreign.clone()) + ); + server.abort(); + } + } + + #[tokio::test] + async fn owned_bucket_encryption_follows_a_spec_change_to_sse_kms() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(sse_s3()))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let mut tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + tenant.spec.encryption = Some(crate::types::v1alpha1::encryption::EncryptionConfig { + enabled: true, + ..Default::default() + }); + let mut run = empty_run(&ctx, &tenant); + run.previous + .buckets + .push(previous_encryption_owner(hash_document(&sse_s3().to_xml()))); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-KMS"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + let applied = AdminEncryptionConfiguration { + algorithm: AdminEncryptionAlgorithm::AwsKms, + kms_master_key_id: Some("rustfs-default-key".to_string()), + bucket_key_enabled: false, + }; + assert_eq!( + capture.encryption.lock().await.clone(), + Some(applied.clone()) + ); + assert_eq!( + item.encryption_last_applied_hash, + Some(hash_document(&applied.to_xml())) + ); + assert_ne!( + item.encryption_desired_hash, + item.encryption_last_applied_hash + ); + drop(run); + + // The RustFS-chosen default key is adopted on the next pass without another write. + let mut run = empty_run(&ctx, &tenant); + run.previous.buckets.push(item.clone()); + let again = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-KMS"})), + ) + .await; + assert_eq!(again.state, ProvisioningItemState::Ready.as_str()); + assert_eq!(requests_for(&capture, "PUT encryption=").await, 1); + server.abort(); + } + + #[tokio::test] + async fn absent_bucket_encryption_removes_only_an_owned_configuration() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(sse_s3()))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + run.previous + .buckets + .push(previous_encryption_owner(hash_document(&sse_s3().to_xml()))); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"state": "Absent"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!(capture.encryption.lock().await.clone(), None); + assert_eq!(item.encryption_desired_hash, None); + assert_eq!(item.encryption_last_applied_hash, None); + assert_eq!(requests_for(&capture, "DELETE encryption=").await, 1); + server.abort(); + } + + #[tokio::test] + async fn changed_owned_bucket_encryption_is_a_conflict() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(AdminEncryptionConfiguration { + bucket_key_enabled: true, + ..sse_s3() + }))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + run.previous + .buckets + .push(previous_encryption_owner(hash_document(&sse_s3().to_xml()))); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-S3"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::BucketEncryptionConflict.as_str()); + assert!( + item.message.as_deref().is_some_and( + |message| message.contains("changed since the operator last applied it") + ) + ); + assert_eq!( + item.encryption_last_applied_hash, + Some(hash_document(&sse_s3().to_xml())) + ); + assert_eq!(requests_for(&capture, "PUT encryption=").await, 0); + server.abort(); + } + + #[tokio::test] + async fn sse_kms_bucket_encryption_requires_tenant_kms() { + let capture = BucketControlCapture { + existing: true, + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-KMS", "kmsKeyId": "app-key"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::BucketEncryptionApplyFailed.as_str()); + assert!( + !capture + .requests + .lock() + .await + .iter() + .any(|request| request.ends_with("encryption=")) + ); + server.abort(); + } + + #[tokio::test] + async fn bucket_encryption_update_is_verified_after_a_success_response() { + let capture = BucketControlCapture { + existing: true, + ignore_encryption_put: true, + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-S3"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::BucketEncryptionApplyFailed.as_str()); + assert_eq!(item.encryption_last_applied_hash, None); + server.abort(); + } + + #[test] + fn bucket_encryption_spec_is_validated_before_any_call() { + let spec = |value: serde_json::Value| -> BucketEncryptionSpec { + serde_json::from_value(value).expect("encryption spec deserializes") + }; + assert_eq!( + bucket_encryption_configuration(&spec(serde_json::json!({"mode": "SSE-S3"}))), + Ok(Some(sse_s3())) + ); + assert_eq!( + bucket_encryption_configuration(&spec(serde_json::json!({"state": "Absent"}))), + Ok(None) + ); + for invalid in [ + serde_json::json!({}), + serde_json::json!({"state": "Absent", "mode": "SSE-S3"}), + serde_json::json!({"mode": "SSE-S3", "kmsKeyId": "app-key"}), + serde_json::json!({"mode": "SSE-KMS", "kmsKeyId": "app key"}), + serde_json::json!({"mode": "SSE-KMS", "kmsKeyId": "app"}), + serde_json::json!({"mode": "SSE-KMS", "kmsKeyId": ""}), + ] { + assert!( + bucket_encryption_configuration(&spec(invalid.clone())).is_err(), + "{invalid} must be rejected" + ); + } + let bucket = encryption_bucket(serde_json::json!({"state": "Absent", "mode": "SSE-S3"})); + assert!(matches!( + validate_bucket_configuration(&bucket), + Err((Reason::BucketEncryptionApplyFailed, _)) + )); + } + + #[tokio::test] + async fn matching_live_object_lock_configuration_is_adopted_without_a_write() { + let live = AdminObjectLockConfiguration { + default_retention: Some(AdminObjectLockDefaultRetention { + mode: AdminObjectLockRetentionMode::Compliance, + period: AdminObjectLockRetentionPeriod::Days(30), + }), + }; + let capture = BucketControlCapture { + existing: true, + versioning: Arc::new(Mutex::new(BucketVersioningState::Enabled)), + object_lock: Arc::new(Mutex::new(Some(live))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket(&mut run, &client, &object_lock_bucket()).await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!( + item.object_lock_configuration_desired_hash, + item.object_lock_configuration_last_applied_hash + ); + assert!( + !capture + .requests + .lock() + .await + .iter() + .any(|request| request == "PUT object-lock=") + ); + server.abort(); + } + + #[tokio::test] + async fn omitted_object_lock_configuration_leaves_live_retention_unmanaged() { + let live = AdminObjectLockConfiguration { + default_retention: Some(AdminObjectLockDefaultRetention { + mode: AdminObjectLockRetentionMode::Governance, + period: AdminObjectLockRetentionPeriod::Days(7), + }), }; let capture = BucketControlCapture { existing: true, @@ -7306,4 +8441,636 @@ mod tests { _ => panic!("pending referenced policy must requeue instead of failing the user"), } } + + #[derive(Clone, Default)] + struct TierCapture { + tiers: Arc>, + add_status: Arc, + list_status: Arc, + adds: Arc>>, + edits: Arc>>, + } + + async fn tier_admin_handler( + State(capture): State, + request: Request, + ) -> Response { + let method = request.method().clone(); + let path = request.uri().path().to_string(); + let query = request.uri().query().unwrap_or("").to_string(); + let body = axum::body::to_bytes(request.into_body(), usize::MAX) + .await + .expect("tier request body"); + let json = || serde_json::from_slice::(&body).expect("tier body is JSON"); + match method { + axum::http::Method::GET => { + let status = capture.list_status.load(Ordering::SeqCst); + if status != 0 { + return StatusCode::from_u16(status as u16) + .expect("configured test status is valid") + .into_response(); + } + (StatusCode::OK, capture.tiers.lock().await.clone()).into_response() + } + axum::http::Method::PUT => { + let added = json(); + capture.adds.lock().await.push((query, added.clone())); + let status = capture.add_status.load(Ordering::SeqCst); + if status != 0 { + return StatusCode::from_u16(status as u16) + .expect("configured test status is valid") + .into_response(); + } + let tier_type = added["type"].as_str().expect("tier type").to_string(); + let mut backend = added[&tier_type].clone(); + backend["secretKey"] = Value::from("REDACTED"); + let mut tiers: Vec = + serde_json::from_str(&capture.tiers.lock().await).unwrap_or_default(); + tiers.push(serde_json::json!({"type": tier_type, tier_type.clone(): backend})); + *capture.tiers.lock().await = Value::from(tiers).to_string(); + StatusCode::OK.into_response() + } + axum::http::Method::POST => { + capture.edits.lock().await.push((path, json())); + StatusCode::OK.into_response() + } + _ => StatusCode::METHOD_NOT_ALLOWED.into_response(), + } + } + + async fn tier_test_client( + capture: TierCapture, + lifecycle: Option, + ) -> (RustfsAdminClient, tokio::task::JoinHandle<()>) { + let mut router = Router::new() + .route("/rustfs/admin/v3/tier", any(tier_admin_handler)) + .route("/rustfs/admin/v3/tier/:name", any(tier_admin_handler)) + .with_state(capture); + if let Some(lifecycle) = lifecycle { + router = router.merge( + Router::new() + .route("/app-data", any(lifecycle_bucket_handler)) + .with_state(lifecycle), + ); + } + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("test server should bind"); + let addr = listener.local_addr().expect("listener should have address"); + let server = tokio::spawn(async move { + axum::serve(listener, router) + .await + .expect("test server should serve") + }); + ( + RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"), + server, + ) + } + + fn cold_tier() -> ProvisioningTier { + serde_json::from_value(serde_json::json!({ + "name": "COLD", + "type": "S3", + "endpoint": "https://gateway.example.com", + "bucket": "cold", + "prefix": "tenant-a/", + "credsSecret": {"name": "cold-tier"} + })) + .expect("tier should deserialize") + } + + fn live_cold_tier(prefix: &str, access_key: &str) -> String { + serde_json::json!([{ + "type": "s3", + "s3": { + "name": "COLD", + "endpoint": "https://gateway.example.com", + "accessKey": access_key, + "secretKey": "REDACTED", + "bucket": "cold", + "prefix": prefix, + "region": "", + "storageClass": "" + } + }]) + .to_string() + } + + fn tier_credentials(resource_version: &str) -> TierCredentials { + TierCredentials { + access_key: "tier-access".to_string(), + secret_key: "tier-secret".to_string(), + secret_name: "cold-tier".to_string(), + resource_version: Some(resource_version.to_string()), + } + } + + fn owned_tier_status(resource_version: &str) -> ProvisioningItemStatus { + let mut item = ProvisioningItemStatus::new( + "COLD", + ProvisioningItemState::Ready, + Reason::ProvisioningConfigured.as_str(), + ); + let hash = tier_config_hash(&tier_config(&cold_tier())); + item.desired_hash = Some(hash.clone()); + item.last_applied_hash = Some(hash); + item.last_applied_generation = Some(2); + item.observed_secret_resource_version = Some(resource_version.to_string()); + item.observed_secret_name = Some("cold-tier".to_string()); + item.last_applied_access_key_hash = Some(access_key_hash("tier-access")); + item + } + + fn tier_test_tenant(previous: Option) -> Tenant { + Tenant { + metadata: ObjectMeta { + name: Some("tenant-a".to_string()), + namespace: Some("storage".to_string()), + generation: Some(3), + ..Default::default() + }, + spec: crate::types::v1alpha1::tenant::TenantSpec { + tiers: vec![cold_tier()], + ..Default::default() + }, + status: Some(crate::types::v1alpha1::status::Status { + provisioning: ProvisioningStatus { + tiers: previous.into_iter().collect(), + ..Default::default() + }, + ..Default::default() + }), + } + } + + fn tier_run<'a>(ctx: &'a Context, tenant: &'a Tenant) -> ProvisioningRun<'a> { + let mut run = empty_run(ctx, tenant); + run.previous = tenant + .status + .as_ref() + .map(|status| status.provisioning.clone()) + .unwrap_or_default(); + run + } + + fn tier_secret_kube_context(resource_version: &'static str) -> Context { + let kube_service = service_fn(move |request: http::Request| async move { + assert_eq!(request.method(), http::Method::GET); + assert_eq!( + request.uri().path(), + "/api/v1/namespaces/storage/secrets/cold-tier" + ); + let secret = Secret { + metadata: ObjectMeta { + name: Some("cold-tier".to_string()), + namespace: Some("storage".to_string()), + resource_version: Some(resource_version.to_string()), + ..Default::default() + }, + data: Some(BTreeMap::from([ + ("accesskey".to_string(), ByteString(b"tier-access".to_vec())), + ( + "secretkey".to_string(), + ByteString(b"tier-secret\n".to_vec()), + ), + ])), + ..Default::default() + }; + Ok::<_, Infallible>( + http::Response::builder() + .header("content-type", "application/json") + .body(KubeBody::from( + serde_json::to_vec(&secret).expect("Secret should serialize"), + )) + .expect("response should build"), + ) + }); + Context::new(Client::new(kube_service, "default")) + } + + #[tokio::test] + async fn missing_tier_is_registered_with_secret_credentials() { + let capture = TierCapture::default(); + let (client, server) = tier_test_client(capture.clone(), None).await; + let ctx = empty_kube_context(); + let tenant = tier_test_tenant(None); + let mut run = tier_run(&ctx, &tenant); + + let item = reconcile_tier( + &mut run, + &client, + &cold_tier(), + None, + Ok(tier_credentials("5")), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + let adds = capture.adds.lock().await.clone(); + assert_eq!(adds.len(), 1); + assert_eq!(adds[0].0, "force=false"); + assert_eq!( + adds[0].1, + serde_json::json!({ + "type": "s3", + "name": "COLD", + "s3": { + "name": "COLD", + "endpoint": "https://gateway.example.com", + "accessKey": "tier-access", + "secretKey": "tier-secret", + "bucket": "cold", + "prefix": "tenant-a/", + "region": "" + } + }) + ); + assert!(capture.edits.lock().await.is_empty()); + assert_eq!(item.observed_secret_resource_version.as_deref(), Some("5")); + assert_eq!(item.observed_secret_name.as_deref(), Some("cold-tier")); + assert_eq!(item.last_applied_hash, item.desired_hash); + assert_eq!(item.last_applied_generation, Some(3)); + server.abort(); + } + + #[tokio::test] + async fn owned_tier_with_unchanged_secret_is_not_rewritten() { + let capture = TierCapture::default(); + let (client, server) = tier_test_client(capture.clone(), None).await; + let ctx = empty_kube_context(); + let tenant = tier_test_tenant(Some(owned_tier_status("5"))); + let mut run = tier_run(&ctx, &tenant); + let live = parse_live_tier(&live_cold_tier("tenant-a/", "tier-access")); + + let item = reconcile_tier( + &mut run, + &client, + &cold_tier(), + Some(&live), + Ok(tier_credentials("5")), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert!(capture.adds.lock().await.is_empty()); + assert!(capture.edits.lock().await.is_empty()); + assert_eq!(item.last_applied_generation, Some(2)); + server.abort(); + } + + #[tokio::test] + async fn rotated_tier_secret_only_updates_credentials() { + let capture = TierCapture::default(); + let (client, server) = tier_test_client(capture.clone(), None).await; + let ctx = empty_kube_context(); + let tenant = tier_test_tenant(Some(owned_tier_status("5"))); + let mut run = tier_run(&ctx, &tenant); + let live = parse_live_tier(&live_cold_tier("tenant-a/", "tier-access")); + + let item = reconcile_tier( + &mut run, + &client, + &cold_tier(), + Some(&live), + Ok(tier_credentials("6")), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert!(capture.adds.lock().await.is_empty()); + assert_eq!( + capture.edits.lock().await.clone(), + vec![( + "/rustfs/admin/v3/tier/COLD".to_string(), + serde_json::json!({"access": "tier-access", "secret": "tier-secret"}) + )] + ); + assert_eq!(item.observed_secret_resource_version.as_deref(), Some("6")); + assert_eq!(item.last_applied_generation, Some(3)); + server.abort(); + } + + #[tokio::test] + async fn existing_tier_is_adopted_once_and_out_of_band_access_key_is_corrected() { + let capture = TierCapture::default(); + let (client, server) = tier_test_client(capture.clone(), None).await; + let ctx = empty_kube_context(); + let tenant = tier_test_tenant(None); + let mut run = tier_run(&ctx, &tenant); + let live = parse_live_tier(&live_cold_tier("tenant-a/", "tier-access")); + + let adopted = reconcile_tier( + &mut run, + &client, + &cold_tier(), + Some(&live), + Ok(tier_credentials("5")), + ) + .await; + + assert_eq!(adopted.state, ProvisioningItemState::Ready.as_str()); + assert!( + adopted + .message + .as_deref() + .is_some_and(|message| message.contains("adopted")) + ); + assert_eq!(capture.edits.lock().await.len(), 1); + + let tenant = tier_test_tenant(Some(adopted)); + let mut run = tier_run(&ctx, &tenant); + let changed = parse_live_tier(&live_cold_tier("tenant-a/", "someone-else")); + let item = reconcile_tier( + &mut run, + &client, + &cold_tier(), + Some(&changed), + Ok(tier_credentials("5")), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!(capture.edits.lock().await.len(), 2); + assert!(capture.adds.lock().await.is_empty()); + server.abort(); + } + + #[tokio::test] + async fn live_tier_with_different_settings_fails_without_writes() { + let capture = TierCapture::default(); + let (client, server) = tier_test_client(capture.clone(), None).await; + let ctx = empty_kube_context(); + let tenant = tier_test_tenant(Some(owned_tier_status("5"))); + let mut run = tier_run(&ctx, &tenant); + let live = parse_live_tier(&live_cold_tier("other/", "tier-access")); + + let item = reconcile_tier( + &mut run, + &client, + &cold_tier(), + Some(&live), + Ok(tier_credentials("6")), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::TierConflict.as_str()); + assert!( + item.message + .as_deref() + .is_some_and(|message| message.contains("prefix")) + ); + assert!(capture.adds.lock().await.is_empty()); + assert!(capture.edits.lock().await.is_empty()); + assert_eq!(item.observed_secret_resource_version.as_deref(), Some("5")); + assert!(run.retry.is_none()); + server.abort(); + } + + #[tokio::test] + async fn non_empty_remote_prefix_is_a_conflict_unless_forced() { + let capture = TierCapture::default(); + capture.add_status.store(409, Ordering::SeqCst); + let (client, server) = tier_test_client(capture.clone(), None).await; + let ctx = empty_kube_context(); + let tenant = tier_test_tenant(None); + let mut run = tier_run(&ctx, &tenant); + + let item = reconcile_tier( + &mut run, + &client, + &cold_tier(), + None, + Ok(tier_credentials("5")), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::TierConflict.as_str()); + assert!( + item.message + .as_deref() + .is_some_and(|message| message.contains("force: true")) + ); + assert!(item.observed_secret_resource_version.is_none()); + + capture.add_status.store(0, Ordering::SeqCst); + let mut forced = cold_tier(); + forced.force = true; + let item = + reconcile_tier(&mut run, &client, &forced, None, Ok(tier_credentials("5"))).await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!(capture.adds.lock().await[1].0, "force=true"); + server.abort(); + } + + #[tokio::test] + async fn tier_name_and_secret_errors_fail_before_rustfs_writes() { + let capture = TierCapture::default(); + let (client, server) = tier_test_client(capture.clone(), None).await; + let ctx = empty_kube_context(); + let tenant = tier_test_tenant(None); + let mut run = tier_run(&ctx, &tenant); + + let mut lowercase = cold_tier(); + lowercase.name = "cold".to_string(); + let item = reconcile_tier( + &mut run, + &client, + &lowercase, + None, + Ok(tier_credentials("5")), + ) + .await; + assert_eq!(item.reason, Reason::TierApplyFailed.as_str()); + assert!( + item.message + .as_deref() + .is_some_and(|message| message.contains("uppercase")) + ); + + let item = reconcile_tier( + &mut run, + &client, + &cold_tier(), + None, + Err(SpecLoadError { + reason: Reason::TierSecretInvalid, + message: "tier Secret 'cold-tier' was not found".to_string(), + transient: false, + }), + ) + .await; + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::TierSecretInvalid.as_str()); + assert!(capture.adds.lock().await.is_empty()); + assert!(capture.edits.lock().await.is_empty()); + server.abort(); + } + + #[test] + fn redacted_live_access_key_falls_back_to_the_secret_version() { + let previous = owned_tier_status("5"); + for live_access_key in ["", "REDACTED"] { + assert!(!tier_credentials_need_apply( + Some(&previous), + &tier_credentials("5"), + live_access_key + )); + assert!(tier_credentials_need_apply( + Some(&previous), + &tier_credentials("6"), + live_access_key + )); + } + assert!(tier_credentials_need_apply( + Some(&previous), + &tier_credentials("5"), + "someone-else" + )); + } + + #[test] + fn tier_names_follow_rustfs_rules() { + assert!(validate_tier_name("COLD_1").is_ok()); + assert!(validate_tier_name("ÄRCHIV").is_ok()); + for name in ["", "Cold", "STANDARD", "REDUCED_REDUNDANCY", "BAD\u{1}"] { + assert!( + validate_tier_name(name).is_err(), + "{name:?} must be rejected" + ); + } + } + + #[test] + fn tier_secret_values_are_trimmed_and_required() { + let data = BTreeMap::from([ + ("accesskey".to_string(), ByteString(b" key \n".to_vec())), + ("secretkey".to_string(), ByteString(b" ".to_vec())), + ("binary".to_string(), ByteString(vec![0xff])), + ]); + + assert_eq!( + read_tier_secret_value(&data, "accesskey", "cold-tier").as_deref(), + Ok("key") + ); + assert!( + read_tier_secret_value(&data, "secretkey", "cold-tier") + .is_err_and(|message| message.contains("is empty")) + ); + assert!( + read_tier_secret_value(&data, "binary", "cold-tier") + .is_err_and(|message| message.contains("UTF-8")) + ); + assert!( + read_tier_secret_value(&data, "missing", "cold-tier") + .is_err_and(|message| message.contains("missing 'missing'")) + ); + } + + #[tokio::test] + async fn transient_tier_list_failure_keeps_ownership_and_retries() { + let capture = TierCapture::default(); + capture.list_status.store(503, Ordering::SeqCst); + let (client, server) = tier_test_client(capture.clone(), None).await; + let ctx = empty_kube_context(); + let tenant = tier_test_tenant(Some(owned_tier_status("5"))); + let mut run = tier_run(&ctx, &tenant); + + reconcile_tiers(&mut run, &client).await; + + let item = &run.status.tiers[0]; + assert_eq!(item.state, ProvisioningItemState::Pending.as_str()); + assert_eq!(item.observed_secret_resource_version.as_deref(), Some("5")); + assert!(item.last_applied_hash.is_some()); + assert!(run.retry.is_some()); + server.abort(); + } + + #[tokio::test] + async fn spec_tier_is_registered_before_a_bucket_lifecycle_references_it() { + let capture = TierCapture::default(); + *capture.tiers.lock().await = "[]".to_string(); + let lifecycle = LifecycleCapture { + tiers: capture.tiers.clone(), + ..Default::default() + }; + let (client, server) = tier_test_client(capture.clone(), Some(lifecycle.clone())).await; + let ctx = tier_secret_kube_context("7"); + let tenant = tier_test_tenant(None); + let mut run = tier_run(&ctx, &tenant); + + reconcile_tiers(&mut run, &client).await; + let bucket = + reconcile_bucket(&mut run, &client, &transition_lifecycle_bucket(1, "COLD")).await; + + assert_eq!( + run.status.tiers[0].state, + ProvisioningItemState::Ready.as_str() + ); + assert_eq!( + run.status.tiers[0] + .observed_secret_resource_version + .as_deref(), + Some("7") + ); + assert_eq!( + capture.adds.lock().await[0].1["s3"]["secretKey"], + "tier-secret" + ); + assert_eq!(bucket.state, ProvisioningItemState::Ready.as_str()); + assert_eq!(lifecycle.put_count.load(Ordering::SeqCst), 1); + assert!(run.retry.is_none()); + server.abort(); + } + + #[tokio::test] + async fn removed_tier_is_retained_with_its_ownership() { + let ctx = empty_kube_context(); + let mut tenant = tier_test_tenant(Some(owned_tier_status("5"))); + tenant.spec.tiers.clear(); + tenant.spec.buckets = vec![ProvisioningBucket { + name: "app-data".to_string(), + ..Default::default() + }]; + let mut run = tier_run(&ctx, &tenant); + + run.prepare_status(ProvisioningPhase::Ready); + + assert_eq!(run.status.tiers.len(), 1); + assert_eq!( + run.status.tiers[0].state, + ProvisioningItemState::Retained.as_str() + ); + assert_eq!( + run.status.tiers[0] + .observed_secret_resource_version + .as_deref(), + Some("5") + ); + assert!(has_active_spec(&tier_test_tenant(None))); + } + + fn parse_live_tier(body: &str) -> RustfsTier { + let tier: Value = serde_json::from_str::>(body).expect("tier list")[0].clone(); + RustfsTier { + config: RustfsTierConfig { + tier_type: tier["type"].as_str().expect("type").to_string(), + name: tier["s3"]["name"].as_str().expect("name").to_string(), + endpoint: tier["s3"]["endpoint"] + .as_str() + .expect("endpoint") + .to_string(), + bucket: tier["s3"]["bucket"].as_str().expect("bucket").to_string(), + prefix: tier["s3"]["prefix"].as_str().expect("prefix").to_string(), + region: tier["s3"]["region"].as_str().expect("region").to_string(), + }, + access_key: tier["s3"]["accessKey"] + .as_str() + .expect("accessKey") + .to_string(), + } + } } diff --git a/src/sts/rustfs_client.rs b/src/sts/rustfs_client.rs index b758226..a99b1e8 100644 --- a/src/sts/rustfs_client.rs +++ b/src/sts/rustfs_client.rs @@ -21,6 +21,7 @@ use crate::Tenant; use crate::cluster_dns; pub use rustfs_admin::{ + BucketEncryptionAlgorithm, BucketEncryptionConfiguration, BucketLifecycleAbortIncompleteMultipartUpload, BucketLifecycleConfiguration, BucketLifecycleExpiration, BucketLifecycleNoncurrentVersionTransition, BucketLifecycleRule, BucketLifecycleRuleStatus, BucketLifecycleTransition, BucketObjectLockConfiguration, @@ -28,7 +29,8 @@ pub use rustfs_admin::{ BucketObjectLockRetentionPeriod, BucketVersioningState, BucketVersioningUpdate, CreateBucketResult, RustfsClientError, RustfsCredentials, RustfsErasureBackend, RustfsErasureSetInfo, RustfsPoolDecommissionInfo, RustfsPoolListItem, RustfsPoolStatus, - RustfsServerInfo, RustfsServerUsage, RustfsUserInfo, canonicalize_bucket_lifecycle_xml, + RustfsServerInfo, RustfsServerUsage, RustfsTier, RustfsTierConfig, RustfsUserInfo, + canonicalize_bucket_lifecycle_xml, }; /// Tenant-aware wrapper around the kube-agnostic RustFS admin client. diff --git a/src/tenant_reference_index.rs b/src/tenant_reference_index.rs index 0c6dd31..b346f4a 100644 --- a/src/tenant_reference_index.rs +++ b/src/tenant_reference_index.rs @@ -348,6 +348,16 @@ mod tests { policies: vec!["readwrite".to_string()], ..Default::default() }); + tenant + .spec + .tiers + .push(crate::types::v1alpha1::provisioning::ProvisioningTier { + name: "COLD".to_string(), + creds_secret: crate::types::v1alpha1::provisioning::TierCredentialsSecretRef { + name: "cold-tier-credentials".to_string(), + }, + ..Default::default() + }); tenant.spec.encryption = Some(EncryptionConfig { kms_secret: Some(local_ref("vault-token")), local: Some(LocalKmsConfig { @@ -404,6 +414,7 @@ mod tests { "oidc-extra-ca", "provisioned-user-credentials", "legacy-user", + "cold-tier-credentials", "vault-token", "local-master-key", "external-ca", diff --git a/src/types/v1alpha1.rs b/src/types/v1alpha1.rs index ec2051d..369761d 100755 --- a/src/types/v1alpha1.rs +++ b/src/types/v1alpha1.rs @@ -146,12 +146,52 @@ mod tenant_provisioning_crd_tests { MAX_BUCKET_OBJECT_LOCK_RETENTION_DAYS, MAX_POLICIES_PER_USER, }, tenant::{ - MAX_TENANT_BUCKETS, MAX_TENANT_POLICIES, MAX_TENANT_POOLS, MAX_TENANT_USERS, Tenant, + MAX_TENANT_BUCKETS, MAX_TENANT_POLICIES, MAX_TENANT_POOLS, MAX_TENANT_TIERS, + MAX_TENANT_USERS, Tenant, }, }; use kube::CustomResourceExt; use serde_json::json; + #[test] + fn tenant_crd_declares_tiers_with_rustfs_name_rules() { + let crd = serde_json::to_value(Tenant::crd()).expect("Tenant CRD serializes"); + let schema = &crd["spec"]["versions"][0]["schema"]["openAPIV3Schema"]; + let tiers = &schema["properties"]["spec"]["properties"]["tiers"]; + + assert_eq!(tiers["x-kubernetes-list-type"], json!("map")); + assert_eq!(tiers["x-kubernetes-list-map-keys"], json!(["name"])); + assert_eq!(tiers["maxItems"], json!(MAX_TENANT_TIERS)); + let tier = &tiers["items"]; + assert_eq!( + tier["required"], + json!(["bucket", "credsSecret", "endpoint", "name", "type"]) + ); + assert_eq!( + tier["properties"]["type"]["enum"], + json!(["S3", "RustFS", "MinIO"]) + ); + assert_eq!( + tier["properties"]["deletionPolicy"]["enum"], + json!(["Retain"]) + ); + assert_eq!( + tier["properties"]["name"]["x-kubernetes-validations"], + json!([ + {"message": "tier name must be uppercase", "rule": "self == self.upperAscii()"}, + { + "message": "tier name must not be a reserved storage class", + "rule": "self != 'STANDARD' && self != 'REDUCED_REDUNDANCY'" + } + ]) + ); + assert_eq!( + schema["properties"]["status"]["properties"]["provisioning"]["properties"]["tiers"]["items"] + ["properties"]["observedSecretResourceVersion"]["type"], + json!("string") + ); + } + #[test] fn tenant_crd_includes_provisioning_spec_status_and_bounded_list_rules() { let crd = serde_json::to_value(Tenant::crd()).expect("Tenant CRD serializes"); @@ -465,6 +505,28 @@ mod tenant_provisioning_crd_tests { object_lock["x-kubernetes-validations"][0]["message"], json!("Present object lock configuration requires mode and days; Absent forbids them") ); + let encryption = &spec["properties"]["buckets"]["items"]["properties"]["encryption"]; + assert_eq!( + encryption["properties"]["mode"]["enum"], + json!(["SSE-S3", "SSE-KMS", null]) + ); + assert_eq!( + encryption["properties"]["kmsKeyId"]["pattern"], + json!("^[A-Za-z0-9._:/+=@-]+$") + ); + let encryption_messages: Vec<_> = encryption["x-kubernetes-validations"] + .as_array() + .expect("encryption validations are present") + .iter() + .map(|validation| validation["message"].clone()) + .collect(); + assert_eq!( + encryption_messages, + vec![ + json!("Present bucket encryption requires mode; Absent forbids mode and kmsKeyId"), + json!("kmsKeyId requires mode SSE-KMS"), + ] + ); let bucket_validations = spec["properties"]["buckets"]["items"]["x-kubernetes-validations"] .as_array() .expect("bucket validations are present"); @@ -492,6 +554,10 @@ mod tenant_provisioning_crd_tests { bucket_status["properties"]["objectLockConfigurationLastAppliedHash"]["nullable"], json!(true) ); + assert_eq!( + bucket_status["properties"]["encryptionLastAppliedHash"]["nullable"], + json!(true) + ); } } diff --git a/src/types/v1alpha1/provisioning.rs b/src/types/v1alpha1/provisioning.rs index 69b8260..c0df897 100644 --- a/src/types/v1alpha1/provisioning.rs +++ b/src/types/v1alpha1/provisioning.rs @@ -30,6 +30,9 @@ pub(crate) const MAX_BUCKET_NAME_LENGTH: u32 = 63; pub(crate) const MAX_BUCKET_LIFECYCLE_RULES: u32 = 1_000; pub(crate) const MAX_BUCKET_LIFECYCLE_RULE_ID_LENGTH: u32 = 255; pub(crate) const MAX_BUCKET_OBJECT_LOCK_RETENTION_DAYS: u32 = 36_500; +pub(crate) const MAX_BUCKET_ENCRYPTION_KMS_KEY_ID_LENGTH: u32 = 1024; +pub(crate) const MAX_PROVISIONING_TIER_NAME_LENGTH: u32 = 253; +pub(crate) const MAX_TIER_CREDENTIALS_SECRET_NAME_LENGTH: u32 = 253; #[derive(Deserialize, Serialize, Clone, Debug, JsonSchema, ToSchema, Default, PartialEq, Eq)] #[serde(rename_all = "PascalCase")] @@ -124,6 +127,80 @@ pub(crate) fn duplicate_user_credentials_secret_names( .collect() } +/// Remote backend of a tier. All three use the S3 API with static credentials. +#[derive( + Deserialize, Serialize, Clone, Copy, Debug, JsonSchema, ToSchema, Default, PartialEq, Eq, +)] +pub enum ProvisioningTierType { + #[default] + #[serde(rename = "S3")] + S3, + #[serde(rename = "RustFS")] + RustFS, + #[serde(rename = "MinIO")] + MinIO, +} + +impl ProvisioningTierType { + /// Backend type as named by the RustFS tier admin API. + pub const fn rustfs_type(self) -> &'static str { + match self { + Self::S3 => "s3", + Self::RustFS => "rustfs", + Self::MinIO => "minio", + } + } +} + +/// Reference to a tier credentials Secret in the Tenant namespace. +#[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, ToSchema, Default, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TierCredentialsSecretRef { + #[schemars(length(min = 1, max = MAX_TIER_CREDENTIALS_SECRET_NAME_LENGTH))] + pub name: String, +} + +/// Remote storage tier registered in RustFS for bucket lifecycle transitions. +#[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, ToSchema, Default, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct ProvisioningTier { + /// Tier name, referenced as `storageClass` by bucket lifecycle transitions. + #[schemars(length(min = 1, max = MAX_PROVISIONING_TIER_NAME_LENGTH))] + #[x_kube(validation = Rule::new("self == self.upperAscii()").message("tier name must be uppercase"))] + #[x_kube(validation = Rule::new("self != 'STANDARD' && self != 'REDUCED_REDUNDANCY'").message("tier name must not be a reserved storage class"))] + pub name: String, + + #[serde(rename = "type")] + pub tier_type: ProvisioningTierType, + + /// Remote S3 endpoint URL, for example `https://gateway.storjshare.io`. + #[schemars(length(min = 1))] + pub endpoint: String, + + /// Remote bucket that receives transitioned objects. + #[schemars(length(min = 1))] + pub bucket: String, + + /// Object prefix inside the remote bucket. + #[serde(default, skip_serializing_if = "String::is_empty")] + pub prefix: String, + + #[serde(default, skip_serializing_if = "String::is_empty")] + pub region: String, + + /// Secret with `accesskey` and `secretkey` for the remote bucket. Updating the Secret + /// rotates the tier credentials in RustFS. + pub creds_secret: TierCredentialsSecretRef, + + /// Register the tier even when the remote prefix already holds objects, for example a + /// prefix written by this Tenant before a restore. RustFS refuses such a prefix otherwise. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub force: bool, + + #[serde(default, skip_serializing_if = "is_retain")] + pub deletion_policy: ProvisioningDeletionPolicy, +} + #[derive( Deserialize, Serialize, Clone, Copy, Debug, JsonSchema, ToSchema, Default, PartialEq, Eq, )] @@ -187,6 +264,51 @@ impl BucketObjectLockConfiguration { } } +#[derive(Deserialize, Serialize, Clone, Copy, Debug, JsonSchema, ToSchema, PartialEq, Eq)] +#[serde(rename_all = "PascalCase")] +pub enum BucketEncryptionState { + Present, + Absent, +} + +#[derive(Deserialize, Serialize, Clone, Copy, Debug, JsonSchema, ToSchema, PartialEq, Eq)] +pub enum BucketEncryptionMode { + #[serde(rename = "SSE-S3")] + SseS3, + #[serde(rename = "SSE-KMS")] + SseKms, +} + +/// Default server-side encryption applied by RustFS to new objects in the bucket. +#[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, ToSchema, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +#[x_kube(validation = Rule::new("(!has(self.state) || self.state == 'Present') ? has(self.mode) : !has(self.mode) && !has(self.kmsKeyId)").message("Present bucket encryption requires mode; Absent forbids mode and kmsKeyId"))] +#[x_kube(validation = Rule::new("!has(self.kmsKeyId) || (has(self.mode) && self.mode == 'SSE-KMS')").message("kmsKeyId requires mode SSE-KMS"))] +pub struct BucketEncryptionSpec { + /// Omission is equivalent to `Present`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub state: Option, + + /// `SSE-S3` stores S3 `AES256`; `SSE-KMS` stores `aws:kms` and requires + /// `spec.encryption.enabled` on the Tenant. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub mode: Option, + + /// KMS key for `SSE-KMS`. Omission uses the RustFS default KMS key. + #[schemars( + length(min = 1, max = MAX_BUCKET_ENCRYPTION_KMS_KEY_ID_LENGTH), + regex(pattern = r"^[A-Za-z0-9._:/+=@-]+$") + )] + #[serde(default, skip_serializing_if = "Option::is_none")] + pub kms_key_id: Option, +} + +impl BucketEncryptionSpec { + pub(crate) fn state(&self) -> BucketEncryptionState { + self.state.unwrap_or(BucketEncryptionState::Present) + } +} + #[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, ToSchema, Default, PartialEq, Eq)] #[serde(rename_all = "camelCase")] pub struct BucketLifecycleRuleFilter { @@ -317,6 +439,11 @@ pub struct ProvisioningBucket { #[serde(default, skip_serializing_if = "Option::is_none")] pub policy: Option, + /// Default bucket encryption. Omission leaves the live configuration unmanaged; `Absent` + /// removes only a configuration previously owned by the operator. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub encryption: Option, + /// Declarative S3 bucket lifecycle configuration. Omission leaves the live lifecycle /// configuration unmanaged; `Absent` removes only a configuration previously owned by the /// operator. @@ -521,6 +648,35 @@ mod tests { assert!(rule.expiration.is_none()); } + #[test] + fn tier_deserializes_issue_yaml_shape_and_omits_defaults() { + let tier: super::ProvisioningTier = serde_json::from_value(serde_json::json!({ + "name": "STORJ", + "type": "S3", + "endpoint": "https://gateway.storjshare.io", + "bucket": "rustfs-ilm", + "prefix": "tenant-a/", + "credsSecret": {"name": "storj-tier"} + })) + .expect("tier should deserialize"); + + assert_eq!(tier.tier_type, super::ProvisioningTierType::S3); + assert_eq!(tier.tier_type.rustfs_type(), "s3"); + assert!(tier.region.is_empty()); + assert!(!tier.force); + let value = serde_json::to_value(&tier).expect("tier serializes"); + assert!(value.get("region").is_none()); + assert!(value.get("force").is_none()); + assert!(value.get("deletionPolicy").is_none()); + assert_eq!(value["type"], "S3"); + + for (name, rustfs_type) in [("RustFS", "rustfs"), ("MinIO", "minio")] { + let tier_type: super::ProvisioningTierType = + serde_json::from_value(serde_json::json!(name)).expect("tier type deserializes"); + assert_eq!(tier_type.rustfs_type(), rustfs_type); + } + } + #[test] fn bucket_object_lock_configuration_preserves_issue_yaml_shape() { let bucket: super::ProvisioningBucket = serde_json::from_value(serde_json::json!({ diff --git a/src/types/v1alpha1/status.rs b/src/types/v1alpha1/status.rs index 0ca12f6..8401376 100755 --- a/src/types/v1alpha1/status.rs +++ b/src/types/v1alpha1/status.rs @@ -189,8 +189,13 @@ pub enum Reason { BucketObjectLockConfigurationConflict, BucketPolicyApplyFailed, BucketPolicyConflict, + BucketEncryptionApplyFailed, + BucketEncryptionConflict, BucketLifecycleApplyFailed, BucketLifecycleConflict, + TierSecretInvalid, + TierApplyFailed, + TierConflict, KubernetesApiError, StatusPatchFailed, ObservedGenerationStale, @@ -273,8 +278,13 @@ impl Reason { Self::BucketObjectLockConfigurationConflict => "BucketObjectLockConfigurationConflict", Self::BucketPolicyApplyFailed => "BucketPolicyApplyFailed", Self::BucketPolicyConflict => "BucketPolicyConflict", + Self::BucketEncryptionApplyFailed => "BucketEncryptionApplyFailed", + Self::BucketEncryptionConflict => "BucketEncryptionConflict", Self::BucketLifecycleApplyFailed => "BucketLifecycleApplyFailed", Self::BucketLifecycleConflict => "BucketLifecycleConflict", + Self::TierSecretInvalid => "TierSecretInvalid", + Self::TierApplyFailed => "TierApplyFailed", + Self::TierConflict => "TierConflict", Self::KubernetesApiError => "KubernetesApiError", Self::StatusPatchFailed => "StatusPatchFailed", Self::ObservedGenerationStale => "ObservedGenerationStale", @@ -560,8 +570,13 @@ pub fn is_blocked_reason(reason: &str) -> bool { | "BucketObjectLockConfigurationConflict" | "BucketPolicyApplyFailed" | "BucketPolicyConflict" + | "BucketEncryptionApplyFailed" + | "BucketEncryptionConflict" | "BucketLifecycleApplyFailed" | "BucketLifecycleConflict" + | "TierSecretInvalid" + | "TierApplyFailed" + | "TierConflict" ) } @@ -679,10 +694,17 @@ pub fn next_actions_for_reason(reason: &str) -> Vec<&'static str> { } "BucketPolicyApplyFailed" => vec!["fixBucketPolicy", "inspectOperatorLogs"], "BucketPolicyConflict" => vec!["inspectLiveBucketPolicy", "updateBucketSpec"], + "BucketEncryptionApplyFailed" => vec!["fixBucketEncryption", "inspectOperatorLogs"], + "BucketEncryptionConflict" => { + vec!["inspectLiveBucketEncryption", "updateBucketSpec"] + } "BucketLifecycleApplyFailed" => vec!["fixBucketLifecycle", "inspectOperatorLogs"], "BucketLifecycleConflict" => { vec!["inspectLiveBucketLifecycle", "updateBucketSpec"] } + "TierSecretInvalid" => vec!["fixTierSecret"], + "TierApplyFailed" => vec!["fixTierSpec", "inspectOperatorLogs"], + "TierConflict" => vec!["inspectLiveTier", "updateTierSpec"], "KubernetesApiError" => vec!["retry", "inspectOperatorLogs"], "ObservedGenerationStale" => vec!["waitForReconcile"], _ => Vec::new(), @@ -800,6 +822,23 @@ mod tests { next_actions_for_reason("BucketLifecycleConflict"), vec!["inspectLiveBucketLifecycle", "updateBucketSpec"] ); + assert!(is_blocked_reason("BucketEncryptionApplyFailed")); + assert!(is_blocked_reason("BucketEncryptionConflict")); + assert_eq!( + next_actions_for_reason("BucketEncryptionApplyFailed"), + vec!["fixBucketEncryption", "inspectOperatorLogs"] + ); + assert_eq!( + next_actions_for_reason("BucketEncryptionConflict"), + vec!["inspectLiveBucketEncryption", "updateBucketSpec"] + ); + for reason in ["TierSecretInvalid", "TierApplyFailed", "TierConflict"] { + assert!(is_blocked_reason(reason), "{reason} must block"); + } + assert_eq!( + next_actions_for_reason("TierConflict"), + vec!["inspectLiveTier", "updateTierSpec"] + ); } #[test] diff --git a/src/types/v1alpha1/status/provisioning.rs b/src/types/v1alpha1/status/provisioning.rs index 9512af4..070c646 100644 --- a/src/types/v1alpha1/status/provisioning.rs +++ b/src/types/v1alpha1/status/provisioning.rs @@ -34,11 +34,17 @@ pub struct ProvisioningStatus { #[serde(default, skip_serializing_if = "Vec::is_empty")] pub buckets: Vec, + + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub tiers: Vec, } impl ProvisioningStatus { pub fn is_empty(&self) -> bool { - self.policies.is_empty() && self.users.is_empty() && self.buckets.is_empty() + self.policies.is_empty() + && self.users.is_empty() + && self.buckets.is_empty() + && self.tiers.is_empty() } } @@ -147,6 +153,12 @@ pub struct ProvisioningBucketStatus { #[serde(default, skip_serializing_if = "Option::is_none")] pub object_lock_configuration_last_applied_hash: Option, + + #[serde(default, skip_serializing_if = "Option::is_none")] + pub encryption_desired_hash: Option, + + #[serde(default, skip_serializing_if = "Option::is_none")] + pub encryption_last_applied_hash: Option, } impl ProvisioningBucketStatus { @@ -158,6 +170,8 @@ impl ProvisioningBucketStatus { versioning: None, object_lock_configuration_desired_hash: None, object_lock_configuration_last_applied_hash: None, + encryption_desired_hash: None, + encryption_last_applied_hash: None, } } } diff --git a/src/types/v1alpha1/tenant.rs b/src/types/v1alpha1/tenant.rs index 90cf298..5a387e7 100755 --- a/src/types/v1alpha1/tenant.rs +++ b/src/types/v1alpha1/tenant.rs @@ -19,7 +19,7 @@ use crate::types::v1alpha1::network::NetworkConfig; use crate::types::v1alpha1::pool::{Pool, validate_pool_collection}; use crate::types::v1alpha1::pool_lifecycle::PoolLifecycleSpec; use crate::types::v1alpha1::provisioning::{ - ProvisioningBucket, ProvisioningPolicy, ProvisioningUser, + ProvisioningBucket, ProvisioningPolicy, ProvisioningTier, ProvisioningUser, }; use crate::types::v1alpha1::security_context::PodSecurityContextOverride; use crate::types::v1alpha1::tls::TlsConfig; @@ -44,6 +44,7 @@ pub(crate) const MAX_TENANT_POOLS: u32 = 32; pub(crate) const MAX_TENANT_POLICIES: u32 = 256; pub(crate) const MAX_TENANT_USERS: u32 = 256; pub(crate) const MAX_TENANT_BUCKETS: u32 = 1024; +pub(crate) const MAX_TENANT_TIERS: u32 = 64; pub(crate) const RUSTFS_TENANT_LABEL: &str = "rustfs.tenant"; /// Reference to one key in a namespaced Kubernetes Secret. @@ -283,6 +284,15 @@ pub struct TenantSpec { #[serde(default, skip_serializing_if = "Vec::is_empty")] pub buckets: Vec, + /// Remote storage tiers that bucket lifecycle transitions can reference by name. + /// Tiers are reconciled before buckets. + #[schemars( + length(max = MAX_TENANT_TIERS), + extend("x-kubernetes-list-type" = "map", "x-kubernetes-list-map-keys" = ["name"]) + )] + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub tiers: Vec, + /// Encryption / KMS configuration for server-side encryption. /// When enabled, the operator injects KMS environment variables and mounts /// secrets into RustFS pods so the in-process `rustfs-kms` library is configured. @@ -344,6 +354,9 @@ impl TenantSpec { for user in &self.users { insert(user.credentials_secret_name()); } + for tier in &self.tiers { + insert(&tier.creds_secret.name); + } } if let Some(tls) = &self.tls { @@ -640,6 +653,13 @@ mod tests { "policies": ["readwrite"] } ], + "tiers": [{ + "name": "COLD", + "type": "S3", + "endpoint": "https://s3.example.com", + "bucket": "cold", + "credsSecret": { "name": "cold-tier-creds" } + }], "tls": { "certManager": { "secretName": "default-tls", @@ -676,6 +696,7 @@ mod tests { spec.referenced_secret_names(), BTreeSet::from([ "admin-creds".to_string(), + "cold-tier-creds".to_string(), "default-ca".to_string(), "default-tls".to_string(), "env-secret".to_string(),