Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@ 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.
- Helm chart `cosiDriver.enabled` to deploy the RustFS COSI (`rustfs.objectstorage.k8s.io`)
driver alongside the upstream provisioner sidecar, with dedicated ServiceAccount/RBAC.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down
17 changes: 17 additions & 0 deletions console-web/types/api.ts
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,20 @@ export interface ProvisioningBucket {
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"
Expand Down Expand Up @@ -198,6 +212,7 @@ export interface ProvisioningStatus {
policies?: ProvisioningItemStatus[]
users?: ProvisioningItemStatus[]
buckets?: ProvisioningBucketStatus[]
tiers?: ProvisioningItemStatus[]
}

export interface CreatePoolRequest {
Expand Down Expand Up @@ -225,6 +240,7 @@ export interface CreateTenantRequest {
policies?: ProvisioningPolicy[]
users?: ProvisioningUser[]
buckets?: ProvisioningBucket[]
tiers?: ProvisioningTier[]
security_context?: CreateSecurityContextRequest
}

Expand All @@ -238,6 +254,7 @@ export interface UpdateTenantRequest {
policies?: ProvisioningPolicy[]
users?: ProvisioningUser[]
buckets?: ProvisioningBucket[]
tiers?: ProvisioningTier[]
logging?: {
logType: string
volumeSize?: string
Expand Down
97 changes: 89 additions & 8 deletions crates/rustfs-admin/src/admin_ops.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<BTreeSet<String>, 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<Vec<RustfsTier>, 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<RustfsServerInfo, RustfsClientError> {
Expand Down
43 changes: 31 additions & 12 deletions crates/rustfs-admin/src/helpers.rs
Original file line number Diff line number Diff line change
Expand Up @@ -13,15 +13,14 @@
// 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;
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 {
Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -153,12 +152,13 @@ pub(super) fn extract_canned_policy_document(body: &str) -> Result<String, Rustf
}

/// Parses `GET /rustfs/admin/v3/tier`, an array of `{"type": "<t>", "<t>": {"name": ..}}`
/// entries where the backend object is keyed by the tier type.
pub(super) fn parse_tier_names(body: &str) -> Result<BTreeSet<String>, 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<Vec<RustfsTier>, RustfsClientError> {
let tiers = match serde_json::from_str::<Value>(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),
};
Expand All @@ -169,12 +169,31 @@ pub(super) fn parse_tier_names(body: &str) -> Result<BTreeSet<String>, 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()
}
Expand Down
77 changes: 75 additions & 2 deletions crates/rustfs-admin/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,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";
Expand Down Expand Up @@ -120,6 +120,24 @@ pub struct RustfsUserInfo {
pub policy_names: Vec<String>,
}

/// 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")]
Expand Down Expand Up @@ -217,6 +235,7 @@ pub enum RustfsClientError {
},
TenantSecretLookupFailed,
InvalidPolicyName,
InvalidTierName,
InvalidPolicyDocument,
TenantTlsRequired,
TenantTlsNotReady,
Expand Down Expand Up @@ -257,6 +276,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"),
Expand Down Expand Up @@ -305,6 +325,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 {
Expand All @@ -326,6 +351,7 @@ impl RustfsClientError {
| Self::EmptyCredentialValue { .. }
| Self::InvalidCredentialValue { .. }
| Self::InvalidPolicyName
| Self::InvalidTierName
| Self::InvalidPolicyDocument
| Self::TenantTlsRequired
| Self::TenantTlsClientCertificateRequired
Expand Down Expand Up @@ -559,7 +585,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<BTreeSet<String>, RustfsClientError> {
Ok(parse_tiers(body)?
.into_iter()
.map(|tier| tier.config.name)
.collect())
}
use super::{RustfsClientError, StatusCode};

#[test]
Expand All @@ -579,6 +614,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 [
Expand Down
Loading